fix: AccountNFT with invalid marker (#1589)

Fixes [#1497](https://github.com/XRPLF/clio/issues/1497)
Mimics the behavior of the [fix on Rippled
side](https://github.com/XRPLF/rippled/pull/5045)

.clang-tidy

@@ -142,7 +142,7 @@ CheckOptions:
   readability-braces-around-statements.ShortStatementLines: 2
   bugprone-unsafe-functions.ReportMoreUnsafeFunctions: true
   bugprone-unused-return-value.CheckedReturnTypes: ::std::error_code;::std::error_condition;::std::errc
-  misc-include-cleaner.IgnoreHeaders: '.*/(detail|impl)/.*;.*(expected|unexpected).*'
+  misc-include-cleaner.IgnoreHeaders: '.*/(detail|impl)/.*;.*(expected|unexpected).*;.*ranges_lower_bound\.h;time.h;stdlib.h'
 
 HeaderFilterRegex: '^.*/(src|tests)/.*\.(h|hpp)$'
 WarningsAsErrors: '*'

.clangd

@@ -8,3 +8,6 @@ Diagnostics:
     IgnoreHeader:
       - ".*/(detail|impl)/.*"
       - ".*expected.*"
+      - ".*ranges_lower_bound.h"
+      - "time.h"
+      - "stdlib.h"

.githooks/check-docs

@@ -13,14 +13,36 @@ TMPDIR=${ROOT}/.cache/doxygen
 TMPFILE=${TMPDIR}/docs.log
 DOCDIR=${TMPDIR}/out
 
+# Check doxygen is at all installed
 if [ -z "$DOXYGEN" ]; then
     # No hard error if doxygen is not installed yet
     cat <<EOF
 
                                    WARNING
 -----------------------------------------------------------------------------
-        'doxygen' is required to check documentation.
-        Please install it for next time. For the time being it's on CI.
+        'doxygen' is required to check documentation. 
+        Please install it for next time.
+
+        Your changes may fail to pass CI once pushed.
+-----------------------------------------------------------------------------
+
+EOF
+    exit 0
+fi
+
+# Check version of doxygen is at least 1.12
+version=$($DOXYGEN --version | grep -o '[0-9\.]*')
+
+if [[ "1.12.0" > "$version" ]]; then
+    # No hard error if doxygen version is not the one we want - let CI deal with it
+    cat <<EOF
+
+                                    ERROR
+-----------------------------------------------------------------------------
+        A minimum of version 1.12 of `which doxygen` is required.
+        Your version is $version. Please upgrade it for next time.
+
+        Your changes may fail to pass CI once pushed.
 -----------------------------------------------------------------------------
 
 EOF

.githooks/check-format

@@ -5,6 +5,20 @@
 # This script checks the format of the code and cmake files.
 # In many cases it will automatically fix the issues and abort the commit.
 
+no_formatted_directories_staged() {
+    staged_directories=$(git diff-index --cached --name-only HEAD | awk -F/ '{print $1}')
+    for sd in $staged_directories; do
+        if [[ "$sd" =~ ^(benchmark|cmake|src|tests)$ ]]; then
+            return 1
+        fi
+    done
+    return 0
+}
+
+if no_formatted_directories_staged ; then
+    exit 0
+fi
+
 echo "+ Checking code format..."
 
 # paths to check and re-format

.githooks/pre-commit

@@ -3,6 +3,5 @@
 # This script is intended to be run from the root of the repository.
 
 source .githooks/check-format
-#source .githooks/check-docs
+source .githooks/check-docs
 
-# TODO: Fix Doxygen issue with reference links. See https://github.com/XRPLF/clio/issues/1431

.githooks/pre-push

@@ -1,3 +1,58 @@
 #!/bin/sh
+
+
+# git for-each-ref refs/tags  # see which tags are annotated and which are lightweight. Annotated tags are "tag" objects.
+# # Set these so your commits and tags are always signed
+# git config commit.gpgsign true
+# git config tag.gpgsign true
+
+verify_commit_signed() {
+    if git verify-commit HEAD &> /dev/null; then
+        :
+        # echo "HEAD commit seems signed..."
+    else
+        echo "HEAD commit isn't signed!"
+        exit 1
+    fi
+}
+
+verify_tag() {
+    if git describe --exact-match --tags HEAD &> /dev/null; then
+        : # You might be ok to push
+        # echo "Tag is annotated."
+        return 0
+    else
+        echo "Tag for [$version] not an annotated tag."
+        exit 1
+    fi
+}
+
+verify_tag_signed() {
+    if git verify-tag "$version" &> /dev/null ; then
+        : # ok, I guess we'll let you push
+        # echo "Tag appears signed"
+        return 0
+    else
+        echo "$version tag isn't signed"
+        echo "Sign it with [git tag -ams\"$version\" $version]"
+        exit 1
+    fi
+}
+
+while read local_ref local_oid remote_ref remote_oid; do
+    # Check some things if we're pushing a branch called "release/"
+    if echo "$remote_ref" | grep ^refs\/heads\/release\/ &> /dev/null ; then
+        version=$(echo $remote_ref | awk -F/ '{print $NF}')
+        echo "Looks like you're trying to push a $version release..."
+        echo "Making sure you've signed and tagged it."
+        if verify_commit_signed && verify_tag && verify_tag_signed ; then
+            : # Ok, I guess you can push
+        else
+            exit 1
+        fi
+    fi
+done
+
 command -v git-lfs >/dev/null 2>&1 || { echo >&2 "\nThis repository is configured for Git LFS but 'git-lfs' was not found on your path. If you no longer wish to use Git LFS, remove this hook by deleting the 'pre-push' file in the hooks directory (set by 'core.hookspath'; usually '.git/hooks').\n"; exit 2; }
+
 git lfs pre-push "$@"

.github/actions/build_docker_image/action.yml

@@ -0,0 +1,66 @@
+name: Build and push Docker image
+description: Build and push Docker image to DockerHub and GitHub Container Registry
+inputs:
+  image_name:
+    description: Name of the image to build
+    required: true
+  push_image:
+    description: Whether to push the image to the registry (true/false)
+    required: true
+  directory:
+    description: The directory containing the Dockerfile
+    required: true
+  tags:
+    description: Comma separated tags to apply to the image
+    required: true
+  platforms:
+    description: Platforms to build the image for (e.g. linux/amd64,linux/arm64)
+    required: true
+  description:
+    description: Short description of the image
+    required: true
+runs:
+  using: composite
+  steps:
+      - name: Login to DockerHub
+        if: ${{ inputs.push_image == 'true' }}
+        uses: docker/login-action@v3
+        with:
+          username: ${{ env.DOCKERHUB_USER }}
+          password: ${{ env.DOCKERHUB_PW }}
+
+      - name: Login to GitHub Container Registry
+        if: ${{ inputs.push_image == 'true' }}
+        uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.repository_owner }}
+          password: ${{ env.GITHUB_TOKEN }}
+
+      - uses: docker/setup-qemu-action@v3
+      - uses: docker/setup-buildx-action@v3
+
+      - uses: docker/metadata-action@v5
+        id: meta
+        with:
+          images: ${{ inputs.image_name }}
+          tags: ${{ inputs.tags }}
+
+      - name: Build and push
+        uses: docker/build-push-action@v5
+        with:
+          context: ${{ inputs.directory }}
+          platforms: ${{ inputs.platforms }}
+          push: ${{ inputs.push_image == 'true' }}
+          tags: ${{ steps.meta.outputs.tags }}
+
+      - name: Update DockerHub description
+        if: ${{ inputs.push_image == 'true' }}
+        uses: peter-evans/dockerhub-description@v4
+        with:
+          username: ${{ env.DOCKERHUB_USER }}
+          password: ${{ env.DOCKERHUB_PW }}
+          repository: ${{ inputs.image_name }}
+          short-description: ${{ inputs.description }}
+          readme-filepath: ${{ inputs.directory }}/README.md
+

.github/dependabot.yml

@@ -0,0 +1,16 @@
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"
+      day: "monday"
+      time: "04:00"
+      timezone: "Etc/GMT"
+    reviewers:
+      - "cindyyan317"
+      - "godexsoft"
+      - "kuznetsss"
+    commit-message:
+      prefix: "[CI] "
+    target-branch: "develop"

.github/workflows/build.yml

@@ -23,7 +23,7 @@ jobs:
         run: |
           ./.githooks/check-format --diff
         shell: bash
-  
+ 
   check_docs:
     name: Check documentation
     runs-on: ubuntu-20.04
@@ -149,6 +149,13 @@ jobs:
           name: clio_tests_${{ runner.os }}_${{ matrix.build_type }}_${{ steps.conan.outputs.conan_profile }}
           path: build/clio_*tests
 
+      - name: Upload test data
+        if: ${{ !matrix.code_coverage }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: clio_test_data_${{ runner.os }}_${{ matrix.build_type }}_${{ steps.conan.outputs.conan_profile }}
+          path: build/tests/unit/test_data
+
       - name: Save cache
         uses: ./.github/actions/save_cache
         with:
@@ -211,6 +218,12 @@ jobs:
       - uses: actions/download-artifact@v4
         with:
           name: clio_tests_${{ runner.os }}_${{ matrix.build_type }}_${{ matrix.conan_profile }}
+
+      - uses: actions/download-artifact@v4
+        with:
+          name: clio_test_data_${{ runner.os }}_${{ matrix.build_type }}_${{ matrix.conan_profile }}
+          path: tests/unit/test_data
+
       - name: Run clio_tests
         run: |
           chmod +x ./clio_tests

.github/workflows/build_clio_docker_image.yml

@@ -0,0 +1,95 @@
+name: Build and publish Clio docker image
+on:
+  workflow_call:
+    inputs:
+      tags:
+        required: true
+        type: string
+        description: Comma separated tags for docker image
+      artifact_name:
+        type: string
+        description: Name of Github artifact to put into docker image
+      strip_binary:
+        type: boolean
+        description: Whether to strip clio binary
+        default: true
+      publish_image:
+        type: boolean
+        description: Whether to publish docker image
+        required: true
+
+  workflow_dispatch:
+    inputs:
+      tags:
+        required: true
+        type: string
+        description: Comma separated tags for docker image
+      clio_server_binary_url:
+        required: true
+        type: string
+        description: Url to download clio_server binary from
+      binary_sha256:
+        required: true
+        type: string
+        description: sha256 hash of the binary
+      strip_binary:
+        type: boolean
+        description: Whether to strip clio binary
+        default: true
+
+jobs:
+  build_and_publish_image:
+    name: Build and publish image
+    runs-on: ubuntu-20.04
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Download Clio binary from artifact
+        if: ${{ inputs.artifact_name != null }}
+        uses: actions/download-artifact@v4
+        with:
+          name: ${{ inputs.artifact_name }}
+          path: ./docker/clio/artifact/
+
+      - name: Download Clio binary from url
+        if: ${{ inputs.clio_server_binary_url != null }}
+        shell: bash
+        run: |
+          wget ${{inputs.clio_server_binary_url}} -P ./docker/clio/artifact/
+          if [ "$(sha256sum ./docker/clio/clio_server | awk '{print $1}')" != "${{inputs.binary_sha256}}" ]; then
+            echo "Binary sha256 sum doesn't match"
+            exit 1
+          fi
+      - name: Unpack binary
+        shell: bash
+        run: |
+          sudo apt update && sudo apt install -y tar unzip
+          cd docker/clio/artifact
+          artifact=$(find . -type f)
+          if [[ $artifact == *.zip ]]; then
+            unzip $artifact
+          elif [[ $artifact == *.tar.gz ]]; then
+            tar -xvf $artifact
+          fi
+          mv clio_server ../
+          cd ../
+          rm -rf ./artifact
+
+      - name: Strip binary
+        if: ${{ inputs.strip_binary }}
+        shell: bash
+        run: strip ./docker/clio/clio_server
+
+      - name: Build Docker image
+        uses: ./.github/actions/build_docker_image
+        env:
+          DOCKERHUB_USER: ${{ secrets.DOCKERHUB_USER }}
+          DOCKERHUB_PW: ${{ secrets.DOCKERHUB_PW }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          image_name: rippleci/clio
+          push_image: ${{ inputs.publish_image }}
+          directory: docker/clio
+          tags: ${{ inputs.tags }}
+          platforms: linux/amd64
+          description: Clio is an XRP Ledger API server.

.github/workflows/check_libxrpl.yml

@@ -47,7 +47,7 @@ jobs:
       - name: Upload clio_tests
         uses: actions/upload-artifact@v4
         with:
-          name: clio_tests_libxrpl-${{ github.event.client_payload.version }}
+          name: clio_tests_check_libxrpl
           path: build/clio_tests
 
   run_tests:
@@ -60,7 +60,7 @@ jobs:
     steps:
       - uses: actions/download-artifact@v4
         with:
-          name: clio_tests_libxrpl-${{ github.event.client_payload.version }}
+          name: clio_tests_check_libxrpl
 
       - name: Run clio_tests
         run: |

.github/workflows/check_pr_title.yml

@@ -0,0 +1,18 @@
+name: Check PR title
+on:
+  pull_request:
+    types: [opened, edited, reopened, synchronize]
+    branches: [develop]
+
+jobs:
+  check_title:
+    runs-on: ubuntu-20.04
+    # permissions:
+    #   pull-requests: write
+    steps:
+      - uses: ytanikin/PRConventionalCommits@1.2.0
+        with:
+          task_types: '["build","feat","fix","docs","test","ci","style","refactor","perf","chore"]'
+          add_label: false
+          # Turned off labelling because it leads to an error, see https://github.com/ytanikin/PRConventionalCommits/issues/19
+          # custom_labels: '{"build":"build", "feat":"enhancement", "fix":"bug", "docs":"documentation", "test":"testability", "ci":"ci", "style":"refactoring", "refactor":"refactoring", "perf":"performance", "chore":"tooling"}'

.github/workflows/clang-tidy.yml

@@ -89,7 +89,7 @@ jobs:
 
             List of the issues found: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}/
 
-      - uses: crazy-max/ghaction-import-gpg@v5
+      - uses: crazy-max/ghaction-import-gpg@v6
         if: ${{ steps.run_clang_tidy.outcome != 'success' }}
         with:
           gpg_private_key: ${{ secrets.ACTIONS_GPG_PRIVATE_KEY }}
@@ -99,7 +99,7 @@ jobs:
 
       - name: Create PR with fixes
         if: ${{ steps.run_clang_tidy.outcome != 'success' }}
-        uses: peter-evans/create-pull-request@v5
+        uses: peter-evans/create-pull-request@v6
         env:
           GH_REPO: ${{ github.repository }}
           GH_TOKEN: ${{ github.token }}
@@ -109,7 +109,7 @@ jobs:
           branch: "clang_tidy/autofix"
           branch-suffix: timestamp
           delete-branch: true
-          title: "[CI] clang-tidy auto fixes"
+          title: "style: clang-tidy auto fixes"
           body: "Fixes #${{ steps.create_issue.outputs.created_issue_id }}. Please review and commit clang-tidy fixes."
           reviewers: "cindyyan317,godexsoft,kuznetsss"
 

.github/workflows/clang-tidy_on_fix_merged.yml

@@ -17,7 +17,7 @@ jobs:
         id: check
         shell: bash
         run: |
-          passed=$(if [[ $(git log -1 --pretty=format:%s | grep '\[CI\] clang-tidy auto fixes') ]]; then echo 'true' ; else echo 'false' ; fi)
+          passed=$(if [[ $(git log -1 --pretty=format:%s | grep 'style: clang-tidy auto fixes') ]]; then echo 'true' ; else echo 'false' ; fi)
           echo "passed=$passed" >> $GITHUB_OUTPUT
 
       - name: Run clang-tidy workflow

.github/workflows/docs.yml

@@ -34,7 +34,7 @@ jobs:
           cmake ../docs && cmake --build . --target docs
       
       - name: Setup Pages
-        uses: actions/configure-pages@v4
+        uses: actions/configure-pages@v5
 
       - name: Upload artifact
         uses: actions/upload-pages-artifact@v3

.github/workflows/nightly.yml

@@ -3,6 +3,10 @@ on:
   schedule:
     - cron:  '0 5 * * 1-5'
   workflow_dispatch:
+  pull_request:
+    paths:
+      - '.github/workflows/nightly.yml'
+      - '.github/workflows/build_clio_docker_image.yml'
 
 jobs:
   build:
@@ -13,12 +17,15 @@ jobs:
         include:
           - os: macos14
             build_type: Release
+            static: false
           - os: heavy
             build_type: Release
+            static: true
             container:
               image: rippleci/clio_ci:latest
           - os: heavy
             build_type: Debug
+            static: true
             container:
               image: rippleci/clio_ci:latest
     runs-on: [self-hosted, "${{ matrix.os }}"]
@@ -50,6 +57,7 @@ jobs:
           conan_profile: ${{ steps.conan.outputs.conan_profile }}
           conan_cache_hit: ${{ steps.restore_cache.outputs.conan_cache_hit }}
           build_type: ${{ matrix.build_type }}
+          static: ${{ matrix.static }}
 
       - name: Build Clio
         uses: ./.github/actions/build_clio
@@ -63,6 +71,12 @@ jobs:
           name: clio_tests_${{ runner.os }}_${{ matrix.build_type }}
           path: build/clio_*tests
 
+      - name: Upload test data
+        uses: actions/upload-artifact@v4
+        with:
+          name: clio_test_data_${{ runner.os }}_${{ matrix.build_type }}
+          path: build/tests/unit/test_data
+
       - name: Compress clio_server
         shell: bash
         run: |
@@ -116,6 +130,11 @@ jobs:
         with:
           name: clio_tests_${{ runner.os }}_${{ matrix.build_type }}
 
+      - uses: actions/download-artifact@v4
+        with:
+          name: clio_test_data_${{ runner.os }}_${{ matrix.build_type }}
+          path: tests/unit/test_data
+
       - name: Run clio_tests
         run: |
           chmod +x ./clio_tests
@@ -130,6 +149,7 @@ jobs:
           ./clio_integration_tests --backend_host=scylladb
 
   nightly_release:
+    if: ${{ github.event_name != 'pull_request' }}
     needs: run_tests
     runs-on: ubuntu-20.04
     env:
@@ -143,13 +163,13 @@ jobs:
       - uses: actions/download-artifact@v4
         with:
           path: nightly_release
+          pattern: clio_server_*
 
       - name: Prepare files
         shell: bash
         run: |
           cp ${{ github.workspace }}/.github/workflows/nightly_notes.md "${RUNNER_TEMP}/nightly_notes.md"
           cd nightly_release
-          rm -r clio_*tests*
           for d in $(ls); do
             archive_name=$(ls $d)
             mv ${d}/${archive_name} ./
@@ -172,9 +192,21 @@ jobs:
             --target $GITHUB_SHA --notes-file "${RUNNER_TEMP}/nightly_notes.md" \
             ./nightly_release/clio_server*
 
+  build_and_publish_docker_image:
+    uses: ./.github/workflows/build_clio_docker_image.yml
+    needs: run_tests
+    secrets: inherit
+    with:
+      tags: |
+          type=raw,value=nightly
+          type=raw,value=${{ github.sha }}
+      artifact_name: clio_server_Linux_Release
+      strip_binary: true
+      publish_image: ${{ github.event_name != 'pull_request' }}
+
   create_issue_on_failure:
-    needs: [build, run_tests, nightly_release]
-    if: ${{ always() && contains(needs.*.result, 'failure') }}
+    needs: [build, run_tests, nightly_release, build_and_publish_docker_image]
+    if: ${{ always() && contains(needs.*.result, 'failure') && github.event_name != 'pull_request' }}
     runs-on: ubuntu-20.04
     permissions:
       contents: write

.github/workflows/update_docker_ci.yml

@@ -18,30 +18,19 @@ jobs:
     name: Build and push docker image
     runs-on: [self-hosted, heavy]
     steps:
-      - name: Login to DockerHub
-        if: ${{ github.event_name != 'pull_request' }}
-        uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKERHUB_USER }}
-          password: ${{ secrets.DOCKERHUB_PW }}
-
       - uses: actions/checkout@v4
-      - uses: docker/setup-qemu-action@v3
-      - uses: docker/setup-buildx-action@v3
-      - uses: docker/metadata-action@v5
-        id: meta
+      - uses: ./.github/actions/build_docker_image
+        env:
+          DOCKERHUB_USER: ${{ secrets.DOCKERHUB_USER }}
+          DOCKERHUB_PW: ${{ secrets.DOCKERHUB_PW }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:
-          images: rippleci/clio_ci
+          image_name: rippleci/clio_ci
+          push_image: ${{ github.event_name != 'pull_request' }}
+          directory: docker/ci
           tags: |
             type=raw,value=latest
             type=raw,value=gcc_12_clang_16
-            type=raw,value=${{ env.GITHUB_SHA }}
-
-      - name: Build and push
-        uses: docker/build-push-action@v5
-        with:
-          context: ${{ github.workspace }}/docker/ci
+            type=raw,value=${{ github.sha }}
           platforms: linux/amd64,linux/arm64
-          push: ${{ github.event_name != 'pull_request' }}
-          tags: ${{ steps.meta.outputs.tags }}
-
+          description: CI image for XRPLF/clio.

.github/workflows/upload_coverage_report.yml

@@ -23,7 +23,7 @@ jobs:
 
       - name: Upload coverage report
         if: ${{ hashFiles('build/coverage_report.xml') != '' }}
-        uses: wandalen/wretry.action@v1.4.10
+        uses: wandalen/wretry.action@v3.5.0
         with:
           action: codecov/codecov-action@v4
           with: |

.gitignore

@@ -8,4 +8,4 @@
 .DS_Store
 CMakeUserPresets.json
 config.json
-src/main/impl/Build.cpp
+src/util/build/Build.cpp

CONTRIBUTING.md

@@ -72,6 +72,9 @@ git push --force
 Clio uses `ccache` to speed up compilation. If you want to use it, please make sure it is installed on your machine.
 CMake will automatically detect it and use it if it is available.
 
+## Opening a pull request
+When a pull request is open CI will perform checks on the new code.
+Title of the pull request and squashed commit should follow [conventional commits specification](https://www.conventionalcommits.org/en/v1.0.0/).
 
 ## Fixing issues found during code review
 While your code is in review, it's possible that some changes will be requested by reviewer(s).

benchmarks/CMakeLists.txt

@@ -12,5 +12,5 @@ target_sources(
 include(deps/gbench)
 
 target_include_directories(clio_benchmark PRIVATE .)
-target_link_libraries(clio_benchmark PUBLIC clio benchmark::benchmark_main)
+target_link_libraries(clio_benchmark PUBLIC clio_etl benchmark::benchmark_main)
 set_target_properties(clio_benchmark PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})

benchmarks/util/async/ExecutionContextBenchmarks.cpp

@@ -31,7 +31,6 @@
 #include <cstdint>
 #include <latch>
 #include <optional>
-#include <stdexcept>
 #include <thread>
 #include <vector>
 

cmake/Build.cpp.in

@@ -17,11 +17,12 @@
 */
 //==============================================================================
 
-#include "main/Build.hpp"
+#include "util/build/Build.hpp"
 
 #include <string>
 
-namespace Build {
+namespace util::build {
+
 static constexpr char versionString[] = "@CLIO_VERSION@";
 
 std::string const&
@@ -38,4 +39,4 @@ getClioFullVersionString()
     return value;
 }
 
-}  // namespace Build
+}  // namespace util::build

cmake/ClioVersion.cmake

@@ -45,4 +45,4 @@ endif ()
 
 message(STATUS "Build version: ${CLIO_VERSION}")
 
-configure_file(${CMAKE_CURRENT_LIST_DIR}/Build.cpp.in ${CMAKE_CURRENT_LIST_DIR}/../src/main/impl/Build.cpp)
+configure_file(${CMAKE_CURRENT_LIST_DIR}/Build.cpp.in ${CMAKE_CURRENT_LIST_DIR}/../src/util/build/Build.cpp)

conanfile.py

@@ -28,7 +28,7 @@ class Clio(ConanFile):
         'protobuf/3.21.9',
         'grpc/1.50.1',
         'openssl/1.1.1u',
-        'xrpl/2.2.0',
+        'xrpl/2.3.0-b1',
         'libbacktrace/cci.20210118'
     ]
 

docker/ci/README.md

@@ -0,0 +1,15 @@
+# CI image for XRPLF/clio
+
+This image contains an environment to build [Clio](https://github.com/XRPLF/clio), check code and documentation.
+It is used in [Clio Github Actions](https://github.com/XRPLF/clio/actions) but can also be used to compile Clio locally.
+
+The image is based on Ubuntu 20.04 and contains:
+- clang 16
+- gcc 12.3
+- doxygen 1.10
+- gh 2.40
+- ccache 4.8.3
+- conan
+- and some other useful tools
+
+Conan is set up to build Clio without any additional steps. There are two preset conan profiles: `clang` and `gcc` to use corresponding compiler.

docker/ci/dockerfile

@@ -6,10 +6,10 @@ SHELL ["/bin/bash", "-c"]
 USER root
 WORKDIR /root
 
-ENV CCACHE_VERSION=4.8.3 \
+ENV CCACHE_VERSION=4.10.2 \
     LLVM_TOOLS_VERSION=18 \
     GH_VERSION=2.40.0 \
-    DOXYGEN_VERSION=1.10.0
+    DOXYGEN_VERSION=1.12.0
  
 # Add repositories
 RUN apt-get -qq update \

docker/clio/README.md

@@ -0,0 +1,23 @@
+# Clio official docker image
+
+[Clio](https://github.com/XRPLF/clio) is an XRP Ledger API server optimized for RPC calls over WebSocket or JSON-RPC.
+It stores validated historical ledger and transaction data in a space efficient format.
+
+This image contains `clio_server` binary allowing users to run Clio easily.
+
+## Clio configuration file
+
+Please note that while Clio requires a configuration file, this image doesn't include any default config.
+Your configuration file should be mounted under the path `/opt/clio/etc/config.json`.
+Clio repository provides an [example](https://github.com/XRPLF/clio/blob/develop/docs/examples/config/example-config.json) of the configuration file.
+
+Config file recommendations:
+- Set `log_to_console` to `false` if you want to avoid logs being written to `stdout`.
+- Set `log_directory` to `/opt/clio/log` to store logs in a volume.
+
+## Usage
+
+The following command can be used to run Clio in docker (assuming server's port is `51233` in your config):
+```bash
+docker run -d -v <path to your config.json>:/opt/clio/etc/config.json -v <path to store logs>:/opt/clio/log -p 51233:51233 rippleci/clio
+```

docker/clio/dockerfile

@@ -0,0 +1,16 @@
+FROM ubuntu:22.04
+
+COPY ./clio_server /opt/clio/bin/clio_server
+
+RUN ln -s /opt/clio/bin/clio_server /usr/local/bin/clio_server && \
+    mkdir -p /opt/clio/etc/ && \
+    mkdir -p /opt/clio/log/ && \
+    groupadd -g 10001 clio && \
+    useradd -u 10000 -g 10001 -s /bin/bash clio && \
+    chown clio:clio /opt/clio/log && \
+    apt update && \
+    apt install -y libatomic1
+
+USER clio
+ENTRYPOINT ["/opt/clio/bin/clio_server"]
+CMD ["--conf", "/opt/clio/etc/config.json"]

docker/develop/compose.yaml

@@ -1,4 +1,3 @@
-version: '3.7'
 services:
   clio_develop:
     image: rippleci/clio_ci:latest

docs/build-clio.md

@@ -141,3 +141,43 @@ If you wish to develop against a `rippled` instance running in standalone mode t
 
 1. Advance the `rippled` ledger to at least ledger 256.
 2. Wait 10 minutes before first starting Clio against this standalone node.
+
+## Building with a Custom `libxrpl`
+
+Sometimes, during development, you need to build against a custom version of `libxrpl`. (For example, you may be developing compatibility for a proposed amendment that is not yet merged to the main `rippled` codebase.) To build Clio with compatibility for a custom fork or branch of `rippled`, follow these steps:
+
+1. First, pull/clone the appropriate `rippled` fork and switch to the branch you want to build. For example, the following example uses an in-development build with [XLS-33d Multi-Purpose Tokens](https://github.com/XRPLF/XRPL-Standards/tree/master/XLS-0033d-multi-purpose-tokens):
+
+    ```sh
+    git clone https://github.com/shawnxie999/rippled/
+    cd rippled
+    git switch mpt-1.1
+    ```
+
+2. Export a custom package to your local Conan store using a user/channel:
+
+    ```sh
+    conan export . my/feature
+    ```
+
+3. Patch your local Clio build to use the right package.
+
+    Edit `conanfile.py` (from the Clio repository root). Replace the `xrpl` requirement with the custom package version from the previous step. This must also include the current version number from your `rippled` branch. For example:
+
+    ```py
+    # ... (excerpt from conanfile.py)
+        requires = [
+        'boost/1.82.0',
+        'cassandra-cpp-driver/2.17.0',
+        'fmt/10.1.1',
+        'protobuf/3.21.9',
+        'grpc/1.50.1',
+        'openssl/1.1.1u',
+        'xrpl/2.3.0-b1@my/feature', # Update this line
+        'libbacktrace/cci.20210118'
+    ]
+    ```
+
+4. Build Clio as you would have before.
+
+    See [Building Clio](#building-clio) for details.

docs/examples/config/example-config.json

@@ -31,7 +31,7 @@
     "etl_sources": [
         {
             "ip": "127.0.0.1",
-            "ws_port": "6006",
+            "ws_port": "6005",
             "grpc_port": "50051"
         }
     ],

src/CMakeLists.txt

@@ -4,4 +4,5 @@ add_subdirectory(etl)
 add_subdirectory(feed)
 add_subdirectory(rpc)
 add_subdirectory(web)
+add_subdirectory(app)
 add_subdirectory(main)

src/app/CMakeLists.txt

@@ -0,0 +1,4 @@
+add_library(clio_app)
+target_sources(clio_app PRIVATE CliArgs.cpp ClioApplication.cpp)
+
+target_link_libraries(clio_app PUBLIC clio_etl clio_feed clio_web clio_rpc)

src/app/CliArgs.cpp

@@ -0,0 +1,70 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include "app/CliArgs.hpp"
+
+#include "util/build/Build.hpp"
+
+#include <boost/program_options/options_description.hpp>
+#include <boost/program_options/parsers.hpp>
+#include <boost/program_options/positional_options.hpp>
+#include <boost/program_options/value_semantic.hpp>
+#include <boost/program_options/variables_map.hpp>
+
+#include <cstdlib>
+#include <iostream>
+#include <string>
+#include <utility>
+
+namespace app {
+
+CliArgs::Action
+CliArgs::parse(int argc, char const* argv[])
+{
+    namespace po = boost::program_options;
+    // clang-format off
+    po::options_description description("Options");
+    description.add_options()
+        ("help,h", "print help message and exit")
+        ("version,v", "print version and exit")
+        ("conf,c", po::value<std::string>()->default_value(defaultConfigPath), "configuration file")
+    ;
+    // clang-format on
+    po::positional_options_description positional;
+    positional.add("conf", 1);
+
+    po::variables_map parsed;
+    po::store(po::command_line_parser(argc, argv).options(description).positional(positional).run(), parsed);
+    po::notify(parsed);
+
+    if (parsed.count("help") != 0u) {
+        std::cout << "Clio server " << util::build::getClioFullVersionString() << "\n\n" << description;
+        return Action{Action::Exit{EXIT_SUCCESS}};
+    }
+
+    if (parsed.count("version") != 0u) {
+        std::cout << util::build::getClioFullVersionString() << '\n';
+        return Action{Action::Exit{EXIT_SUCCESS}};
+    }
+
+    auto configPath = parsed["conf"].as<std::string>();
+    return Action{Action::Run{std::move(configPath)}};
+}
+
+}  // namespace app

src/app/CliArgs.hpp

@@ -0,0 +1,96 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#pragma once
+
+#include "util/OverloadSet.hpp"
+
+#include <string>
+#include <variant>
+
+namespace app {
+
+/**
+ * @brief Parsed command line arguments representation.
+ */
+class CliArgs {
+public:
+    /**
+     * @brief Default configuration path.
+     */
+    static constexpr char defaultConfigPath[] = "/etc/opt/clio/config.json";
+
+    /**
+     * @brief An action parsed from the command line.
+     */
+    class Action {
+    public:
+        /** @brief Run action. */
+        struct Run {
+            /** @brief Configuration file path. */
+            std::string configPath;
+        };
+
+        /** @brief Exit action. */
+        struct Exit {
+            /** @brief Exit code. */
+            int exitCode;
+        };
+
+        /**
+         * @brief Construct an action from a Run.
+         *
+         * @param action Run action.
+         */
+        template <typename ActionType>
+            requires std::is_same_v<ActionType, Run> or std::is_same_v<ActionType, Exit>
+        explicit Action(ActionType&& action) : action_(std::forward<ActionType>(action))
+        {
+        }
+
+        /**
+         * @brief Apply a function to the action.
+         *
+         * @tparam Processors Action processors types. Must be callable with the action type and return int.
+         * @param processors Action processors.
+         * @return Exit code.
+         */
+        template <typename... Processors>
+        int
+        apply(Processors&&... processors) const
+        {
+            return std::visit(util::OverloadSet{std::forward<Processors>(processors)...}, action_);
+        }
+
+    private:
+        std::variant<Run, Exit> action_;
+    };
+
+    /**
+     * @brief Parse command line arguments.
+     *
+     * @param argc Number of arguments.
+     * @param argv Array of arguments.
+     * @return Parsed command line arguments.
+     */
+    static Action
+    parse(int argc, char const* argv[]);
+};
+
+}  // namespace app

src/app/ClioApplication.cpp

@@ -0,0 +1,142 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include "app/ClioApplication.hpp"
+
+#include "data/AmendmentCenter.hpp"
+#include "data/BackendFactory.hpp"
+#include "etl/ETLService.hpp"
+#include "etl/LoadBalancer.hpp"
+#include "etl/NetworkValidatedLedgers.hpp"
+#include "feed/SubscriptionManager.hpp"
+#include "rpc/Counters.hpp"
+#include "rpc/RPCEngine.hpp"
+#include "rpc/WorkQueue.hpp"
+#include "rpc/common/impl/HandlerProvider.hpp"
+#include "util/build/Build.hpp"
+#include "util/config/Config.hpp"
+#include "util/log/Logger.hpp"
+#include "util/prometheus/Prometheus.hpp"
+#include "web/RPCServerHandler.hpp"
+#include "web/Server.hpp"
+#include "web/dosguard/DOSGuard.hpp"
+#include "web/dosguard/IntervalSweepHandler.hpp"
+#include "web/dosguard/WhitelistHandler.hpp"
+
+#include <boost/asio/io_context.hpp>
+
+#include <cstdint>
+#include <cstdlib>
+#include <memory>
+#include <thread>
+#include <vector>
+
+namespace app {
+
+namespace {
+
+/**
+ * @brief Start context threads
+ *
+ * @param ioc Context
+ * @param numThreads Number of worker threads to start
+ */
+void
+start(boost::asio::io_context& ioc, std::uint32_t numThreads)
+{
+    std::vector<std::thread> v;
+    v.reserve(numThreads - 1);
+    for (auto i = numThreads - 1; i > 0; --i)
+        v.emplace_back([&ioc] { ioc.run(); });
+
+    ioc.run();
+    for (auto& t : v)
+        t.join();
+}
+
+}  // namespace
+
+ClioApplication::ClioApplication(util::Config const& config) : config_(config), signalsHandler_{config_}
+{
+    LOG(util::LogService::info()) << "Clio version: " << util::build::getClioFullVersionString();
+    PrometheusService::init(config);
+}
+
+int
+ClioApplication::run()
+{
+    auto const threads = config_.valueOr("io_threads", 2);
+    if (threads <= 0) {
+        LOG(util::LogService::fatal()) << "io_threads is less than 1";
+        return EXIT_FAILURE;
+    }
+    LOG(util::LogService::info()) << "Number of io threads = " << threads;
+
+    // IO context to handle all incoming requests, as well as other things.
+    // This is not the only io context in the application.
+    boost::asio::io_context ioc{threads};
+
+    // Rate limiter, to prevent abuse
+    auto whitelistHandler = web::dosguard::WhitelistHandler{config_};
+    auto dosGuard = web::dosguard::DOSGuard{config_, whitelistHandler};
+    auto sweepHandler = web::dosguard::IntervalSweepHandler{config_, ioc, dosGuard};
+
+    // Interface to the database
+    auto backend = data::make_Backend(config_);
+
+    // Manages clients subscribed to streams
+    auto subscriptionsRunner = feed::SubscriptionManagerRunner(config_, backend);
+
+    auto const subscriptions = subscriptionsRunner.getManager();
+
+    // Tracks which ledgers have been validated by the network
+    auto ledgers = etl::NetworkValidatedLedgers::make_ValidatedLedgers();
+
+    // Handles the connection to one or more rippled nodes.
+    // ETL uses the balancer to extract data.
+    // The server uses the balancer to forward RPCs to a rippled node.
+    // The balancer itself publishes to streams (transactions_proposed and accounts_proposed)
+    auto balancer = etl::LoadBalancer::make_LoadBalancer(config_, ioc, backend, subscriptions, ledgers);
+
+    // ETL is responsible for writing and publishing to streams. In read-only mode, ETL only publishes
+    auto etl = etl::ETLService::make_ETLService(config_, ioc, backend, subscriptions, balancer, ledgers);
+
+    auto workQueue = rpc::WorkQueue::make_WorkQueue(config_);
+    auto counters = rpc::Counters::make_Counters(workQueue);
+    auto const amendmentCenter = std::make_shared<data::AmendmentCenter const>(backend);
+    auto const handlerProvider = std::make_shared<rpc::impl::ProductionHandlerProvider const>(
+        config_, backend, subscriptions, balancer, etl, amendmentCenter, counters
+    );
+    auto const rpcEngine =
+        rpc::RPCEngine::make_RPCEngine(backend, balancer, dosGuard, workQueue, counters, handlerProvider);
+
+    // Init the web server
+    auto handler =
+        std::make_shared<web::RPCServerHandler<rpc::RPCEngine, etl::ETLService>>(config_, backend, rpcEngine, etl);
+    auto const httpServer = web::make_HttpServer(config_, ioc, dosGuard, handler);
+
+    // Blocks until stopped.
+    // When stopped, shared_ptrs fall out of scope
+    // Calls destructors on all resources, and destructs in order
+    start(ioc, threads);
+
+    return EXIT_SUCCESS;
+}
+
+}  // namespace app

src/app/ClioApplication.hpp

@@ -0,0 +1,51 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#pragma once
+
+#include "util/SignalsHandler.hpp"
+#include "util/config//Config.hpp"
+
+namespace app {
+
+/**
+ * @brief The main application class
+ */
+class ClioApplication {
+    util::Config const& config_;
+    util::SignalsHandler signalsHandler_;
+
+public:
+    /**
+     * @brief Construct a new ClioApplication object
+     *
+     * @param config The configuration of the application
+     */
+    ClioApplication(util::Config const& config);
+
+    /**
+     * @brief Run the application
+     *
+     * @return exit code
+     */
+    int
+    run();
+};
+
+}  // namespace app

src/data/AmendmentCenter.cpp

@@ -0,0 +1,203 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include "data/AmendmentCenter.hpp"
+
+#include "data/BackendInterface.hpp"
+#include "data/Types.hpp"
+#include "util/Assert.hpp"
+
+#include <boost/asio/spawn.hpp>
+#include <xrpl/basics/Slice.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/Feature.h>
+#include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STLedgerEntry.h>
+#include <xrpl/protocol/Serializer.h>
+#include <xrpl/protocol/digest.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <iterator>
+#include <map>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <stdexcept>
+#include <string>
+#include <string_view>
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
+namespace {
+
+std::unordered_set<std::string>&
+SUPPORTED_AMENDMENTS()
+{
+    static std::unordered_set<std::string> amendments = {};
+    return amendments;
+}
+
+bool
+lookupAmendment(auto const& allAmendments, std::vector<ripple::uint256> const& ledgerAmendments, std::string_view name)
+{
+    namespace rg = std::ranges;
+    if (auto const am = rg::find(allAmendments, name, &data::Amendment::name); am != rg::end(allAmendments))
+        return rg::find(ledgerAmendments, am->feature) != rg::end(ledgerAmendments);
+    return false;
+}
+
+}  // namespace
+
+namespace data {
+namespace impl {
+
+WritingAmendmentKey::WritingAmendmentKey(std::string amendmentName) : AmendmentKey{std::move(amendmentName)}
+{
+    ASSERT(not SUPPORTED_AMENDMENTS().contains(name), "Attempt to register the same amendment twice");
+    SUPPORTED_AMENDMENTS().insert(name);
+}
+
+}  // namespace impl
+
+AmendmentKey::operator std::string const&() const
+{
+    return name;
+}
+
+AmendmentKey::operator std::string_view() const
+{
+    return name;
+}
+
+AmendmentKey::operator ripple::uint256() const
+{
+    return Amendment::GetAmendmentId(name);
+}
+
+AmendmentCenter::AmendmentCenter(std::shared_ptr<data::BackendInterface> const& backend) : backend_{backend}
+{
+    namespace rg = std::ranges;
+    namespace vs = std::views;
+
+    rg::copy(
+        ripple::allAmendments() | vs::transform([&](auto const& p) {
+            auto const& [name, support] = p;
+            return Amendment{
+                .name = name,
+                .feature = Amendment::GetAmendmentId(name),
+                .isSupportedByXRPL = support != ripple::AmendmentSupport::Unsupported,
+                .isSupportedByClio = rg::find(SUPPORTED_AMENDMENTS(), name) != rg::end(SUPPORTED_AMENDMENTS()),
+                .isRetired = support == ripple::AmendmentSupport::Retired
+            };
+        }),
+        std::back_inserter(all_)
+    );
+
+    for (auto const& am : all_ | vs::filter([](auto const& am) { return am.isSupportedByClio; }))
+        supported_.insert_or_assign(am.name, am);
+}
+
+bool
+AmendmentCenter::isSupported(AmendmentKey const& key) const
+{
+    return supported_.contains(key);
+}
+
+std::map<std::string, Amendment> const&
+AmendmentCenter::getSupported() const
+{
+    return supported_;
+}
+
+std::vector<Amendment> const&
+AmendmentCenter::getAll() const
+{
+    return all_;
+}
+
+bool
+AmendmentCenter::isEnabled(AmendmentKey const& key, uint32_t seq) const
+{
+    return data::synchronous([this, &key, seq](auto yield) { return isEnabled(yield, key, seq); });
+}
+
+bool
+AmendmentCenter::isEnabled(boost::asio::yield_context yield, AmendmentKey const& key, uint32_t seq) const
+{
+    if (auto const listAmendments = fetchAmendmentsList(yield, seq); listAmendments)
+        return lookupAmendment(all_, *listAmendments, key);
+
+    return false;
+}
+
+std::vector<bool>
+AmendmentCenter::isEnabled(boost::asio::yield_context yield, std::vector<AmendmentKey> const& keys, uint32_t seq) const
+{
+    namespace rg = std::ranges;
+
+    if (auto const listAmendments = fetchAmendmentsList(yield, seq); listAmendments) {
+        std::vector<bool> out;
+        rg::transform(keys, std::back_inserter(out), [this, &listAmendments](auto const& key) {
+            return lookupAmendment(all_, *listAmendments, key);
+        });
+
+        return out;
+    }
+
+    return std::vector<bool>(keys.size(), false);
+}
+
+Amendment const&
+AmendmentCenter::getAmendment(AmendmentKey const& key) const
+{
+    ASSERT(supported_.contains(key), "The amendment '{}' must be present in supported amendments list", key.name);
+    return supported_.at(key);
+}
+
+Amendment const&
+AmendmentCenter::operator[](AmendmentKey const& key) const
+{
+    return getAmendment(key);
+}
+
+ripple::uint256
+Amendment::GetAmendmentId(std::string_view name)
+{
+    return ripple::sha512Half(ripple::Slice(name.data(), name.size()));
+}
+
+std::optional<std::vector<ripple::uint256>>
+AmendmentCenter::fetchAmendmentsList(boost::asio::yield_context yield, uint32_t seq) const
+{
+    // the amendments should always be present on the ledger
+    auto const amendments = backend_->fetchLedgerObject(ripple::keylet::amendments().key, seq, yield);
+    if (not amendments.has_value())
+        throw std::runtime_error("Amendments ledger object must be present in the database");
+
+    ripple::SLE const amendmentsSLE{
+        ripple::SerialIter{amendments->data(), amendments->size()}, ripple::keylet::amendments().key
+    };
+
+    return amendmentsSLE[~ripple::sfAmendments];
+}
+
+}  // namespace data

src/data/AmendmentCenter.hpp

@@ -0,0 +1,252 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#pragma once
+
+#include "data/AmendmentCenterInterface.hpp"
+#include "data/BackendInterface.hpp"
+#include "data/Types.hpp"
+
+#include <boost/asio/spawn.hpp>
+#include <boost/preprocessor.hpp>
+#include <boost/preprocessor/seq/for_each.hpp>
+#include <boost/preprocessor/stringize.hpp>
+#include <boost/preprocessor/variadic/to_seq.hpp>
+#include <xrpl/basics/Slice.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/Feature.h>
+#include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STLedgerEntry.h>
+#include <xrpl/protocol/Serializer.h>
+#include <xrpl/protocol/digest.h>
+
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#define REGISTER(name)                                   \
+    inline static impl::WritingAmendmentKey const name = \
+        impl::WritingAmendmentKey(std::string(BOOST_PP_STRINGIZE(name)))
+
+namespace data {
+namespace impl {
+
+struct WritingAmendmentKey : AmendmentKey {
+    explicit WritingAmendmentKey(std::string amendmentName);
+};
+
+}  // namespace impl
+
+/**
+ * @brief List of supported amendments
+ */
+struct Amendments {
+    // NOTE: if Clio wants to report it supports an Amendment it should be listed here.
+    // Whether an amendment is obsolete and/or supported by libxrpl is extracted directly from libxrpl.
+    // If an amendment is in the list below it just means Clio did whatever changes needed to support it.
+    // Most of the time it's going to be no changes at all.
+
+    /** @cond */
+    REGISTER(OwnerPaysFee);
+    REGISTER(Flow);
+    REGISTER(FlowCross);
+    REGISTER(fix1513);
+    REGISTER(DepositAuth);
+    REGISTER(Checks);
+    REGISTER(fix1571);
+    REGISTER(fix1543);
+    REGISTER(fix1623);
+    REGISTER(DepositPreauth);
+    REGISTER(fix1515);
+    REGISTER(fix1578);
+    REGISTER(MultiSignReserve);
+    REGISTER(fixTakerDryOfferRemoval);
+    REGISTER(fixMasterKeyAsRegularKey);
+    REGISTER(fixCheckThreading);
+    REGISTER(fixPayChanRecipientOwnerDir);
+    REGISTER(DeletableAccounts);
+    REGISTER(fixQualityUpperBound);
+    REGISTER(RequireFullyCanonicalSig);
+    REGISTER(fix1781);
+    REGISTER(HardenedValidations);
+    REGISTER(fixAmendmentMajorityCalc);
+    REGISTER(NegativeUNL);
+    REGISTER(TicketBatch);
+    REGISTER(FlowSortStrands);
+    REGISTER(fixSTAmountCanonicalize);
+    REGISTER(fixRmSmallIncreasedQOffers);
+    REGISTER(CheckCashMakesTrustLine);
+    REGISTER(ExpandedSignerList);
+    REGISTER(NonFungibleTokensV1_1);
+    REGISTER(fixTrustLinesToSelf);
+    REGISTER(fixRemoveNFTokenAutoTrustLine);
+    REGISTER(ImmediateOfferKilled);
+    REGISTER(DisallowIncoming);
+    REGISTER(XRPFees);
+    REGISTER(fixUniversalNumber);
+    REGISTER(fixNonFungibleTokensV1_2);
+    REGISTER(fixNFTokenRemint);
+    REGISTER(fixReducedOffersV1);
+    REGISTER(Clawback);
+    REGISTER(AMM);
+    REGISTER(XChainBridge);
+    REGISTER(fixDisallowIncomingV1);
+    REGISTER(DID);
+    REGISTER(fixFillOrKill);
+    REGISTER(fixNFTokenReserve);
+    REGISTER(fixInnerObjTemplate);
+    REGISTER(fixAMMOverflowOffer);
+    REGISTER(PriceOracle);
+    REGISTER(fixEmptyDID);
+    REGISTER(fixXChainRewardRounding);
+    REGISTER(fixPreviousTxnID);
+    REGISTER(fixAMMv1_1);
+    REGISTER(NFTokenMintOffer);
+    REGISTER(fixReducedOffersV2);
+    REGISTER(fixEnforceNFTokenTrustline);
+
+    // Obsolete but supported by libxrpl
+    REGISTER(CryptoConditionsSuite);
+    REGISTER(NonFungibleTokensV1);
+    REGISTER(fixNFTokenDirV1);
+    REGISTER(fixNFTokenNegOffer);
+
+    // Retired amendments
+    REGISTER(MultiSign);
+    REGISTER(TrustSetAuth);
+    REGISTER(FeeEscalation);
+    REGISTER(PayChan);
+    REGISTER(fix1368);
+    REGISTER(CryptoConditions);
+    REGISTER(Escrow);
+    REGISTER(TickSize);
+    REGISTER(fix1373);
+    REGISTER(EnforceInvariants);
+    REGISTER(SortedDirectories);
+    REGISTER(fix1201);
+    REGISTER(fix1512);
+    REGISTER(fix1523);
+    REGISTER(fix1528);
+    /** @endcond */
+};
+
+#undef REGISTER
+
+/**
+ * @brief Knowledge center for amendments within XRPL
+ */
+class AmendmentCenter : public AmendmentCenterInterface {
+    std::shared_ptr<data::BackendInterface> backend_;
+
+    std::map<std::string, Amendment> supported_;
+    std::vector<Amendment> all_;
+
+public:
+    /**
+     * @brief Construct a new AmendmentCenter instance
+     *
+     * @param backend The backend
+     */
+    explicit AmendmentCenter(std::shared_ptr<data::BackendInterface> const& backend);
+
+    /**
+     * @brief Check whether an amendment is supported by Clio
+     *
+     * @param key The key of the amendment to check
+     * @return true if supported; false otherwise
+     */
+    [[nodiscard]] bool
+    isSupported(AmendmentKey const& key) const final;
+
+    /**
+     * @brief Get all supported amendments as a map
+     *
+     * @return The amendments supported by Clio
+     */
+    [[nodiscard]] std::map<std::string, Amendment> const&
+    getSupported() const final;
+
+    /**
+     * @brief Get all known amendments
+     *
+     * @return All known amendments as a vector
+     */
+    [[nodiscard]] std::vector<Amendment> const&
+    getAll() const final;
+
+    /**
+     * @brief Check whether an amendment was/is enabled for a given sequence
+     *
+     * @param key The key of the amendment to check
+     * @param seq The sequence to check for
+     * @return true if enabled; false otherwise
+     */
+    [[nodiscard]] bool
+    isEnabled(AmendmentKey const& key, uint32_t seq) const final;
+
+    /**
+     * @brief Check whether an amendment was/is enabled for a given sequence
+     *
+     * @param yield The coroutine context to use
+     * @param key The key of the amendment to check
+     * @param seq The sequence to check for
+     * @return true if enabled; false otherwise
+     */
+    [[nodiscard]] bool
+    isEnabled(boost::asio::yield_context yield, AmendmentKey const& key, uint32_t seq) const final;
+
+    /**
+     * @brief Check whether an amendment was/is enabled for a given sequence
+     *
+     * @param yield The coroutine context to use
+     * @param keys The keys of the amendments to check
+     * @param seq The sequence to check for
+     * @return A vector of bools representing enabled state for each of the given keys
+     */
+    [[nodiscard]] std::vector<bool>
+    isEnabled(boost::asio::yield_context yield, std::vector<AmendmentKey> const& keys, uint32_t seq) const final;
+
+    /**
+     * @brief Get an amendment
+     *
+     * @param key The key of the amendment to get
+     * @return The amendment as a const ref; asserts if the amendment is unknown
+     */
+    [[nodiscard]] Amendment const&
+    getAmendment(AmendmentKey const& key) const final;
+
+    /**
+     * @brief Get an amendment by its key
+
+     * @param key The amendment key from @see Amendments
+     * @return The amendment as a const ref; asserts if the amendment is unknown
+     */
+    [[nodiscard]] Amendment const&
+    operator[](AmendmentKey const& key) const final;
+
+private:
+    [[nodiscard]] std::optional<std::vector<ripple::uint256>>
+    fetchAmendmentsList(boost::asio::yield_context yield, uint32_t seq) const;
+};
+
+}  // namespace data

src/data/AmendmentCenterInterface.hpp

@@ -0,0 +1,116 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#pragma once
+
+#include "data/Types.hpp"
+
+#include <boost/asio/spawn.hpp>
+
+#include <cstdint>
+#include <map>
+#include <string>
+#include <vector>
+
+namespace data {
+
+/**
+ * @brief The interface of an amendment center
+ */
+class AmendmentCenterInterface {
+public:
+    virtual ~AmendmentCenterInterface() = default;
+
+    /**
+     * @brief Check whether an amendment is supported by Clio
+     *
+     * @param key The key of the amendment to check
+     * @return true if supported; false otherwise
+     */
+    [[nodiscard]] virtual bool
+    isSupported(AmendmentKey const& key) const = 0;
+
+    /**
+     * @brief Get all supported amendments as a map
+     *
+     * @return The amendments supported by Clio
+     */
+    [[nodiscard]] virtual std::map<std::string, Amendment> const&
+    getSupported() const = 0;
+
+    /**
+     * @brief Get all known amendments
+     *
+     * @return All known amendments as a vector
+     */
+    [[nodiscard]] virtual std::vector<Amendment> const&
+    getAll() const = 0;
+
+    /**
+     * @brief Check whether an amendment was/is enabled for a given sequence
+     *
+     * @param key The key of the amendment to check
+     * @param seq The sequence to check for
+     * @return true if enabled; false otherwise
+     */
+    [[nodiscard]] virtual bool
+    isEnabled(AmendmentKey const& key, uint32_t seq) const = 0;
+
+    /**
+     * @brief Check whether an amendment was/is enabled for a given sequence
+     *
+     * @param yield The coroutine context to use
+     * @param key The key of the amendment to check
+     * @param seq The sequence to check for
+     * @return true if enabled; false otherwise
+     */
+    [[nodiscard]] virtual bool
+    isEnabled(boost::asio::yield_context yield, AmendmentKey const& key, uint32_t seq) const = 0;
+
+    /**
+     * @brief Check whether an amendment was/is enabled for a given sequence
+     *
+     * @param yield The coroutine context to use
+     * @param keys The keys of the amendments to check
+     * @param seq The sequence to check for
+     * @return A vector of bools representing enabled state for each of the given keys
+     */
+    [[nodiscard]] virtual std::vector<bool>
+    isEnabled(boost::asio::yield_context yield, std::vector<AmendmentKey> const& keys, uint32_t seq) const = 0;
+
+    /**
+     * @brief Get an amendment
+     *
+     * @param key The key of the amendment to get
+     * @return The amendment as a const ref; asserts if the amendment is unknown
+     */
+    [[nodiscard]] virtual Amendment const&
+    getAmendment(AmendmentKey const& key) const = 0;
+
+    /**
+     * @brief Get an amendment by its key
+     *
+     * @param key The amendment key from @see Amendments
+     * @return The amendment as a const ref; asserts if the amendment is unknown
+     */
+    [[nodiscard]] virtual Amendment const&
+    operator[](AmendmentKey const& key) const = 0;
+};
+
+}  // namespace data

src/data/BackendFactory.hpp

@@ -51,8 +51,7 @@ make_Backend(util::Config const& config)
     auto const type = config.value<std::string>("database.type");
     std::shared_ptr<BackendInterface> backend = nullptr;
 
-    // TODO: retire `cassandra-new` by next release after 2.0
-    if (boost::iequals(type, "cassandra") or boost::iequals(type, "cassandra-new")) {
+    if (boost::iequals(type, "cassandra")) {
         auto cfg = config.section("database." + type);
         backend = std::make_shared<data::cassandra::CassandraBackend>(data::cassandra::SettingsProvider{cfg}, readOnly);
     }

src/data/BackendInterface.cpp

@@ -24,13 +24,13 @@
 #include "util/log/Logger.hpp"
 
 #include <boost/asio/spawn.hpp>
-#include <ripple/basics/base_uint.h>
-#include <ripple/basics/strHex.h>
-#include <ripple/protocol/Fees.h>
-#include <ripple/protocol/Indexes.h>
-#include <ripple/protocol/SField.h>
-#include <ripple/protocol/STLedgerEntry.h>
-#include <ripple/protocol/Serializer.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/strHex.h>
+#include <xrpl/protocol/Fees.h>
+#include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STLedgerEntry.h>
+#include <xrpl/protocol/Serializer.h>
 
 #include <chrono>
 #include <cstddef>
@@ -93,7 +93,6 @@ BackendInterface::fetchLedgerObject(
         return obj;
     }
 
-    LOG(gLog.trace()) << "Cache miss - " << ripple::strHex(key);
     auto dbObj = doFetchLedgerObject(key, sequence, yield);
     if (!dbObj) {
         LOG(gLog.trace()) << "Missed cache and missed in db";
@@ -103,6 +102,19 @@ BackendInterface::fetchLedgerObject(
     return dbObj;
 }
 
+std::optional<std::uint32_t>
+BackendInterface::fetchLedgerObjectSeq(
+    ripple::uint256 const& key,
+    std::uint32_t const sequence,
+    boost::asio::yield_context yield
+) const
+{
+    auto seq = doFetchLedgerObjectSeq(key, sequence, yield);
+    if (!seq)
+        LOG(gLog.trace()) << "Missed in db";
+    return seq;
+}
+
 std::vector<Blob>
 BackendInterface::fetchLedgerObjects(
     std::vector<ripple::uint256> const& keys,

src/data/BackendInterface.hpp

@@ -31,10 +31,10 @@
 #include <boost/json.hpp>
 #include <boost/json/object.hpp>
 #include <boost/utility/result_of.hpp>
-#include <ripple/basics/base_uint.h>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/Fees.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Fees.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <chrono>
 #include <cstddef>
@@ -378,6 +378,19 @@ public:
     std::optional<Blob>
     fetchLedgerObject(ripple::uint256 const& key, std::uint32_t sequence, boost::asio::yield_context yield) const;
 
+    /**
+     * @brief Fetches a specific ledger object sequence.
+     *
+     * Currently the real fetch happens in doFetchLedgerObjectSeq
+     *
+     * @param key The key of the object
+     * @param sequence The ledger sequence to fetch for
+     * @param yield The coroutine context
+     * @return The sequence in unit32_t on success; nullopt otherwise
+     */
+    std::optional<std::uint32_t>
+    fetchLedgerObjectSeq(ripple::uint256 const& key, std::uint32_t sequence, boost::asio::yield_context yield) const;
+
     /**
      * @brief Fetches all ledger objects by their keys.
      *
@@ -407,6 +420,18 @@ public:
     virtual std::optional<Blob>
     doFetchLedgerObject(ripple::uint256 const& key, std::uint32_t sequence, boost::asio::yield_context yield) const = 0;
 
+    /**
+     * @brief The database-specific implementation for fetching a ledger object sequence.
+     *
+     * @param key The key to fetch for
+     * @param sequence The ledger sequence to fetch for
+     * @param yield The coroutine context
+     * @return The sequence in unit32_t on success; nullopt otherwise
+     */
+    virtual std::optional<std::uint32_t>
+    doFetchLedgerObjectSeq(ripple::uint256 const& key, std::uint32_t sequence, boost::asio::yield_context yield)
+        const = 0;
+
     /**
      * @brief The database-specific implementation for fetching ledger objects.
      *

src/data/CMakeLists.txt

@@ -1,7 +1,8 @@
 add_library(clio_data)
 target_sources(
   clio_data
-  PRIVATE BackendCounters.cpp
+  PRIVATE AmendmentCenter.cpp
+          BackendCounters.cpp
           BackendInterface.cpp
           LedgerCache.cpp
           cassandra/impl/Future.cpp

src/data/CassandraBackend.hpp

@@ -22,7 +22,6 @@
 #include "data/BackendInterface.hpp"
 #include "data/DBHelpers.hpp"
 #include "data/Types.hpp"
-#include "data/cassandra/Concepts.hpp"
 #include "data/cassandra/Handle.hpp"
 #include "data/cassandra/Schema.hpp"
 #include "data/cassandra/SettingsProvider.hpp"
@@ -36,13 +35,13 @@
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
 #include <cassandra.h>
-#include <ripple/basics/Blob.h>
-#include <ripple/basics/base_uint.h>
-#include <ripple/basics/strHex.h>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/Indexes.h>
-#include <ripple/protocol/LedgerHeader.h>
-#include <ripple/protocol/nft.h>
+#include <xrpl/basics/Blob.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/strHex.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/LedgerHeader.h>
+#include <xrpl/protocol/nft.h>
 
 #include <atomic>
 #include <chrono>
@@ -94,7 +93,7 @@ public:
         , executor_{settingsProvider_.getSettings(), handle_}
     {
         if (auto const res = handle_.connect(); not res)
-            throw std::runtime_error("Could not connect to Cassandra: " + res.error());
+            throw std::runtime_error("Could not connect to databse: " + res.error());
 
         if (not readOnly) {
             if (auto const res = handle_.execute(schema_.createKeyspace); not res) {
@@ -336,8 +335,8 @@ public:
 
         auto const& result = res.value();
         if (not result.hasRows()) {
-            LOG(log_.error()) << "Could not fetch all transaction hashes - no rows; ledger = "
-                              << std::to_string(ledgerSequence);
+            LOG(log_.warn()) << "Could not fetch all transaction hashes - no rows; ledger = "
+                             << std::to_string(ledgerSequence);
             return {};
         }
 
@@ -346,7 +345,7 @@ public:
             hashes.push_back(std::move(hash));
 
         auto end = std::chrono::system_clock::now();
-        LOG(log_.debug()) << "Fetched " << hashes.size() << " transaction hashes from Cassandra in "
+        LOG(log_.debug()) << "Fetched " << hashes.size() << " transaction hashes from database in "
                           << std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count()
                           << " milliseconds";
 
@@ -567,6 +566,25 @@ public:
         return std::nullopt;
     }
 
+    std::optional<std::uint32_t>
+    doFetchLedgerObjectSeq(ripple::uint256 const& key, std::uint32_t const sequence, boost::asio::yield_context yield)
+        const override
+    {
+        LOG(log_.debug()) << "Fetching ledger object for seq " << sequence << ", key = " << ripple::to_string(key);
+        if (auto const res = executor_.read(yield, schema_->selectObject, key, sequence); res) {
+            if (auto const result = res->template get<Blob, std::uint32_t>(); result) {
+                auto [_, seq] = result.value();
+                return seq;
+            }
+            LOG(log_.debug()) << "Could not fetch ledger object sequence - no rows";
+
+        } else {
+            LOG(log_.error()) << "Could not fetch ledger object sequence: " << res.error();
+        }
+
+        return std::nullopt;
+    }
+
     std::optional<TransactionAndMetadata>
     fetchTransaction(ripple::uint256 const& hash, boost::asio::yield_context yield) const override
     {
@@ -640,7 +658,7 @@ public:
         });
 
         ASSERT(numHashes == results.size(), "Number of hashes and results must match");
-        LOG(log_.debug()) << "Fetched " << numHashes << " transactions from Cassandra in " << timeDiff
+        LOG(log_.debug()) << "Fetched " << numHashes << " transactions from database in " << timeDiff
                           << " milliseconds";
         return results;
     }
@@ -760,7 +778,7 @@ public:
         if (keys.empty())
             return {};
 
-        LOG(log_.debug()) << "Fetched " << keys.size() << " diff hashes from Cassandra in " << timeDiff
+        LOG(log_.debug()) << "Fetched " << keys.size() << " diff hashes from database in " << timeDiff
                           << " milliseconds";
 
         auto const objs = fetchLedgerObjects(keys, ledgerSequence, yield);
@@ -848,7 +866,7 @@ public:
         std::string&& metadata
     ) override
     {
-        LOG(log_.trace()) << "Writing txn to cassandra";
+        LOG(log_.trace()) << "Writing txn to database";
 
         executor_.write(schema_->insertLedgerTransaction, seq, hash);
         executor_.write(

src/data/DBHelpers.hpp

@@ -23,16 +23,16 @@
 #include "util/Assert.hpp"
 
 #include <boost/container/flat_set.hpp>
-#include <ripple/basics/Blob.h>
-#include <ripple/basics/Log.h>
-#include <ripple/basics/StringUtilities.h>
-#include <ripple/basics/base_uint.h>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/SField.h>
-#include <ripple/protocol/STAccount.h>
-#include <ripple/protocol/STLedgerEntry.h>
-#include <ripple/protocol/Serializer.h>
-#include <ripple/protocol/TxMeta.h>
+#include <xrpl/basics/Blob.h>
+#include <xrpl/basics/Log.h>
+#include <xrpl/basics/StringUtilities.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STAccount.h>
+#include <xrpl/protocol/STLedgerEntry.h>
+#include <xrpl/protocol/Serializer.h>
+#include <xrpl/protocol/TxMeta.h>
 
 #include <cstddef>
 #include <cstdint>

src/data/LedgerCache.cpp

@@ -22,7 +22,7 @@
 #include "data/Types.hpp"
 #include "util/Assert.hpp"
 
-#include <ripple/basics/base_uint.h>
+#include <xrpl/basics/base_uint.h>
 
 #include <cstddef>
 #include <cstdint>

src/data/LedgerCache.hpp

@@ -24,8 +24,8 @@
 #include "util/prometheus/Label.hpp"
 #include "util/prometheus/Prometheus.hpp"
 
-#include <ripple/basics/base_uint.h>
-#include <ripple/basics/hardened_hash.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/hardened_hash.h>
 
 #include <atomic>
 #include <condition_variable>

src/data/Types.hpp

@@ -19,11 +19,13 @@
 
 #pragma once
 
-#include <ripple/basics/base_uint.h>
-#include <ripple/protocol/AccountID.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/AccountID.h>
 
+#include <concepts>
 #include <cstdint>
 #include <optional>
+#include <string>
 #include <string_view>
 #include <tuple>
 #include <utility>
@@ -239,6 +241,69 @@ struct LedgerRange {
     std::uint32_t maxSequence = 0;
 };
 
+/**
+ * @brief Represents an amendment in the XRPL
+ */
+struct Amendment {
+    std::string name;
+    ripple::uint256 feature;
+    bool isSupportedByXRPL = false;
+    bool isSupportedByClio = false;
+    bool isRetired = false;
+
+    /**
+     * @brief Get the amendment Id from its name
+     *
+     * @param name The name of the amendment
+     * @return The amendment Id as uint256
+     */
+    static ripple::uint256
+    GetAmendmentId(std::string_view const name);
+
+    /**
+     * @brief Equality comparison operator
+     * @param other The object to compare to
+     * @return Whether the objects are equal
+     */
+    bool
+    operator==(Amendment const& other) const
+    {
+        return name == other.name;
+    }
+};
+
+/**
+ * @brief A helper for amendment name to feature conversions
+ */
+struct AmendmentKey {
+    std::string name;
+
+    /**
+     * @brief Construct a new AmendmentKey
+     * @param val Anything convertible to a string
+     */
+    AmendmentKey(std::convertible_to<std::string> auto&& val) : name{std::forward<decltype(val)>(val)}
+    {
+    }
+
+    /** @brief Conversion to string */
+    operator std::string const&() const;
+
+    /** @brief Conversion to string_view */
+    operator std::string_view() const;
+
+    /** @brief Conversion to uint256 */
+    operator ripple::uint256() const;
+
+    /**
+     * @brief Comparison operators
+     * @param other The object to compare to
+     * @return Whether the objects are equal, greater or less
+     */
+    auto
+    operator<=>(AmendmentKey const& other) const = default;
+};
+
 constexpr ripple::uint256 firstKey{"0000000000000000000000000000000000000000000000000000000000000000"};
 constexpr ripple::uint256 lastKey{"FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF"};
 constexpr ripple::uint256 hi192{"0000000000000000000000000000000000000000000000001111111111111111"};

src/data/cassandra/impl/Cluster.cpp

@@ -21,6 +21,7 @@
 
 #include "data/cassandra/impl/ManagedObject.hpp"
 #include "data/cassandra/impl/SslContext.hpp"
+#include "util/OverloadSet.hpp"
 #include "util/log/Logger.hpp"
 
 #include <cassandra.h>
@@ -31,16 +32,9 @@
 #include <variant>
 
 namespace {
+
 constexpr auto clusterDeleter = [](CassCluster* ptr) { cass_cluster_free(ptr); };
 
-template <typename... Ts>
-struct overloadSet : Ts... {
-    using Ts::operator()...;
-};
-
-// explicit deduction guide (not needed as of C++20, but clang be clang)
-template <typename... Ts>
-overloadSet(Ts...) -> overloadSet<Ts...>;
 };  // namespace
 
 namespace data::cassandra::impl {
@@ -90,7 +84,7 @@ void
 Cluster::setupConnection(Settings const& settings)
 {
     std::visit(
-        overloadSet{
+        util::OverloadSet{
             [this](Settings::ContactPoints const& points) { setupContactPoints(points); },
             [this](Settings::SecureConnectionBundle const& bundle) { setupSecureBundle(bundle); }
         },

src/data/cassandra/impl/Collection.hpp

@@ -22,7 +22,7 @@
 #include "data/cassandra/impl/ManagedObject.hpp"
 
 #include <cassandra.h>
-#include <ripple/basics/base_uint.h>
+#include <xrpl/basics/base_uint.h>
 
 #include <cstdint>
 #include <stdexcept>

src/data/cassandra/impl/Result.hpp

@@ -21,10 +21,11 @@
 
 #include "data/cassandra/impl/ManagedObject.hpp"
 #include "data/cassandra/impl/Tuple.hpp"
+#include "util/UnsupportedType.hpp"
 
 #include <cassandra.h>
-#include <ripple/basics/base_uint.h>
-#include <ripple/protocol/AccountID.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/AccountID.h>
 
 #include <cstddef>
 #include <cstdint>
@@ -41,9 +42,6 @@
 
 namespace data::cassandra::impl {
 
-template <typename>
-static constexpr bool unsupported_v = false;
-
 template <typename Type>
 inline Type
 extractColumn(CassRow const* row, std::size_t idx)
@@ -103,7 +101,7 @@ extractColumn(CassRow const* row, std::size_t idx)
         output = static_cast<DecayedType>(out);
     } else {
         // type not supported for extraction
-        static_assert(unsupported_v<DecayedType>);
+        static_assert(util::Unsupported<DecayedType>);
     }
 
     return output;

src/data/cassandra/impl/Statement.hpp

@@ -23,12 +23,13 @@
 #include "data/cassandra/impl/Collection.hpp"
 #include "data/cassandra/impl/ManagedObject.hpp"
 #include "data/cassandra/impl/Tuple.hpp"
+#include "util/UnsupportedType.hpp"
 
 #include <cassandra.h>
 #include <fmt/core.h>
-#include <ripple/basics/base_uint.h>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/STAccount.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/STAccount.h>
 
 #include <cstddef>
 #include <cstdint>
@@ -44,9 +45,6 @@ namespace data::cassandra::impl {
 class Statement : public ManagedObject<CassStatement> {
     static constexpr auto deleter = [](CassStatement* ptr) { cass_statement_free(ptr); };
 
-    template <typename>
-    static constexpr bool unsupported_v = false;
-
 public:
     /**
      * @brief Construct a new statement with optionally provided arguments.
@@ -141,7 +139,7 @@ public:
             throwErrorIfNeeded(rc, "Bind int64");
         } else {
             // type not supported for binding
-            static_assert(unsupported_v<DecayedType>);
+            static_assert(util::Unsupported<DecayedType>);
         }
     }
 };

src/data/cassandra/impl/Tuple.hpp

@@ -20,9 +20,10 @@
 #pragma once
 
 #include "data/cassandra/impl/ManagedObject.hpp"
+#include "util/UnsupportedType.hpp"
 
 #include <cassandra.h>
-#include <ripple/basics/base_uint.h>
+#include <xrpl/basics/base_uint.h>
 
 #include <cstddef>
 #include <cstdint>
@@ -38,9 +39,6 @@ namespace data::cassandra::impl {
 class Tuple : public ManagedObject<CassTuple> {
     static constexpr auto deleter = [](CassTuple* ptr) { cass_tuple_free(ptr); };
 
-    template <typename>
-    static constexpr bool unsupported_v = false;
-
 public:
     /* implicit */ Tuple(CassTuple* ptr);
 
@@ -91,15 +89,12 @@ public:
             throwErrorIfNeeded(rc, "Bind ripple::uint256");
         } else {
             // type not supported for binding
-            static_assert(unsupported_v<DecayedType>);
+            static_assert(util::Unsupported<DecayedType>);
         }
     }
 };
 
 class TupleIterator : public ManagedObject<CassIterator> {
-    template <typename>
-    static constexpr bool unsupported_v = false;
-
 public:
     /* implicit */ TupleIterator(CassIterator* ptr);
 
@@ -141,7 +136,7 @@ private:
             output = static_cast<DecayedType>(out);
         } else {
             // type not supported for extraction
-            static_assert(unsupported_v<DecayedType>);
+            static_assert(util::Unsupported<DecayedType>);
         }
 
         return output;

src/etl/CMakeLists.txt

@@ -10,6 +10,7 @@ target_sources(
           NetworkValidatedLedgers.cpp
           NFTHelpers.cpp
           Source.cpp
+          impl/AmendmentBlockHandler.cpp
           impl/ForwardingCache.cpp
           impl/ForwardingSource.cpp
           impl/GrpcSource.cpp

src/etl/ETLHelpers.cpp

@@ -21,7 +21,7 @@
 
 #include "util/Assert.hpp"
 
-#include <ripple/basics/base_uint.h>
+#include <xrpl/basics/base_uint.h>
 
 #include <cstddef>
 #include <vector>

src/etl/ETLHelpers.hpp

@@ -20,7 +20,7 @@
 /** @file */
 #pragma once
 
-#include <ripple/basics/base_uint.h>
+#include <xrpl/basics/base_uint.h>
 
 #include <condition_variable>
 #include <cstddef>

src/etl/ETLService.cpp

@@ -30,8 +30,8 @@
 #include "util/log/Logger.hpp"
 
 #include <boost/asio/io_context.hpp>
-#include <ripple/beast/core/CurrentThreadName.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/beast/core/CurrentThreadName.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <chrono>
 #include <cstddef>

src/etl/ETLService.hpp

@@ -22,11 +22,11 @@
 #include "data/BackendInterface.hpp"
 #include "data/LedgerCache.hpp"
 #include "etl/CacheLoader.hpp"
-#include "etl/ETLHelpers.hpp"
 #include "etl/ETLState.hpp"
 #include "etl/LoadBalancer.hpp"
+#include "etl/NetworkValidatedLedgersInterface.hpp"
 #include "etl/SystemState.hpp"
-#include "etl/impl/AmendmentBlock.hpp"
+#include "etl/impl/AmendmentBlockHandler.hpp"
 #include "etl/impl/ExtractionDataPipe.hpp"
 #include "etl/impl/Extractor.hpp"
 #include "etl/impl/LedgerFetcher.hpp"
@@ -37,11 +37,10 @@
 #include "util/log/Logger.hpp"
 
 #include <boost/asio/io_context.hpp>
-#include <boost/asio/steady_timer.hpp>
 #include <boost/json/object.hpp>
 #include <grpcpp/grpcpp.h>
 #include <org/xrpl/rpc/v1/get_ledger.pb.h>
-#include <ripple/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
+#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
 
 #include <cstddef>
 #include <cstdint>
@@ -53,9 +52,6 @@
 struct AccountTransactionsData;
 struct NFTTransactionsData;
 struct NFTsData;
-namespace feed {
-class SubscriptionManager;
-}  // namespace feed
 
 /**
  * @brief This namespace contains everything to do with the ETL and ETL sources.
@@ -85,7 +81,7 @@ class ETLService {
     using ExtractorType = etl::impl::Extractor<DataPipeType, LedgerFetcherType>;
     using LedgerLoaderType = etl::impl::LedgerLoader<LoadBalancerType, LedgerFetcherType>;
     using LedgerPublisherType = etl::impl::LedgerPublisher<CacheType>;
-    using AmendmentBlockHandlerType = etl::impl::AmendmentBlockHandler<>;
+    using AmendmentBlockHandlerType = etl::impl::AmendmentBlockHandler;
     using TransformerType =
         etl::impl::Transformer<DataPipeType, LedgerLoaderType, LedgerPublisherType, AmendmentBlockHandlerType>;
 

src/etl/ETLState.cpp

@@ -24,7 +24,7 @@
 #include <boost/json/conversion.hpp>
 #include <boost/json/value.hpp>
 #include <boost/json/value_to.hpp>
-#include <ripple/protocol/jss.h>
+#include <xrpl/protocol/jss.h>
 
 #include <cstdint>
 #include <optional>

src/etl/ETLState.hpp

@@ -23,6 +23,7 @@
 
 #include <boost/json.hpp>
 #include <boost/json/conversion.hpp>
+#include <boost/json/object.hpp>
 #include <boost/json/value.hpp>
 #include <boost/json/value_to.hpp>
 
@@ -46,8 +47,11 @@ struct ETLState {
     static std::optional<ETLState>
     fetchETLStateFromSource(Forward& source) noexcept
     {
-        auto const serverInfoRippled = data::synchronous([&source](auto yield) {
-            return source.forwardToRippled({{"command", "server_info"}}, std::nullopt, {}, yield);
+        auto const serverInfoRippled = data::synchronous([&source](auto yield) -> std::optional<boost::json::object> {
+            if (auto result = source.forwardToRippled({{"command", "server_info"}}, std::nullopt, {}, yield)) {
+                return std::move(result).value();
+            }
+            return std::nullopt;
         });
 
         if (serverInfoRippled)

src/etl/LoadBalancer.cpp

@@ -24,6 +24,7 @@
 #include "etl/NetworkValidatedLedgersInterface.hpp"
 #include "etl/Source.hpp"
 #include "feed/SubscriptionManagerInterface.hpp"
+#include "rpc/Errors.hpp"
 #include "util/Assert.hpp"
 #include "util/Random.hpp"
 #include "util/log/Logger.hpp"
@@ -39,6 +40,7 @@
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
+#include <expected>
 #include <memory>
 #include <optional>
 #include <stdexcept>
@@ -109,13 +111,10 @@ LoadBalancer::LoadBalancer(
             validatedLedgers,
             forwardingTimeout,
             [this]() {
-                if (not hasForwardingSource_.lock().get())
-                    chooseForwardingSource();
-            },
-            [this](bool wasForwarding) {
-                if (wasForwarding)
+                if (not hasForwardingSource_)
                     chooseForwardingSource();
             },
+            [this]() { chooseForwardingSource(); },
             [this]() {
                 if (forwardingCache_.has_value())
                     forwardingCache_->invalidate();
@@ -214,7 +213,7 @@ LoadBalancer::fetchLedger(
     return response;
 }
 
-std::optional<boost::json::object>
+std::expected<boost::json::object, rpc::ClioError>
 LoadBalancer::forwardToRippled(
     boost::json::object const& request,
     std::optional<std::string> const& clientIp,
@@ -224,7 +223,7 @@ LoadBalancer::forwardToRippled(
 {
     if (forwardingCache_) {
         if (auto cachedResponse = forwardingCache_->get(request); cachedResponse) {
-            return cachedResponse;
+            return std::move(cachedResponse).value();
         }
     }
 
@@ -236,20 +235,26 @@ LoadBalancer::forwardToRippled(
     auto xUserValue = isAdmin ? ADMIN_FORWARDING_X_USER_VALUE : USER_FORWARDING_X_USER_VALUE;
 
     std::optional<boost::json::object> response;
+    rpc::ClioError error = rpc::ClioError::etlCONNECTION_ERROR;
     while (numAttempts < sources_.size()) {
-        if (auto res = sources_[sourceIdx]->forwardToRippled(request, clientIp, xUserValue, yield)) {
-            response = std::move(res);
+        auto res = sources_[sourceIdx]->forwardToRippled(request, clientIp, xUserValue, yield);
+        if (res) {
+            response = std::move(res).value();
             break;
         }
+        error = std::max(error, res.error());  // Choose the best result between all sources
 
         sourceIdx = (sourceIdx + 1) % sources_.size();
         ++numAttempts;
     }
 
-    if (response and forwardingCache_ and not response->contains("error"))
-        forwardingCache_->put(request, *response);
+    if (response) {
+        if (forwardingCache_ and not response->contains("error"))
+            forwardingCache_->put(request, *response);
+        return std::move(response).value();
+    }
 
-    return response;
+    return std::unexpected{error};
 }
 
 boost::json::value
@@ -317,13 +322,11 @@ LoadBalancer::getETLState() noexcept
 void
 LoadBalancer::chooseForwardingSource()
 {
-    LOG(log_.info()) << "Choosing a new source to forward subscriptions";
-    auto hasForwardingSourceLock = hasForwardingSource_.lock();
-    hasForwardingSourceLock.get() = false;
+    hasForwardingSource_ = false;
     for (auto& source : sources_) {
-        if (not hasForwardingSourceLock.get() and source->isConnected()) {
+        if (not hasForwardingSource_ and source->isConnected()) {
             source->setForwarding(true);
-            hasForwardingSourceLock.get() = true;
+            hasForwardingSource_ = true;
         } else {
             source->setForwarding(false);
         }

src/etl/LoadBalancer.hpp

@@ -25,7 +25,7 @@
 #include "etl/Source.hpp"
 #include "etl/impl/ForwardingCache.hpp"
 #include "feed/SubscriptionManagerInterface.hpp"
-#include "util/Mutex.hpp"
+#include "rpc/Errors.hpp"
 #include "util/config/Config.hpp"
 #include "util/log/Logger.hpp"
 
@@ -37,10 +37,12 @@
 #include <grpcpp/grpcpp.h>
 #include <org/xrpl/rpc/v1/get_ledger.pb.h>
 #include <org/xrpl/rpc/v1/ledger.pb.h>
-#include <ripple/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
+#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
 
+#include <atomic>
 #include <chrono>
 #include <cstdint>
+#include <expected>
 #include <memory>
 #include <optional>
 #include <string>
@@ -74,10 +76,7 @@ private:
     std::optional<ETLState> etlState_;
     std::uint32_t downloadRanges_ =
         DEFAULT_DOWNLOAD_RANGES; /*< The number of markers to use when downloading initial ledger */
-
-    // Using mutext instead of atomic_bool because choosing a new source to
-    // forward messages should be done with a mutual exclusion otherwise there will be a race condition
-    util::Mutex<bool> hasForwardingSource_{false};
+    std::atomic_bool hasForwardingSource_{false};
 
 public:
     /**
@@ -184,9 +183,9 @@ public:
      * @param clientIp The IP address of the peer, if known
      * @param isAdmin Whether the request is from an admin
      * @param yield The coroutine context
-     * @return Response received from rippled node as JSON object on success; nullopt on failure
+     * @return Response received from rippled node as JSON object on success or error on failure
      */
-    std::optional<boost::json::object>
+    std::expected<boost::json::object, rpc::ClioError>
     forwardToRippled(
         boost::json::object const& request,
         std::optional<std::string> const& clientIp,

src/etl/NFTHelpers.cpp

@@ -20,20 +20,20 @@
 #include "data/DBHelpers.hpp"
 
 #include <fmt/core.h>
-#include <ripple/basics/base_uint.h>
-#include <ripple/basics/strHex.h>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/LedgerFormats.h>
-#include <ripple/protocol/SField.h>
-#include <ripple/protocol/STArray.h>
-#include <ripple/protocol/STBase.h>
-#include <ripple/protocol/STLedgerEntry.h>
-#include <ripple/protocol/STObject.h>
-#include <ripple/protocol/STTx.h>
-#include <ripple/protocol/Serializer.h>
-#include <ripple/protocol/TER.h>
-#include <ripple/protocol/TxFormats.h>
-#include <ripple/protocol/TxMeta.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/strHex.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/LedgerFormats.h>
+#include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STArray.h>
+#include <xrpl/protocol/STBase.h>
+#include <xrpl/protocol/STLedgerEntry.h>
+#include <xrpl/protocol/STObject.h>
+#include <xrpl/protocol/STTx.h>
+#include <xrpl/protocol/Serializer.h>
+#include <xrpl/protocol/TER.h>
+#include <xrpl/protocol/TxFormats.h>
+#include <xrpl/protocol/TxMeta.h>
 
 #include <algorithm>
 #include <cstdint>
@@ -116,8 +116,8 @@ getNFTokenMintData(ripple::TxMeta const& txMeta, ripple::STTx const& sttx)
         }
     }
 
-    std::sort(finalIDs.begin(), finalIDs.end());
-    std::sort(prevIDs.begin(), prevIDs.end());
+    std::ranges::sort(finalIDs);
+    std::ranges::sort(prevIDs);
 
     // Find the first NFT ID that doesn't match.  We're looking for an
     // added NFT, so the one we want will be the mismatch in finalIDs.
@@ -294,14 +294,12 @@ getNFTokenCancelOfferData(ripple::TxMeta const& txMeta, ripple::STTx const& sttx
         txs.emplace_back(tokenID, txMeta, sttx.getTransactionID());
     }
 
-    // Deduplicate any transactions based on tokenID/txIdx combo. Can't just
-    // use txIdx because in this case one tx can cancel offers for several
-    // NFTs.
-    std::sort(txs.begin(), txs.end(), [](NFTTransactionsData const& a, NFTTransactionsData const& b) {
-        return a.tokenID < b.tokenID && a.transactionIndex < b.transactionIndex;
+    // Deduplicate any transactions based on tokenID
+    std::ranges::sort(txs, [](NFTTransactionsData const& a, NFTTransactionsData const& b) {
+        return a.tokenID < b.tokenID;
     });
     auto last = std::unique(txs.begin(), txs.end(), [](NFTTransactionsData const& a, NFTTransactionsData const& b) {
-        return a.tokenID == b.tokenID && a.transactionIndex == b.transactionIndex;
+        return a.tokenID == b.tokenID;
     });
     txs.erase(last, txs.end());
     return {txs, {}};
@@ -358,4 +356,21 @@ getNFTDataFromObj(std::uint32_t const seq, std::string const& key, std::string c
 
     return nfts;
 }
+
+std::vector<NFTsData>
+getUniqueNFTsDatas(std::vector<NFTsData> const& nfts)
+{
+    std::vector<NFTsData> results = nfts;
+
+    std::ranges::sort(results, [](NFTsData const& a, NFTsData const& b) {
+        return a.tokenID == b.tokenID ? a.transactionIndex > b.transactionIndex : a.tokenID > b.tokenID;
+    });
+
+    auto const last = std::unique(results.begin(), results.end(), [](NFTsData const& a, NFTsData const& b) {
+        return a.tokenID == b.tokenID;
+    });
+    results.erase(last, results.end());
+    return results;
+}
+
 }  // namespace etl

src/etl/NFTHelpers.hpp

@@ -22,8 +22,8 @@
 
 #include "data/DBHelpers.hpp"
 
-#include <ripple/protocol/STTx.h>
-#include <ripple/protocol/TxMeta.h>
+#include <xrpl/protocol/STTx.h>
+#include <xrpl/protocol/TxMeta.h>
 
 #include <cstdint>
 #include <optional>
@@ -104,4 +104,14 @@ getNFTDataFromTx(ripple::TxMeta const& txMeta, ripple::STTx const& sttx);
 std::vector<NFTsData>
 getNFTDataFromObj(std::uint32_t seq, std::string const& key, std::string const& blob);
 
+/**
+ * @brief Get the unique NFTs data from a vector of NFTsData happening in the same ledger. For example, if a NFT has
+ * both accept offer and burn happening in the same ledger,we only keep the final state of the NFT.
+
+ * @param nfts The NFTs data to filter, happening in the same ledger
+ * @return The unique NFTs data
+ */
+std::vector<NFTsData>
+getUniqueNFTsDatas(std::vector<NFTsData> const& nfts);
+
 }  // namespace etl

src/etl/Source.hpp

@@ -22,6 +22,7 @@
 #include "data/BackendInterface.hpp"
 #include "etl/NetworkValidatedLedgersInterface.hpp"
 #include "feed/SubscriptionManagerInterface.hpp"
+#include "rpc/Errors.hpp"
 #include "util/config/Config.hpp"
 #include "util/log/Logger.hpp"
 
@@ -34,6 +35,7 @@
 
 #include <chrono>
 #include <cstdint>
+#include <expected>
 #include <functional>
 #include <memory>
 #include <optional>
@@ -51,7 +53,7 @@ namespace etl {
 class SourceBase {
 public:
     using OnConnectHook = std::function<void()>;
-    using OnDisconnectHook = std::function<void(bool)>;
+    using OnDisconnectHook = std::function<void()>;
     using OnLedgerClosedHook = std::function<void()>;
 
     virtual ~SourceBase() = default;
@@ -131,9 +133,9 @@ public:
      * @param forwardToRippledClientIp IP of the client forwarding this request if known
      * @param xUserValue Value of the X-User header
      * @param yield The coroutine context
-     * @return Response wrapped in an optional on success; nullopt otherwise
+     * @return Response on success or error on failure
      */
-    virtual std::optional<boost::json::object>
+    virtual std::expected<boost::json::object, rpc::ClioError>
     forwardToRippled(
         boost::json::object const& request,
         std::optional<std::string> const& forwardToRippledClientIp,

src/etl/impl/AmendmentBlock.hpp

@@ -1,95 +0,0 @@
-//------------------------------------------------------------------------------
-/*
-    This file is part of clio: https://github.com/XRPLF/clio
-    Copyright (c) 2023, the clio developers.
-
-    Permission to use, copy, modify, and distribute this software for any
-    purpose with or without fee is hereby granted, provided that the above
-    copyright notice and this permission notice appear in all copies.
-
-    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
-    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
-    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
-    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
-    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
-    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
-    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-*/
-//==============================================================================
-
-#pragma once
-
-#include "etl/SystemState.hpp"
-#include "util/log/Logger.hpp"
-
-#include <boost/asio/io_context.hpp>
-#include <boost/asio/post.hpp>
-#include <boost/asio/steady_timer.hpp>
-
-#include <chrono>
-#include <functional>
-
-namespace etl::impl {
-
-struct AmendmentBlockAction {
-    void
-    operator()()
-    {
-        static util::Logger const log{"ETL"};
-        LOG(log.fatal()) << "Can't process new ledgers: The current ETL source is not compatible with the version of "
-                         << "the libxrpl Clio is currently using. Please upgrade Clio to a newer version.";
-    }
-};
-
-template <typename ActionCallableType = AmendmentBlockAction>
-class AmendmentBlockHandler {
-    std::reference_wrapper<boost::asio::io_context> ctx_;
-    std::reference_wrapper<SystemState> state_;
-    boost::asio::steady_timer timer_;
-    std::chrono::milliseconds interval_;
-
-    ActionCallableType action_;
-
-public:
-    template <typename DurationType = std::chrono::seconds>
-    AmendmentBlockHandler(
-        boost::asio::io_context& ioc,
-        SystemState& state,
-        DurationType interval = DurationType{1},
-        ActionCallableType&& action = ActionCallableType()
-    )
-        : ctx_{std::ref(ioc)}
-        , state_{std::ref(state)}
-        , timer_{ioc}
-        , interval_{std::chrono::duration_cast<std::chrono::milliseconds>(interval)}
-        , action_{std::move(action)}
-    {
-    }
-
-    ~AmendmentBlockHandler()
-    {
-        boost::asio::post(ctx_.get(), [this]() { timer_.cancel(); });
-    }
-
-    void
-    onAmendmentBlock()
-    {
-        state_.get().isAmendmentBlocked = true;
-        startReportingTimer();
-    }
-
-private:
-    void
-    startReportingTimer()
-    {
-        action_();
-
-        timer_.expires_after(interval_);
-        timer_.async_wait([this](auto ec) {
-            if (!ec)
-                boost::asio::post(ctx_.get(), [this] { startReportingTimer(); });
-        });
-    }
-};
-
-}  // namespace etl::impl

src/etl/impl/AmendmentBlockHandler.cpp

@@ -0,0 +1,56 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include "etl/impl/AmendmentBlockHandler.hpp"
+
+#include "etl/SystemState.hpp"
+#include "util/log/Logger.hpp"
+
+#include <boost/asio/io_context.hpp>
+
+#include <chrono>
+#include <functional>
+#include <utility>
+
+namespace etl::impl {
+
+AmendmentBlockHandler::ActionType const AmendmentBlockHandler::defaultAmendmentBlockAction = []() {
+    static util::Logger const log{"ETL"};
+    LOG(log.fatal()) << "Can't process new ledgers: The current ETL source is not compatible with the version of "
+                     << "the libxrpl Clio is currently using. Please upgrade Clio to a newer version.";
+};
+
+AmendmentBlockHandler::AmendmentBlockHandler(
+    boost::asio::io_context& ioc,
+    SystemState& state,
+    std::chrono::steady_clock::duration interval,
+    ActionType action
+)
+    : state_{std::ref(state)}, repeat_{ioc}, interval_{interval}, action_{std::move(action)}
+{
+}
+
+void
+AmendmentBlockHandler::onAmendmentBlock()
+{
+    state_.get().isAmendmentBlocked = true;
+    repeat_.start(interval_, action_);
+}
+
+}  // namespace etl::impl

src/etl/impl/AmendmentBlockHandler.hpp

@@ -0,0 +1,59 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2023, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#pragma once
+
+#include "etl/SystemState.hpp"
+#include "util/Repeat.hpp"
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/post.hpp>
+#include <boost/asio/steady_timer.hpp>
+
+#include <chrono>
+#include <functional>
+
+namespace etl::impl {
+
+class AmendmentBlockHandler {
+public:
+    using ActionType = std::function<void()>;
+
+private:
+    std::reference_wrapper<SystemState> state_;
+    util::Repeat repeat_;
+    std::chrono::steady_clock::duration interval_;
+
+    ActionType action_;
+
+public:
+    static ActionType const defaultAmendmentBlockAction;
+
+    AmendmentBlockHandler(
+        boost::asio::io_context& ioc,
+        SystemState& state,
+        std::chrono::steady_clock::duration interval = std::chrono::seconds{1},
+        ActionType action = defaultAmendmentBlockAction
+    );
+
+    void
+    onAmendmentBlock();
+};
+
+}  // namespace etl::impl

src/etl/impl/AsyncData.hpp

@@ -30,9 +30,9 @@
 #include <grpcpp/grpcpp.h>
 #include <grpcpp/support/status.h>
 #include <org/xrpl/rpc/v1/get_ledger_data.pb.h>
-#include <ripple/basics/base_uint.h>
-#include <ripple/basics/strHex.h>
-#include <ripple/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/strHex.h>
+#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
 
 #include <cstddef>
 #include <cstdint>

src/etl/impl/BaseCursorProvider.hpp

@@ -19,7 +19,7 @@
 
 #pragma once
 
-#include <ripple/basics/base_uint.h>
+#include <xrpl/basics/base_uint.h>
 
 #include <cstdint>
 #include <vector>

src/etl/impl/CacheLoader.hpp

@@ -28,9 +28,9 @@
 
 #include <boost/algorithm/string/predicate.hpp>
 #include <boost/context/detail/config.hpp>
-#include <ripple/basics/Blob.h>
-#include <ripple/basics/base_uint.h>
-#include <ripple/basics/strHex.h>
+#include <xrpl/basics/Blob.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/strHex.h>
 
 #include <algorithm>
 #include <atomic>

src/etl/impl/CursorFromAccountProvider.hpp

@@ -23,7 +23,7 @@
 #include "data/Types.hpp"
 #include "etl/impl/BaseCursorProvider.hpp"
 
-#include <ripple/basics/base_uint.h>
+#include <xrpl/basics/base_uint.h>
 
 #include <algorithm>
 #include <cstddef>

src/etl/impl/CursorFromDiffProvider.hpp

@@ -24,7 +24,7 @@
 #include "etl/impl/BaseCursorProvider.hpp"
 #include "util/Assert.hpp"
 
-#include <ripple/basics/base_uint.h>
+#include <xrpl/basics/base_uint.h>
 
 #include <algorithm>
 #include <cstddef>

src/etl/impl/CursorFromFixDiffNumProvider.hpp

@@ -25,8 +25,8 @@
 
 #include <boost/algorithm/string/predicate.hpp>
 #include <boost/asio/io_context.hpp>
-#include <ripple/basics/base_uint.h>
-#include <ripple/basics/strHex.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/strHex.h>
 
 #include <algorithm>
 #include <cstddef>

src/etl/impl/Extractor.hpp

@@ -25,7 +25,7 @@
 #include "util/Profiler.hpp"
 #include "util/log/Logger.hpp"
 
-#include <ripple/beast/core/CurrentThreadName.h>
+#include <xrpl/beast/core/CurrentThreadName.h>
 
 #include <chrono>
 #include <cstdint>

src/etl/impl/ForwardingSource.cpp

@@ -19,6 +19,7 @@
 
 #include "etl/impl/ForwardingSource.hpp"
 
+#include "rpc/Errors.hpp"
 #include "util/log/Logger.hpp"
 
 #include <boost/asio/spawn.hpp>
@@ -55,7 +56,7 @@ ForwardingSource::ForwardingSource(
         );
 }
 
-std::optional<boost::json::object>
+std::expected<boost::json::object, rpc::ClioError>
 ForwardingSource::forwardToRippled(
     boost::json::object const& request,
     std::optional<std::string> const& forwardToRippledClientIp,
@@ -74,18 +75,26 @@ ForwardingSource::forwardToRippled(
 
     auto expectedConnection = connectionBuilder.connect(yield);
     if (not expectedConnection) {
-        return std::nullopt;
+        LOG(log_.debug()) << "Couldn't connect to rippled to forward request.";
+        return std::unexpected{rpc::ClioError::etlCONNECTION_ERROR};
     }
     auto& connection = expectedConnection.value();
 
     auto writeError = connection->write(boost::json::serialize(request), yield, forwardingTimeout_);
     if (writeError) {
-        return std::nullopt;
+        LOG(log_.debug()) << "Error sending request to rippled to forward request.";
+        return std::unexpected{rpc::ClioError::etlREQUEST_ERROR};
     }
 
     auto response = connection->read(yield, forwardingTimeout_);
     if (not response) {
-        return std::nullopt;
+        if (auto errorCode = response.error().errorCode();
+            errorCode.has_value() and errorCode->value() == boost::system::errc::timed_out) {
+            LOG(log_.debug()) << "Request to rippled timed out";
+            return std::unexpected{rpc::ClioError::etlREQUEST_TIMEOUT};
+        }
+        LOG(log_.debug()) << "Error sending request to rippled to forward request.";
+        return std::unexpected{rpc::ClioError::etlREQUEST_ERROR};
     }
 
     boost::json::value parsedResponse;
@@ -94,8 +103,8 @@ ForwardingSource::forwardToRippled(
         if (not parsedResponse.is_object())
             throw std::runtime_error("response is not an object");
     } catch (std::exception const& e) {
-        LOG(log_.error()) << "Error parsing response from rippled: " << e.what() << ". Response: " << *response;
-        return std::nullopt;
+        LOG(log_.debug()) << "Error parsing response from rippled: " << e.what() << ". Response: " << *response;
+        return std::unexpected{rpc::ClioError::etlINVALID_RESPONSE};
     }
 
     auto responseObject = parsedResponse.as_object();

src/etl/impl/ForwardingSource.hpp

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include "rpc/Errors.hpp"
 #include "util/log/Logger.hpp"
 #include "util/requests/WsConnection.hpp"
 
@@ -26,6 +27,7 @@
 #include <boost/json/object.hpp>
 
 #include <chrono>
+#include <expected>
 #include <optional>
 #include <string>
 #include <string_view>
@@ -54,9 +56,9 @@ public:
      * @param forwardToRippledClientIp IP of the client forwarding this request if known
      * @param xUserValue Optional value for X-User header
      * @param yield The coroutine context
-     * @return Response wrapped in an optional on success; nullopt otherwise
+     * @return Response on success or error on failure
      */
-    std::optional<boost::json::object>
+    std::expected<boost::json::object, rpc::ClioError>
     forwardToRippled(
         boost::json::object const& request,
         std::optional<std::string> const& forwardToRippledClientIp,

src/etl/impl/GrpcSource.cpp

@@ -24,7 +24,7 @@
 #include "util/Assert.hpp"
 #include "util/log/Logger.hpp"
 
-#include <boost/asio/ip/address.hpp>
+#include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <fmt/core.h>
 #include <grpcpp/client_context.h>
@@ -39,6 +39,7 @@
 #include <exception>
 #include <memory>
 #include <sstream>
+#include <stdexcept>
 #include <string>
 #include <utility>
 #include <vector>
@@ -46,12 +47,17 @@
 namespace etl::impl {
 
 GrpcSource::GrpcSource(std::string const& ip, std::string const& grpcPort, std::shared_ptr<BackendInterface> backend)
-    : log_(fmt::format("GrpcSource[{}:{}]", ip, grpcPort)), backend_(std::move(backend))
+    : log_(fmt::format("ETL_Grpc[{}:{}]", ip, grpcPort)), backend_(std::move(backend))
 {
     try {
-        boost::asio::ip::tcp::endpoint const endpoint{boost::asio::ip::make_address(ip), std::stoi(grpcPort)};
+        boost::asio::io_context ctx;
+        boost::asio::ip::tcp::resolver resolver{ctx};
+        auto const resolverResult = resolver.resolve(ip, grpcPort);
+        if (resolverResult.empty()) {
+            throw std::runtime_error("Failed to resolve " + ip + ":" + grpcPort);
+        }
         std::stringstream ss;
-        ss << endpoint;
+        ss << resolverResult.begin()->endpoint();
         grpc::ChannelArguments chArgs;
         chArgs.SetMaxReceiveMessageSize(-1);
         stub_ = org::xrpl::rpc::v1::XRPLedgerAPIService::NewStub(

src/etl/impl/GrpcSource.hpp

@@ -24,7 +24,7 @@
 
 #include <grpcpp/support/status.h>
 #include <org/xrpl/rpc/v1/get_ledger.pb.h>
-#include <ripple/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
+#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
 
 #include <cstdint>
 #include <memory>

src/etl/impl/LedgerFetcher.hpp

@@ -23,7 +23,7 @@
 #include "util/log/Logger.hpp"
 
 #include <grpcpp/grpcpp.h>
-#include <ripple/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
+#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
 
 #include <cstdint>
 #include <memory>

src/etl/impl/LedgerLoader.hpp

@@ -30,13 +30,13 @@
 #include "util/Profiler.hpp"
 #include "util/log/Logger.hpp"
 
-#include <ripple/basics/base_uint.h>
-#include <ripple/basics/strHex.h>
-#include <ripple/beast/core/CurrentThreadName.h>
-#include <ripple/protocol/LedgerHeader.h>
-#include <ripple/protocol/STTx.h>
-#include <ripple/protocol/Serializer.h>
-#include <ripple/protocol/TxMeta.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/strHex.h>
+#include <xrpl/beast/core/CurrentThreadName.h>
+#include <xrpl/protocol/LedgerHeader.h>
+#include <xrpl/protocol/STTx.h>
+#include <xrpl/protocol/Serializer.h>
+#include <xrpl/protocol/TxMeta.h>
 
 #include <chrono>
 #include <cstddef>
@@ -136,20 +136,7 @@ public:
             );
         }
 
-        // Remove all but the last NFTsData for each id. unique removes all but the first of a group, so we want to
-        // reverse sort by transaction index
-        std::sort(result.nfTokensData.begin(), result.nfTokensData.end(), [](NFTsData const& a, NFTsData const& b) {
-            return a.tokenID > b.tokenID && a.transactionIndex > b.transactionIndex;
-        });
-
-        // Now we can unique the NFTs by tokenID.
-        auto last = std::unique(
-            result.nfTokensData.begin(),
-            result.nfTokensData.end(),
-            [](NFTsData const& a, NFTsData const& b) { return a.tokenID == b.tokenID; }
-        );
-        result.nfTokensData.erase(last, result.nfTokensData.end());
-
+        result.nfTokensData = getUniqueNFTsDatas(result.nfTokensData);
         return result;
     }
 

src/etl/impl/LedgerPublisher.hpp

@@ -31,12 +31,12 @@
 
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/strand.hpp>
-#include <ripple/basics/chrono.h>
-#include <ripple/protocol/Fees.h>
-#include <ripple/protocol/LedgerHeader.h>
-#include <ripple/protocol/SField.h>
-#include <ripple/protocol/STObject.h>
-#include <ripple/protocol/Serializer.h>
+#include <xrpl/basics/chrono.h>
+#include <xrpl/protocol/Fees.h>
+#include <xrpl/protocol/LedgerHeader.h>
+#include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STObject.h>
+#include <xrpl/protocol/Serializer.h>
 
 #include <chrono>
 #include <cstddef>
@@ -166,13 +166,16 @@ public:
             LOG(log_.info()) << "Publishing ledger " << std::to_string(lgrInfo.seq);
 
             if (!state_.get().isWriting) {
-                LOG(log_.info()) << "Updating cache";
+                LOG(log_.info()) << "Updating ledger range for read node.";
 
-                std::vector<data::LedgerObject> const diff = data::synchronousAndRetryOnTimeout([&](auto yield) {
-                    return backend_->fetchLedgerDiff(lgrInfo.seq, yield);
-                });
+                if (!cache_.get().isDisabled()) {
+                    std::vector<data::LedgerObject> const diff = data::synchronousAndRetryOnTimeout([&](auto yield) {
+                        return backend_->fetchLedgerDiff(lgrInfo.seq, yield);
+                    });
+
+                    cache_.get().update(diff, lgrInfo.seq);
+                }
 
-                cache_.get().update(diff, lgrInfo.seq);
                 backend_->updateRange(lgrInfo.seq);
             }
 

src/etl/impl/SourceImpl.hpp

@@ -23,6 +23,7 @@
 #include "etl/impl/ForwardingSource.hpp"
 #include "etl/impl/GrpcSource.hpp"
 #include "etl/impl/SubscriptionSource.hpp"
+#include "rpc/Errors.hpp"
 
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
@@ -31,12 +32,14 @@
 
 #include <chrono>
 #include <cstdint>
+#include <expected>
 #include <memory>
 #include <optional>
 #include <string>
 #include <string_view>
 #include <utility>
 #include <vector>
+
 namespace etl::impl {
 
 /**
@@ -205,9 +208,9 @@ public:
      * @param forwardToRippledClientIp IP of the client forwarding this request if known
      * @param xUserValue Optional value of the X-User header
      * @param yield The coroutine context
-     * @return Response wrapped in an optional on success; nullopt otherwise
+     * @return Response or ClioError
      */
-    std::optional<boost::json::object>
+    std::expected<boost::json::object, rpc::ClioError>
     forwardToRippled(
         boost::json::object const& request,
         std::optional<std::string> const& forwardToRippledClientIp,

src/etl/impl/SubscriptionSource.cpp

@@ -24,8 +24,6 @@
 #include "rpc/JS.hpp"
 #include "util/Retry.hpp"
 #include "util/log/Logger.hpp"
-#include "util/prometheus/Label.hpp"
-#include "util/prometheus/Prometheus.hpp"
 #include "util/requests/Types.hpp"
 
 #include <boost/algorithm/string/classification.hpp>
@@ -42,7 +40,7 @@
 #include <boost/json/serialize.hpp>
 #include <boost/json/value_to.hpp>
 #include <fmt/core.h>
-#include <ripple/protocol/jss.h>
+#include <xrpl/protocol/jss.h>
 
 #include <algorithm>
 #include <chrono>
@@ -68,28 +66,22 @@ SubscriptionSource::SubscriptionSource(
     OnConnectHook onConnect,
     OnDisconnectHook onDisconnect,
     OnLedgerClosedHook onLedgerClosed,
-    std::chrono::steady_clock::duration const wsTimeout,
+    std::chrono::steady_clock::duration const connectionTimeout,
     std::chrono::steady_clock::duration const retryDelay
 )
-    : log_(fmt::format("SubscriptionSource[{}:{}]", ip, wsPort))
+    : log_(fmt::format("GrpcSource[{}:{}]", ip, wsPort))
     , wsConnectionBuilder_(ip, wsPort)
     , validatedLedgers_(std::move(validatedLedgers))
     , subscriptions_(std::move(subscriptions))
     , strand_(boost::asio::make_strand(ioContext))
-    , wsTimeout_(wsTimeout)
     , retry_(util::makeRetryExponentialBackoff(retryDelay, RETRY_MAX_DELAY, strand_))
     , onConnect_(std::move(onConnect))
     , onDisconnect_(std::move(onDisconnect))
     , onLedgerClosed_(std::move(onLedgerClosed))
-    , lastMessageTimeSecondsSinceEpoch_(PrometheusService::gaugeInt(
-          "subscription_source_last_message_time",
-          util::prometheus::Labels({{"source", fmt::format("{}:{}", ip, wsPort)}}),
-          "Seconds since epoch of the last message received from rippled subscription streams"
-      ))
 {
     wsConnectionBuilder_.addHeader({boost::beast::http::field::user_agent, "clio-client"})
         .addHeader({"X-User", "clio-client"})
-        .setConnectionTimeout(wsTimeout_);
+        .setConnectionTimeout(connectionTimeout);
 }
 
 SubscriptionSource::~SubscriptionSource()
@@ -141,7 +133,6 @@ void
 SubscriptionSource::setForwarding(bool isForwarding)
 {
     isForwarding_ = isForwarding;
-    LOG(log_.info()) << "Forwarding set to " << isForwarding_;
 }
 
 std::chrono::steady_clock::time_point
@@ -175,22 +166,20 @@ SubscriptionSource::subscribe()
             }
 
             wsConnection_ = std::move(connection).value();
+            isConnected_ = true;
+            onConnect_();
 
             auto const& subscribeCommand = getSubscribeCommandJson();
-            auto const writeErrorOpt = wsConnection_->write(subscribeCommand, yield, wsTimeout_);
+            auto const writeErrorOpt = wsConnection_->write(subscribeCommand, yield);
             if (writeErrorOpt) {
                 handleError(writeErrorOpt.value(), yield);
                 return;
             }
 
-            isConnected_ = true;
-            LOG(log_.info()) << "Connected";
-            onConnect_();
-
             retry_.reset();
 
             while (!stop_) {
-                auto const message = wsConnection_->read(yield, wsTimeout_);
+                auto const message = wsConnection_->read(yield);
                 if (not message) {
                     handleError(message.error(), yield);
                     return;
@@ -235,11 +224,10 @@ SubscriptionSource::handleMessage(std::string const& message)
                 auto validatedLedgers = boost::json::value_to<std::string>(result.at(JS(validated_ledgers)));
                 setValidatedRange(std::move(validatedLedgers));
             }
-            LOG(log_.debug()) << "Received a message on ledger subscription stream. Message: " << object;
+            LOG(log_.info()) << "Received a message on ledger subscription stream. Message : " << object;
 
         } else if (object.contains(JS(type)) && object.at(JS(type)) == JS_LedgerClosed) {
-            LOG(log_.debug()) << "Received a message of type 'ledgerClosed' on ledger subscription stream. Message: "
-                              << object;
+            LOG(log_.info()) << "Received a message on ledger subscription stream. Message : " << object;
             if (object.contains(JS(ledger_index))) {
                 ledgerIndex = object.at(JS(ledger_index)).as_int64();
             }
@@ -257,13 +245,10 @@ SubscriptionSource::handleMessage(std::string const& message)
                 // 2 - Validated transaction
                 // Only forward proposed transaction, validated transactions are sent by Clio itself
                 if (object.contains(JS(transaction)) and !object.contains(JS(meta))) {
-                    LOG(log_.debug()) << "Forwarding proposed transaction: " << object;
                     subscriptions_->forwardProposedTransaction(object);
                 } else if (object.contains(JS(type)) && object.at(JS(type)) == JS_ValidationReceived) {
-                    LOG(log_.debug()) << "Forwarding validation: " << object;
                     subscriptions_->forwardValidation(object);
                 } else if (object.contains(JS(type)) && object.at(JS(type)) == JS_ManifestReceived) {
-                    LOG(log_.debug()) << "Forwarding manifest: " << object;
                     subscriptions_->forwardManifest(object);
                 }
             }
@@ -276,7 +261,7 @@ SubscriptionSource::handleMessage(std::string const& message)
 
         return std::nullopt;
     } catch (std::exception const& e) {
-        LOG(log_.error()) << "Exception in handleMessage: " << e.what();
+        LOG(log_.error()) << "Exception in handleMessage : " << e.what();
         return util::requests::RequestError{fmt::format("Error handling message: {}", e.what())};
     }
 }
@@ -285,14 +270,16 @@ void
 SubscriptionSource::handleError(util::requests::RequestError const& error, boost::asio::yield_context yield)
 {
     isConnected_ = false;
-    bool const wasForwarding = isForwarding_.exchange(false);
+    isForwarding_ = false;
     if (not stop_) {
-        LOG(log_.info()) << "Disconnected";
-        onDisconnect_(wasForwarding);
+        onDisconnect_();
     }
 
     if (wsConnection_ != nullptr) {
-        wsConnection_->close(yield);
+        auto const err = wsConnection_->close(yield);
+        if (err) {
+            LOG(log_.error()) << "Error closing websocket connection: " << err->message();
+        }
         wsConnection_.reset();
     }
 
@@ -319,11 +306,7 @@ SubscriptionSource::logError(util::requests::RequestError const& error) const
 void
 SubscriptionSource::setLastMessageTime()
 {
-    lastMessageTimeSecondsSinceEpoch_.get().set(
-        std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch()).count()
-    );
-    auto lock = lastMessageTime_.lock();
-    lock.get() = std::chrono::steady_clock::now();
+    lastMessageTime_.lock().get() = std::chrono::steady_clock::now();
 }
 
 void

src/etl/impl/SubscriptionSource.hpp

@@ -25,7 +25,6 @@
 #include "util/Mutex.hpp"
 #include "util/Retry.hpp"
 #include "util/log/Logger.hpp"
-#include "util/prometheus/Gauge.hpp"
 #include "util/requests/Types.hpp"
 #include "util/requests/WsConnection.hpp"
 
@@ -38,7 +37,6 @@
 #include <atomic>
 #include <chrono>
 #include <cstdint>
-#include <functional>
 #include <future>
 #include <memory>
 #include <optional>
@@ -73,8 +71,6 @@ private:
 
     boost::asio::strand<boost::asio::io_context::executor_type> strand_;
 
-    std::chrono::steady_clock::duration wsTimeout_;
-
     util::Retry retry_;
 
     OnConnectHook onConnect_;
@@ -87,11 +83,9 @@ private:
 
     util::Mutex<std::chrono::steady_clock::time_point> lastMessageTime_;
 
-    std::reference_wrapper<util::prometheus::GaugeInt> lastMessageTimeSecondsSinceEpoch_;
-
     std::future<void> runFuture_;
 
-    static constexpr std::chrono::seconds WS_TIMEOUT{30};
+    static constexpr std::chrono::seconds CONNECTION_TIMEOUT{30};
     static constexpr std::chrono::seconds RETRY_MAX_DELAY{30};
     static constexpr std::chrono::seconds RETRY_DELAY{1};
 
@@ -105,11 +99,11 @@ public:
      * @param wsPort The port of the source
      * @param validatedLedgers The network validated ledgers object
      * @param subscriptions The subscription manager object
+     * @param onConnect The onConnect hook. Called when the connection is established
      * @param onDisconnect The onDisconnect hook. Called when the connection is lost
-     * @param onNewLedger The onNewLedger hook. Called when a new ledger is received
      * @param onLedgerClosed The onLedgerClosed hook. Called when the ledger is closed but only if the source is
      * forwarding
-     * @param wsTimeout A timeout for websocket operations. Defaults to 30 seconds
+     * @param connectionTimeout The connection timeout. Defaults to 30 seconds
      * @param retryDelay The retry delay. Defaults to 1 second
      */
     SubscriptionSource(
@@ -121,7 +115,7 @@ public:
         OnConnectHook onConnect,
         OnDisconnectHook onDisconnect,
         OnLedgerClosedHook onLedgerClosed,
-        std::chrono::steady_clock::duration const wsTimeout = WS_TIMEOUT,
+        std::chrono::steady_clock::duration const connectionTimeout = CONNECTION_TIMEOUT,
         std::chrono::steady_clock::duration const retryDelay = RETRY_DELAY
     );
 

src/etl/impl/Transformer.hpp

@@ -23,7 +23,7 @@
 #include "data/DBHelpers.hpp"
 #include "data/Types.hpp"
 #include "etl/SystemState.hpp"
-#include "etl/impl/AmendmentBlock.hpp"
+#include "etl/impl/AmendmentBlockHandler.hpp"
 #include "etl/impl/LedgerLoader.hpp"
 #include "util/Assert.hpp"
 #include "util/LedgerUtils.hpp"
@@ -31,11 +31,11 @@
 #include "util/log/Logger.hpp"
 
 #include <grpcpp/grpcpp.h>
-#include <ripple/basics/base_uint.h>
-#include <ripple/basics/strHex.h>
-#include <ripple/beast/core/CurrentThreadName.h>
-#include <ripple/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/strHex.h>
+#include <xrpl/beast/core/CurrentThreadName.h>
+#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <chrono>
 #include <cstdint>

src/feed/SubscriptionManager.cpp

@@ -24,10 +24,10 @@
 
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/Book.h>
-#include <ripple/protocol/Fees.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/Fees.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <cstdint>
 #include <string>

src/feed/SubscriptionManager.hpp

@@ -28,22 +28,22 @@
 #include "feed/impl/LedgerFeed.hpp"
 #include "feed/impl/ProposedTransactionFeed.hpp"
 #include "feed/impl/TransactionFeed.hpp"
+#include "util/async/AnyExecutionContext.hpp"
+#include "util/async/context/BasicExecutionContext.hpp"
 #include "util/log/Logger.hpp"
 
 #include <boost/asio/executor_work_guard.hpp>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/Book.h>
-#include <ripple/protocol/Fees.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/Fees.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <cstdint>
-#include <functional>
 #include <memory>
 #include <string>
-#include <thread>
 #include <vector>
 
 /**
@@ -57,9 +57,8 @@ namespace feed {
  * @brief A subscription manager is responsible for managing the subscriptions and publishing the feeds
  */
 class SubscriptionManager : public SubscriptionManagerInterface {
-    std::reference_wrapper<boost::asio::io_context> ioContext_;
     std::shared_ptr<data::BackendInterface const> backend_;
-
+    util::async::AnyExecutionContext ctx_;
     impl::ForwardFeed manifestFeed_;
     impl::ForwardFeed validationsFeed_;
     impl::LedgerFeed ledgerFeed_;
@@ -71,24 +70,31 @@ public:
     /**
      * @brief Construct a new Subscription Manager object
      *
-     * @param ioContext The io context to use
+     * @param executor The executor to use to publish the feeds
      * @param backend The backend to use
      */
-    SubscriptionManager(
-        boost::asio::io_context& ioContext,
-        std::shared_ptr<data::BackendInterface const> const& backend
-    )
-        : ioContext_(ioContext)
-        , backend_(backend)
-        , manifestFeed_(ioContext, "manifest")
-        , validationsFeed_(ioContext, "validations")
-        , ledgerFeed_(ioContext)
-        , bookChangesFeed_(ioContext)
-        , transactionFeed_(ioContext)
-        , proposedTransactionFeed_(ioContext)
+    template <class ExecutorCtx>
+    SubscriptionManager(ExecutorCtx& executor, std::shared_ptr<data::BackendInterface const> const& backend)
+        : backend_(backend)
+        , ctx_(executor)
+        , manifestFeed_(ctx_, "manifest")
+        , validationsFeed_(ctx_, "validations")
+        , ledgerFeed_(ctx_)
+        , bookChangesFeed_(ctx_)
+        , transactionFeed_(ctx_)
+        , proposedTransactionFeed_(ctx_)
     {
     }
 
+    /**
+     * @brief Destructor of the SubscriptionManager object. It will block until all running jobs finished.
+     */
+    ~SubscriptionManager() override
+    {
+        ctx_.stop();
+        ctx_.join();
+    }
+
     /**
      * @brief Subscribe to the book changes feed.
      * @param subscriber
@@ -286,16 +292,15 @@ public:
 };
 
 /**
- * @brief The help class to run the subscription manager. The container of io_context which is used to publish the
- * feeds.
+ * @brief The help class to run the subscription manager. The container of PoolExecutionContext which is used to publish
+ * the feeds.
  */
 class SubscriptionManagerRunner {
-    boost::asio::io_context ioContext_;
+    std::uint64_t workersNum_;
+    using ActualExecutionCtx = util::async::PoolExecutionContext;
+    ActualExecutionCtx ctx_;
     std::shared_ptr<SubscriptionManager> subscriptionManager_;
     util::Logger logger_{"Subscriptions"};
-    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work_ =
-        boost::asio::make_work_guard(ioContext_);
-    std::vector<std::thread> workers_;
 
 public:
     /**
@@ -305,13 +310,11 @@ public:
      * @param backend The backend to use
      */
     SubscriptionManagerRunner(util::Config const& config, std::shared_ptr<data::BackendInterface> const& backend)
-        : subscriptionManager_(std::make_shared<SubscriptionManager>(ioContext_, backend))
+        : workersNum_(config.valueOr<std::uint64_t>("subscription_workers", 1))
+        , ctx_(workersNum_)
+        , subscriptionManager_(std::make_shared<SubscriptionManager>(ctx_, backend))
     {
-        auto numThreads = config.valueOr<uint64_t>("subscription_workers", 1);
-        LOG(logger_.info()) << "Starting subscription manager with " << numThreads << " workers";
-        workers_.reserve(numThreads);
-        for (auto i = numThreads; i > 0; --i)
-            workers_.emplace_back([&] { ioContext_.run(); });
+        LOG(logger_.info()) << "Starting subscription manager with " << workersNum_ << " workers";
     }
 
     /**
@@ -324,12 +327,5 @@ public:
     {
         return subscriptionManager_;
     }
-
-    ~SubscriptionManagerRunner()
-    {
-        work_.reset();
-        for (auto& worker : workers_)
-            worker.join();
-    }
 };
 }  // namespace feed

src/feed/SubscriptionManagerInterface.hpp

@@ -26,10 +26,10 @@
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/Book.h>
-#include <ripple/protocol/Fees.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/Fees.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <cstdint>
 #include <string>
@@ -109,6 +109,7 @@ public:
      * @brief Subscribe to the ledger feed.
      * @param yield The coroutine context
      * @param subscriber The subscriber to the ledger feed
+     *
      * @return The ledger feed
      */
     virtual boost::json::object

src/feed/impl/BookChangesFeed.hpp

@@ -22,10 +22,11 @@
 #include "data/Types.hpp"
 #include "feed/impl/SingleFeedBase.hpp"
 #include "rpc/BookChangesHelper.hpp"
+#include "util/async/AnyExecutionContext.hpp"
 
 #include <boost/asio/io_context.hpp>
 #include <boost/json/serialize.hpp>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <vector>
 
@@ -37,7 +38,7 @@ namespace feed::impl {
  * '0A5010342D8AAFABDCA58A68F6F588E1C6E58C21B63ED6CA8DB2478F58F3ECD5', 'ledger_time': 756395682, 'changes': []}
  */
 struct BookChangesFeed : public SingleFeedBase {
-    BookChangesFeed(boost::asio::io_context& ioContext) : SingleFeedBase(ioContext, "book_changes")
+    BookChangesFeed(util::async::AnyExecutionContext& executionCtx) : SingleFeedBase(executionCtx, "book_changes")
     {
     }
 

src/feed/impl/LedgerFeed.cpp

@@ -28,9 +28,9 @@
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
 #include <boost/json/serialize.hpp>
-#include <ripple/basics/base_uint.h>
-#include <ripple/protocol/Fees.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/Fees.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <cstdint>
 #include <memory>

src/feed/impl/LedgerFeed.hpp

@@ -22,13 +22,14 @@
 #include "data/BackendInterface.hpp"
 #include "feed/Types.hpp"
 #include "feed/impl/SingleFeedBase.hpp"
+#include "util/async/AnyExecutionContext.hpp"
 
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
 #include <boost/json/serialize.hpp>
-#include <ripple/protocol/Fees.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/protocol/Fees.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <cstdint>
 #include <memory>
@@ -46,9 +47,9 @@ class LedgerFeed : public SingleFeedBase {
 public:
     /**
      * @brief Construct a new Ledger Feed object
-     * @param ioContext The actual publish will be called in the strand of this.
+     * @param executionCtx The actual publish will be called in the strand of this.
      */
-    LedgerFeed(boost::asio::io_context& ioContext) : SingleFeedBase(ioContext, "ledger")
+    LedgerFeed(util::async::AnyExecutionContext& executionCtx) : SingleFeedBase(executionCtx, "ledger")
     {
     }
 

src/feed/impl/ProposedTransactionFeed.cpp

@@ -23,10 +23,9 @@
 #include "rpc/RPCHelpers.hpp"
 #include "util/log/Logger.hpp"
 
-#include <boost/asio/post.hpp>
 #include <boost/json/object.hpp>
 #include <boost/json/serialize.hpp>
-#include <ripple/protocol/AccountID.h>
+#include <xrpl/protocol/AccountID.h>
 
 #include <cstdint>
 #include <memory>
@@ -101,21 +100,18 @@ ProposedTransactionFeed::pub(boost::json::object const& receivedTxJson)
     auto const accounts = rpc::getAccountsFromTransaction(transaction);
     auto affectedAccounts = std::unordered_set<ripple::AccountID>(accounts.cbegin(), accounts.cend());
 
-    boost::asio::post(strand_, [this, pubMsg = std::move(pubMsg), affectedAccounts = std::move(affectedAccounts)]() {
-        notified_.clear();
-        signal_.emit(pubMsg);
-
-        // Prevent the same connection from receiving the same message twice if it is subscribed to multiple accounts
-        // However, if the same connection subscribe both stream and account, it will still receive the message twice.
-        // notified_ can be cleared before signal_ emit to improve this, but let's keep it as is for now, since rippled
-        // acts like this.
-        notified_.clear();
-
-        for (auto const& account : affectedAccounts)
-            accountSignal_.emit(account, pubMsg);
-
-        ++pubCount_.get();
-    });
+    [[maybe_unused]] auto task =
+        strand_.execute([this, pubMsg = std::move(pubMsg), affectedAccounts = std::move(affectedAccounts)]() {
+            notified_.clear();
+            signal_.emit(pubMsg);
+            // Prevent the same connection from receiving the same message twice if it is subscribed to multiple
+            // accounts However, if the same connection subscribe both stream and account, it will still receive the
+            // message twice. notified_ can be cleared before signal_ emit to improve this, but let's keep it as is for
+            // now, since rippled acts like this.
+            notified_.clear();
+            for (auto const& account : affectedAccounts)
+                accountSignal_.emit(account, pubMsg);
+        });
 }
 
 std::uint64_t

src/feed/impl/ProposedTransactionFeed.hpp

@@ -23,17 +23,18 @@
 #include "feed/impl/TrackableSignal.hpp"
 #include "feed/impl/TrackableSignalMap.hpp"
 #include "feed/impl/Util.hpp"
+#include "util/async/AnyExecutionContext.hpp"
+#include "util/async/AnyStrand.hpp"
 #include "util/log/Logger.hpp"
-#include "util/prometheus/Counter.hpp"
 #include "util/prometheus/Gauge.hpp"
 
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/strand.hpp>
 #include <boost/json/object.hpp>
 #include <fmt/core.h>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/Book.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <cstdint>
 #include <functional>
@@ -52,10 +53,9 @@ class ProposedTransactionFeed {
 
     std::unordered_set<SubscriberPtr>
         notified_;  // Used by slots to prevent double notifications if tx contains multiple subscribed accounts
-    boost::asio::strand<boost::asio::io_context::executor_type> strand_;
+    util::async::AnyStrand strand_;
     std::reference_wrapper<util::prometheus::GaugeInt> subAllCount_;
     std::reference_wrapper<util::prometheus::GaugeInt> subAccountCount_;
-    std::reference_wrapper<util::prometheus::CounterInt> pubCount_;
 
     TrackableSignalMap<ripple::AccountID, Subscriber, std::shared_ptr<std::string>> accountSignal_;
     TrackableSignal<Subscriber, std::shared_ptr<std::string>> signal_;
@@ -63,13 +63,13 @@ class ProposedTransactionFeed {
 public:
     /**
      * @brief Construct a Proposed Transaction Feed object.
-     * @param ioContext The actual publish will be called in the strand of this.
+     * @param executionCtx The actual publish will be called in the strand of this.
      */
-    ProposedTransactionFeed(boost::asio::io_context& ioContext)
-        : strand_(boost::asio::make_strand(ioContext))
+    ProposedTransactionFeed(util::async::AnyExecutionContext& executionCtx)
+        : strand_(executionCtx.makeStrand())
         , subAllCount_(getSubscriptionsGaugeInt("tx_proposed"))
         , subAccountCount_(getSubscriptionsGaugeInt("account_proposed"))
-        , pubCount_(getPublishedMessagesCounterInt("tx_proposed"))
+
     {
     }
 

src/feed/impl/SingleFeedBase.cpp

@@ -22,12 +22,9 @@
 #include "feed/Types.hpp"
 #include "feed/impl/TrackableSignal.hpp"
 #include "feed/impl/Util.hpp"
+#include "util/async/AnyExecutionContext.hpp"
 #include "util/log/Logger.hpp"
 
-#include <boost/asio/io_context.hpp>
-#include <boost/asio/post.hpp>
-#include <boost/asio/strand.hpp>
-
 #include <cstdint>
 #include <memory>
 #include <string>
@@ -35,11 +32,8 @@
 
 namespace feed::impl {
 
-SingleFeedBase::SingleFeedBase(boost::asio::io_context& ioContext, std::string const& name)
-    : strand_(boost::asio::make_strand(ioContext))
-    , subCount_(getSubscriptionsGaugeInt(name))
-    , pubCount_(getPublishedMessagesCounterInt(name))
-    , name_(name)
+SingleFeedBase::SingleFeedBase(util::async::AnyExecutionContext& executionCtx, std::string const& name)
+    : strand_(executionCtx.makeStrand()), subCount_(getSubscriptionsGaugeInt(name)), name_(name)
 {
 }
 
@@ -70,10 +64,9 @@ SingleFeedBase::unsub(SubscriberSharedPtr const& subscriber)
 void
 SingleFeedBase::pub(std::string msg) const
 {
-    boost::asio::post(strand_, [this, msg = std::move(msg)]() mutable {
-        auto const msgPtr = std::make_shared<std::string>(std::move(msg));
+    [[maybe_unused]] auto task = strand_.execute([this, msg = std::move(msg)]() {
+        auto const msgPtr = std::make_shared<std::string>(msg);
         signal_.emit(msgPtr);
-        ++pubCount_.get();
     });
 }
 

src/feed/impl/SingleFeedBase.hpp

@@ -21,8 +21,9 @@
 
 #include "feed/Types.hpp"
 #include "feed/impl/TrackableSignal.hpp"
+#include "util/async/AnyExecutionContext.hpp"
+#include "util/async/AnyStrand.hpp"
 #include "util/log/Logger.hpp"
-#include "util/prometheus/Counter.hpp"
 #include "util/prometheus/Gauge.hpp"
 
 #include <boost/asio/io_context.hpp>
@@ -39,9 +40,8 @@ namespace feed::impl {
  * @brief Base class for single feed.
  */
 class SingleFeedBase {
-    boost::asio::strand<boost::asio::io_context::executor_type> strand_;
+    util::async::AnyStrand strand_;
     std::reference_wrapper<util::prometheus::GaugeInt> subCount_;
-    std::reference_wrapper<util::prometheus::CounterInt> pubCount_;
     TrackableSignal<Subscriber, std::shared_ptr<std::string> const&> signal_;
     util::Logger logger_{"Subscriptions"};
     std::string name_;
@@ -49,10 +49,10 @@ class SingleFeedBase {
 public:
     /**
      * @brief Construct a new Single Feed Base object
-     * @param ioContext The actual publish will be called in the strand of this.
+     * @param executionCtx The actual publish will be called in the strand of this.
      * @param name The promethues counter name of the feed.
      */
-    SingleFeedBase(boost::asio::io_context& ioContext, std::string const& name);
+    SingleFeedBase(util::async::AnyExecutionContext& executionCtx, std::string const& name);
 
     /**
      * @brief Subscribe the feed.

src/feed/impl/TransactionFeed.cpp

@@ -26,21 +26,20 @@
 #include "rpc/RPCHelpers.hpp"
 #include "util/log/Logger.hpp"
 
-#include <boost/asio/post.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
 #include <boost/json/serialize.hpp>
-#include <ripple/basics/chrono.h>
-#include <ripple/basics/strHex.h>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/Book.h>
-#include <ripple/protocol/LedgerFormats.h>
-#include <ripple/protocol/LedgerHeader.h>
-#include <ripple/protocol/SField.h>
-#include <ripple/protocol/STObject.h>
-#include <ripple/protocol/TER.h>
-#include <ripple/protocol/TxFormats.h>
-#include <ripple/protocol/jss.h>
+#include <xrpl/basics/chrono.h>
+#include <xrpl/basics/strHex.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/LedgerFormats.h>
+#include <xrpl/protocol/LedgerHeader.h>
+#include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STObject.h>
+#include <xrpl/protocol/TER.h>
+#include <xrpl/protocol/TxFormats.h>
+#include <xrpl/protocol/jss.h>
 
 #include <cstdint>
 #include <memory>
@@ -276,39 +275,30 @@ TransactionFeed::pub(
         }
     }
 
-    boost::asio::post(
-        strand_,
-        [this,
-         allVersionsMsgs = std::move(allVersionsMsgs),
-         affectedAccounts = std::move(affectedAccounts),
-         affectedBooks = std::move(affectedBooks)]() {
-            notified_.clear();
-            signal_.emit(allVersionsMsgs);
-
-            // clear the notified set. If the same connection subscribes both transactions + proposed_transactions,
-            // rippled SENDS the same message twice
-            notified_.clear();
-            txProposedsignal_.emit(allVersionsMsgs);
-            notified_.clear();
-
-            // check duplicate for account and proposed_account, this prevents sending the same message multiple times
-            // if it affects multiple accounts watched by the same connection
-            for (auto const& account : affectedAccounts) {
-                accountSignal_.emit(account, allVersionsMsgs);
-                accountProposedSignal_.emit(account, allVersionsMsgs);
-            }
-
-            notified_.clear();
-
-            // check duplicate for books, this prevents sending the same message multiple times if it affects multiple
-            // books watched by the same connection
-            for (auto const& book : affectedBooks) {
-                bookSignal_.emit(book, allVersionsMsgs);
-            }
-
-            ++pubCount_.get();
+    [[maybe_unused]] auto task = strand_.execute([this,
+                                                  allVersionsMsgs = std::move(allVersionsMsgs),
+                                                  affectedAccounts = std::move(affectedAccounts),
+                                                  affectedBooks = std::move(affectedBooks)]() {
+        notified_.clear();
+        signal_.emit(allVersionsMsgs);
+        // clear the notified set. If the same connection subscribes both transactions + proposed_transactions,
+        // rippled SENDS the same message twice
+        notified_.clear();
+        txProposedsignal_.emit(allVersionsMsgs);
+        notified_.clear();
+        // check duplicate for account and proposed_account, this prevents sending the same message multiple times
+        // if it affects multiple accounts watched by the same connection
+        for (auto const& account : affectedAccounts) {
+            accountSignal_.emit(account, allVersionsMsgs);
+            accountProposedSignal_.emit(account, allVersionsMsgs);
         }
-    );
+        notified_.clear();
+        // check duplicate for books, this prevents sending the same message multiple times if it affects multiple
+        // books watched by the same connection
+        for (auto const& book : affectedBooks) {
+            bookSignal_.emit(book, allVersionsMsgs);
+        }
+    });
 }
 
 void

src/feed/impl/TransactionFeed.hpp

@@ -25,16 +25,17 @@
 #include "feed/impl/TrackableSignal.hpp"
 #include "feed/impl/TrackableSignalMap.hpp"
 #include "feed/impl/Util.hpp"
+#include "util/async/AnyExecutionContext.hpp"
+#include "util/async/AnyStrand.hpp"
 #include "util/log/Logger.hpp"
-#include "util/prometheus/Counter.hpp"
 #include "util/prometheus/Gauge.hpp"
 
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/strand.hpp>
 #include <fmt/core.h>
-#include <ripple/protocol/AccountID.h>
-#include <ripple/protocol/Book.h>
-#include <ripple/protocol/LedgerHeader.h>
+#include <xrpl/protocol/AccountID.h>
+#include <xrpl/protocol/Book.h>
+#include <xrpl/protocol/LedgerHeader.h>
 
 #include <array>
 #include <cstdint>
@@ -64,11 +65,10 @@ class TransactionFeed {
 
     util::Logger logger_{"Subscriptions"};
 
-    boost::asio::strand<boost::asio::io_context::executor_type> strand_;
+    util::async::AnyStrand strand_;
     std::reference_wrapper<util::prometheus::GaugeInt> subAllCount_;
     std::reference_wrapper<util::prometheus::GaugeInt> subAccountCount_;
     std::reference_wrapper<util::prometheus::GaugeInt> subBookCount_;
-    std::reference_wrapper<util::prometheus::CounterInt> pubCount_;
 
     TrackableSignalMap<ripple::AccountID, Subscriber, AllVersionTransactionsType const&> accountSignal_;
     TrackableSignalMap<ripple::Book, Subscriber, AllVersionTransactionsType const&> bookSignal_;
@@ -84,14 +84,13 @@ class TransactionFeed {
 public:
     /**
      * @brief Construct a new Transaction Feed object.
-     * @param ioContext The actual publish will be called in the strand of this.
+     * @param executionCtx The actual publish will be called in the strand of this.
      */
-    TransactionFeed(boost::asio::io_context& ioContext)
-        : strand_(boost::asio::make_strand(ioContext))
+    TransactionFeed(util::async::AnyExecutionContext& executionCtx)
+        : strand_(executionCtx.makeStrand())
         , subAllCount_(getSubscriptionsGaugeInt("tx"))
         , subAccountCount_(getSubscriptionsGaugeInt("account"))
         , subBookCount_(getSubscriptionsGaugeInt("book"))
-        , pubCount_(getPublishedMessagesCounterInt("tx"))
     {
     }