clang-tidy

Simulation results revealed that by requiring a minimum number of peers to send a message, valdiations will stop propagating, as the servers that are directly connected to peers will receive a unique message from only a single peer, the validator. Therefore, they will fail to select a slot.
Furthermore, by increasing untrusted slots to 30 we guarantee that some validators will propagate through most of the network.

.cspell.config.yaml

@@ -216,7 +216,6 @@ words:
   - Nyffenegger
   - onlatest
   - ostr
-  - oxalica
   - pargs
   - partitioner
   - paychan
@@ -341,7 +340,6 @@ words:
   - unsquelch
   - unsquelched
   - unsquelching
-  - unsuffixed
   - unvalidated
   - unveto
   - unvetoed

.github/pull_request_template.md

@@ -1,10 +1,10 @@
 <!--
-This PR template helps you write a good pull request description.
+This PR template helps you to write a good pull request description.
 Please feel free to include additional useful information even beyond what is requested below.
 
 If your branch is on a personal fork and has a name that allows it to
 run CI build/test jobs (e.g. "ci/foo"), remember to rename it BEFORE
-opening the PR. This avoids redundant test runs. Renaming
+opening the PR.  This avoids unnecessary redundant test runs. Renaming
 the branch after opening the PR will close the PR.
 https://docs.github.com/en/repositories/configuring-branches-and-merges-in-your-repository/managing-branches-in-your-repository/renaming-a-branch
 -->
@@ -15,7 +15,7 @@ https://docs.github.com/en/repositories/configuring-branches-and-merges-in-your-
 Please include a summary of the changes.
 This may be a direct input to the release notes.
 If too broad, please consider splitting into multiple PRs.
-If there is a relevant task or issue, please link it here.
+If a relevant task or issue, please link it here.
 -->
 
 ### Context of Change
@@ -65,5 +65,5 @@ This section may not be needed if your change includes thoroughly commented unit
 
 <!--
 ## Future Tasks
-For future tasks related to this PR.
+For future tasks related to PR.
 -->

.github/scripts/strategy-matrix/linux.json

@@ -1,5 +1,5 @@
 {
-  "image_tag": "sha-3122de8",
+  "image_tag": "sha-2e25435",
   "configs": {
     "ubuntu": [
       {

.github/workflows/build-nix-images.yml

@@ -8,7 +8,6 @@ on:
       - ".github/workflows/build-nix-images.yml"
       - "flake.nix"
       - "flake.lock"
-      - "rust-toolchain.toml"
       - "nix/**"
       - "!nix/docker/README.md"
       - "!nix/devshell.nix"
@@ -19,7 +18,6 @@ on:
       - ".github/workflows/build-nix-images.yml"
       - "flake.nix"
       - "flake.lock"
-      - "rust-toolchain.toml"
       - "nix/**"
       - "!nix/docker/README.md"
       - "!nix/devshell.nix"
@@ -56,7 +54,7 @@ jobs:
             base_image: debian:bookworm
           - name: rhel
             base_image: registry.access.redhat.com/ubi9/ubi:latest
-    uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@9e7e4e80af9e684c116b38369add8eea64451f32
+    uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@ee03d31bcc4501d7599dc1b1ecd7a34af582ad1c
     with:
       image_name: xrpld/nix-${{ matrix.distro.name }}
       dockerfile: nix/docker/Dockerfile

.github/workflows/build-packaging-images.yml

@@ -38,7 +38,7 @@ jobs:
             base_image: debian:bookworm
           - name: rhel
             base_image: registry.access.redhat.com/ubi9/ubi:latest
-    uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@9e7e4e80af9e684c116b38369add8eea64451f32
+    uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@ee03d31bcc4501d7599dc1b1ecd7a34af582ad1c
     with:
       image_name: xrpld/packaging-${{ matrix.distro.name }}
       dockerfile: package/Dockerfile

.github/workflows/build-pre-commit-image.yml

@@ -30,7 +30,7 @@ jobs:
     permissions:
       contents: read
       packages: write
-    uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@9e7e4e80af9e684c116b38369add8eea64451f32
+    uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@ee03d31bcc4501d7599dc1b1ecd7a34af582ad1c
     with:
       image_name: xrpld/pre-commit
       dockerfile: bin/pre-commit/Dockerfile

.github/workflows/check-pr-title.yml

@@ -20,4 +20,4 @@ on:
 jobs:
   check_title:
     if: ${{ github.event.pull_request.draft != true }}
-    uses: XRPLF/actions/.github/workflows/check-pr-title.yml@d7c65e49225a38f6d8010eacf017bb5a98d7476c
+    uses: XRPLF/actions/.github/workflows/check-pr-title.yml@cba1f0891650baf1a9c88624dc2d72573be2eb81

.github/workflows/pre-commit.yml

@@ -14,7 +14,7 @@ on:
 jobs:
   # Call the workflow in the XRPLF/actions repo that runs the pre-commit hooks.
   run-hooks:
-    uses: XRPLF/actions/.github/workflows/pre-commit.yml@3ba08d6ddf114092891d48491fc2e26c3ba15552
+    uses: XRPLF/actions/.github/workflows/pre-commit.yml@1bde119a1ab71305ba5d3716e7a82cea1c7bdede
     with:
       runs_on: ubuntu-latest
       container: '{ "image": "ghcr.io/xrplf/xrpld/pre-commit:sha-f56b79f" }'

.github/workflows/publish-docs.yml

@@ -41,13 +41,13 @@ env:
 jobs:
   build:
     runs-on: ubuntu-latest
-    container: ghcr.io/xrplf/xrpld/nix-ubuntu:sha-3122de8
+    container: ghcr.io/xrplf/xrpld/nix-ubuntu:sha-2e25435
     steps:
       - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
 
       - name: Prepare runner
-        uses: XRPLF/actions/prepare-runner@e4b6449d55a61c002d7c3fdfa6c20f721ede0606
+        uses: XRPLF/actions/prepare-runner@ad188deb3dae79dc39816e16ddfdad1e06c6fab2
         with:
           enable_ccache: false
 

.github/workflows/reusable-build-test-config.yml

@@ -113,7 +113,7 @@ jobs:
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
 
       - name: Prepare runner
-        uses: XRPLF/actions/prepare-runner@e4b6449d55a61c002d7c3fdfa6c20f721ede0606
+        uses: XRPLF/actions/prepare-runner@ad188deb3dae79dc39816e16ddfdad1e06c6fab2
         with:
           enable_ccache: ${{ inputs.ccache_enabled }}
 

.github/workflows/reusable-clang-tidy.yml

@@ -34,7 +34,7 @@ jobs:
     needs: [determine-files]
     if: ${{ needs.determine-files.outputs.cpp_changed_files != '' || needs.determine-files.outputs.need_full_run == 'true' }}
     runs-on: ["self-hosted", "Linux", "X64", "heavy"]
-    container: "ghcr.io/xrplf/xrpld/nix-debian:sha-3122de8"
+    container: "ghcr.io/xrplf/xrpld/nix-debian:sha-2e25435"
     permissions:
       contents: read
       issues: write
@@ -43,7 +43,7 @@ jobs:
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
 
       - name: Prepare runner
-        uses: XRPLF/actions/prepare-runner@e4b6449d55a61c002d7c3fdfa6c20f721ede0606
+        uses: XRPLF/actions/prepare-runner@ad188deb3dae79dc39816e16ddfdad1e06c6fab2
         with:
           enable_ccache: false
 

.github/workflows/reusable-upload-recipe.yml

@@ -40,7 +40,7 @@ defaults:
 jobs:
   upload:
     runs-on: ubuntu-latest
-    container: ghcr.io/xrplf/xrpld/nix-ubuntu:sha-3122de8
+    container: ghcr.io/xrplf/xrpld/nix-ubuntu:sha-2e25435
     env:
       REMOTE_NAME: ${{ inputs.remote_name }}
       CONAN_LOGIN_USERNAME_XRPLF: ${{ secrets.remote_username }}

.github/workflows/upload-conan-deps.yml

@@ -68,7 +68,7 @@ jobs:
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
 
       - name: Prepare runner
-        uses: XRPLF/actions/prepare-runner@e4b6449d55a61c002d7c3fdfa6c20f721ede0606
+        uses: XRPLF/actions/prepare-runner@ad188deb3dae79dc39816e16ddfdad1e06c6fab2
         with:
           enable_ccache: false
 

.gitignore

@@ -81,9 +81,6 @@ DerivedData
 # Python
 __pycache__
 
-# Rust build artifacts.
-target/
-
 # Direnv's directory
 /.direnv
 

CMakeLists.txt

@@ -96,7 +96,6 @@ find_package(OpenSSL REQUIRED)
 find_package(secp256k1 REQUIRED)
 find_package(SOCI REQUIRED)
 find_package(SQLite3 REQUIRED)
-find_package(wasmi REQUIRED)
 find_package(xxHash REQUIRED)
 
 target_link_libraries(

cmake/XrplCore.cmake

@@ -67,7 +67,6 @@ target_link_libraries(
         Xrpl::opts
         Xrpl::syslibs
         secp256k1::secp256k1
-        wasmi::wasmi
         xrpl.libpb
         xxHash::xxhash
         $<$<BOOL:${voidstar}>:antithesis-sdk-cpp>

conan.lock

@@ -3,7 +3,6 @@
     "requires": [
         "zlib/1.3.2#1cb806da49011867778ffb6ac7190fcb%1782392402.122708",
         "xxhash/0.8.3#681d36a0a6111fc56e5e45ea182c19cc%1782392402.420688",
-        "wasmi/1.0.9#1fecdab9b90c96698eb35ea99ca4f5cb%1782307153.343419",
         "sqlite3/3.53.0#324ada52333108388a9a6108bfa96734%1782392403.185447",
         "soci/4.0.3#e726491a03468795453f7c83fc924a96%1782392402.679521",
         "snappy/1.1.10#968fef506ff261592ec30c574d4a7809%1782307151.633168",

conanfile.py

@@ -34,7 +34,6 @@ class Xrpl(ConanFile):
         "nudb/2.0.9",
         "openssl/3.6.3",
         "soci/4.0.3",
-        "wasmi/1.0.9",
         "zlib/1.3.2",
     ]
 
@@ -222,7 +221,6 @@ class Xrpl(ConanFile):
             "soci::soci",
             "secp256k1::secp256k1",
             "sqlite3::sqlite",
-            "wasmi::wasmi",
             "xxhash::xxhash",
             "zlib::zlib",
         ]

flake.lock

@@ -36,28 +36,7 @@
     "root": {
       "inputs": {
         "nixpkgs": "nixpkgs",
-        "nixpkgs-custom-glibc": "nixpkgs-custom-glibc",
-        "rust-overlay": "rust-overlay"
-      }
-    },
-    "rust-overlay": {
-      "inputs": {
-        "nixpkgs": [
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1784611586,
-        "narHash": "sha256-OfqgY+0hp/zseZB7uyH0U8kIDPS4scZZCyAurEplvG0=",
-        "owner": "oxalica",
-        "repo": "rust-overlay",
-        "rev": "14f58845249f3552a89b07772626b8d3c632fa86",
-        "type": "github"
-      },
-      "original": {
-        "owner": "oxalica",
-        "repo": "rust-overlay",
-        "type": "github"
+        "nixpkgs-custom-glibc": "nixpkgs-custom-glibc"
       }
     }
   },

flake.nix

@@ -10,25 +10,12 @@
       url = "github:NixOS/nixpkgs/9cd98386a38891d1074fc18036b842dc4416f562";
       flake = false;
     };
-    # Pinned Rust toolchains, delivered from the Nix store. Lets the Nix CI
-    # image and dev shell honour the single `rust-toolchain.toml` pin (shared
-    # with the rustup-based non-Nix runners) while staying hermetic — the
-    # toolchain lands in the image's Nix closure and is locked by flake.lock.
-    rust-overlay = {
-      url = "github:oxalica/rust-overlay";
-      inputs.nixpkgs.follows = "nixpkgs";
-    };
   };
 
   outputs =
-    {
-      nixpkgs,
-      nixpkgs-custom-glibc,
-      rust-overlay,
-      ...
-    }:
+    { nixpkgs, nixpkgs-custom-glibc, ... }:
     let
-      forEachSystem = import ./nix/utils.nix { inherit nixpkgs nixpkgs-custom-glibc rust-overlay; };
+      forEachSystem = import ./nix/utils.nix { inherit nixpkgs nixpkgs-custom-glibc; };
     in
     {
       devShells = forEachSystem (import ./nix/devshell.nix);

include/xrpl/config/Constants.h

@@ -175,6 +175,7 @@ struct Keys
     static constexpr auto kVpBaseSquelchEnable = "vp_base_squelch_enable";
     static constexpr auto kVpBaseSquelchMaxSelectedPeers = "vp_base_squelch_max_selected_peers";
     static constexpr auto kVpEnable = "vp_enable";
+    static constexpr auto kVpEnhancedSquelchEnable = "vp_enhanced_squelch_enable";
 };
 
 }  // namespace xrpl

nix/ci-env.nix

@@ -7,10 +7,8 @@ let
   inherit (import ./packages.nix { inherit pkgs; })
     commonPackages
     gccPackage
-    gccVersion
     llvmPackages
     llvmVersion
-    mkVersionedToolLinks
     ;
 
   # Underlying compiler toolchains to wrap (versions pinned in packages.nix).
@@ -129,25 +127,6 @@ in
       customGcov
       customClangForCiEnv
       customBinutils
-      (mkVersionedToolLinks {
-        name = "gcc";
-        package = customGcc;
-        version = gccVersion;
-        tools = [
-          "gcc"
-          "g++"
-          "cpp"
-        ];
-      })
-      (mkVersionedToolLinks {
-        name = "clang";
-        package = customClang;
-        version = llvmVersion;
-        tools = [
-          "clang"
-          "clang++"
-        ];
-      })
       # CA certificate bundle so HTTPS clients (git, curl, conan) can verify
       # TLS connections without ca-certificates being installed in the system.
       pkgs.cacert

nix/devshell.nix

@@ -3,9 +3,7 @@ let
   inherit (import ./packages.nix { inherit pkgs; })
     commonPackages
     gccVersion
-    llvmVersion
     llvmPackages
-    mkVersionedToolLinks
     ;
 
   # Plain nixpkgs stdenvs — no custom glibc, unlike ci-env.nix.
@@ -17,8 +15,6 @@ let
     {
       stdenv,
       compilerName,
-      version ? null,
-      versionedTools ? [ ],
     }:
     let
       compilerVersion =
@@ -29,15 +25,9 @@ let
             echo "Compiler: "
             ${compilerName} --version
           '';
-      versionedLinks = pkgs.lib.optional (version != null) (mkVersionedToolLinks {
-        name = compilerName;
-        package = stdenv.cc;
-        inherit version;
-        tools = versionedTools;
-      });
     in
     (pkgs.mkShell.override { inherit stdenv; }) {
-      packages = commonPackages ++ versionedLinks;
+      packages = commonPackages;
       shellHook = ''
         echo "Welcome to xrpld development shell";
         ${compilerVersion}
@@ -51,22 +41,11 @@ rec {
   gcc = makeShell {
     stdenv = gccStdenv;
     compilerName = "gcc";
-    version = gccVersion;
-    versionedTools = [
-      "gcc"
-      "g++"
-      "cpp"
-    ];
   };
 
   clang = makeShell {
     stdenv = clangStdenv;
     compilerName = "clang";
-    version = llvmVersion;
-    versionedTools = [
-      "clang"
-      "clang++"
-    ];
   };
 
   # Nix provides no compiler; use the one from your system (e.g. Apple Clang).

nix/docker/Dockerfile

@@ -12,7 +12,6 @@ COPY nix/packages.nix /tmp/build/nix/packages.nix
 COPY nix/utils.nix /tmp/build/nix/utils.nix
 COPY flake.nix /tmp/build/
 COPY flake.lock /tmp/build/
-COPY rust-toolchain.toml /tmp/build/
 WORKDIR /tmp/build
 
 FROM builder-source AS builder

nix/docker/README.md

@@ -47,9 +47,7 @@ work without `ca-certificates` being installed in the base image.
      [`test_files/cpp/sources/`](./test_files/cpp/sources) with both `g++` and
      `clang++`, and sanitizers, and
    - compiles the Rust test programs in
-     [`test_files/rust/sources/`](./test_files/rust/sources) with `rustc`, and
-     builds the [`test_files/rust/proc_macro/`](./test_files/rust/proc_macro)
-     workspace with `cargo` to exercise proc-macro dylib loading.
+     [`test_files/rust/sources/`](./test_files/rust/sources) with `rustc`.
 3. **`tester`** — Start again from a clean `BASE_IMAGE` (no Nix toolchain),
    install only the sanitizer runtime libraries
    ([`install-sanitizer-libs.sh`](./install-sanitizer-libs.sh)), and run the
@@ -78,23 +76,20 @@ toolchain being present at runtime. Two pieces make that work:
   [`loader-path.sh`](./loader-path.sh) reports the expected loader path for the
   current architecture, so we can patch the binaries to use the correct loader.
 
-The build then verifies all of this end to end, and the C++ and Rust programs
-go through the same pipeline: each is compiled in `final`, has its `PT_INTERP`
-patched to the target loader, and is then run in the clean `tester` stage to
-confirm it emits the expected diagnostic on a stock base image. The C++ programs
-are in `test_files/cpp/sources/` (a regular binary plus ASan/TSan/UBSan
-variants); the Rust programs are in `test_files/rust/sources/` (a hello binary
-plus panic and overflow-check variants), plus the `test_files/rust/proc_macro/`
-workspace — a crate whose compilation additionally loads a proc-macro dylib, and
-whose resulting binary is patched and run like the others.
+The build then verifies all of this end to end: the C++ test programs in
+`test_files/cpp/sources/` (a regular binary plus ASan/TSan/UBSan variants) and
+the Rust test programs in `test_files/rust/sources/` (a hello binary plus panic
+and overflow-check variants) are compiled in `final`, their `PT_INTERP` is
+patched to the target loader, and they are run in the clean `tester` stage to
+confirm each emits the expected diagnostic on a stock base image.
 
 ## Files
 
-| File                                                                    | Purpose                                                                              |
-| ----------------------------------------------------------------------- | ------------------------------------------------------------------------------------ |
-| [`./Dockerfile`](./Dockerfile)                                          | Multi-stage build described above.                                                   |
-| [`./loader-path.sh`](./loader-path.sh)                                  | Print the dynamic-linker (`PT_INTERP`) path for the current architecture.            |
-| [`./test_files/cpp/`](./test_files/cpp)                                 | C++ sanitizer smoke test: sources + compile/run scripts.                             |
-| [`./test_files/rust/`](./test_files/rust)                               | Rust smoke test: rustc sources + a cargo proc-macro workspace + compile/run scripts. |
-| [`/bin/check-tools.sh`](../../bin/check-tools.sh)                       | Verify every expected tools are present and runnable.                                |
-| [`/bin/install-sanitizer-libs.sh`](../../bin/install-sanitizer-libs.sh) | Install `libasan`/`libtsan`/`libubsan` runtimes on the supported base images.        |
+| File                                                                    | Purpose                                                                       |
+| ----------------------------------------------------------------------- | ----------------------------------------------------------------------------- |
+| [`./Dockerfile`](./Dockerfile)                                          | Multi-stage build described above.                                            |
+| [`./loader-path.sh`](./loader-path.sh)                                  | Print the dynamic-linker (`PT_INTERP`) path for the current architecture.     |
+| [`./test_files/cpp/`](./test_files/cpp)                                 | C++ sanitizer smoke test: sources + compile/run scripts.                      |
+| [`./test_files/rust/`](./test_files/rust)                               | Rust rustc smoke test: sources + compile/run scripts.                         |
+| [`/bin/check-tools.sh`](../../bin/check-tools.sh)                       | Verify every expected tools are present and runnable.                         |
+| [`/bin/install-sanitizer-libs.sh`](../../bin/install-sanitizer-libs.sh) | Install `libasan`/`libtsan`/`libubsan` runtimes on the supported base images. |

nix/docker/test_files/rust/compile-sources.sh

@@ -40,29 +40,6 @@ compile hello
 compile panic
 compile overflow "-C overflow-checks=on"
 
-function compile_proc_macro() {
-    local proj="${src_dir}/../proc_macro"
-
-    echo "=== Building proc-macro workspace (cargo) ==="
-    cargo build --manifest-path "${proj}/Cargo.toml" --offline
-
-    local built="${proj}/target/debug/test_macro"
-    if [ ! -f "${built}" ]; then
-        echo "ERROR: built test_macro binary not found at ${built}" >&2
-        exit 1
-    fi
-
-    local binary="${dst_dir}/proc_macro"
-    cp "${built}" "${binary}"
-
-    echo "=== Patching ${binary} to use ${loader} as PT_INTERP ==="
-    patchelf --set-interpreter "${loader}" --remove-rpath "${binary}"
-
-    rm -rf "${proj}/target"
-}
-
-compile_proc_macro
-
 echo "=== All binaries compiled ==="
 
 ls -la "${dst_dir}"

nix/docker/test_files/rust/proc_macro/Cargo.lock

@@ -1,14 +0,0 @@
-# This file is automatically @generated by Cargo.
-# It is not intended for manual editing.
-version = 4
-
-[[package]]
-name = "echo_macro"
-version = "0.0.0"
-
-[[package]]
-name = "test_macro"
-version = "0.0.0"
-dependencies = [
- "echo_macro",
-]

nix/docker/test_files/rust/proc_macro/Cargo.toml

@@ -1,3 +0,0 @@
-[workspace]
-resolver = "2"
-members = ["echo_macro", "test_macro"]

nix/docker/test_files/rust/proc_macro/echo_macro/Cargo.toml

@@ -1,8 +0,0 @@
-[package]
-name = "echo_macro"
-version = "0.0.0"
-edition = "2024"
-publish = false
-
-[lib]
-proc-macro = true

nix/docker/test_files/rust/proc_macro/echo_macro/src/lib.rs

@@ -1,6 +0,0 @@
-use proc_macro::TokenStream;
-
-#[proc_macro]
-pub fn define_echo(item: TokenStream) -> TokenStream {
-    format!("fn echo() -> u32 {{ {item} }}").parse().unwrap()
-}

nix/docker/test_files/rust/proc_macro/test_macro/Cargo.toml

@@ -1,8 +0,0 @@
-[package]
-name = "test_macro"
-version = "0.0.0"
-edition = "2024"
-publish = false
-
-[dependencies]
-echo_macro = { path = "../echo_macro" }

nix/docker/test_files/rust/proc_macro/test_macro/src/main.rs

@@ -1,9 +0,0 @@
-use echo_macro::define_echo;
-
-define_echo!(42);
-
-fn main() {
-    let a = echo();
-    println!("proc-macro answer = {a}");
-    assert_eq!(a, 42, "proc-macro expansion produced the wrong value");
-}

nix/docker/test_files/rust/run-binaries.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 # Run pre-compiled Rust binaries and confirm each emits its expected diagnostic.
 # Binaries must already exist in <bins_dir> as <name> for name in
-# {hello,panic,overflow,proc_macro}.
+# {hello,panic,overflow}.
 
 set -eo pipefail
 
@@ -54,13 +54,12 @@ declare -A expect=(
     [hello]="Hello from main thread"
     [panic]="explicit panic from test"
     [overflow]="attempt to add with overflow"
-    [proc_macro]="proc-macro answer = 42"
 )
 
-for name in hello panic overflow proc_macro; do
+for name in hello panic overflow; do
     binary="${bins_dir}/${name}"
 
-    if [ "${name}" = "hello" ] || [ "${name}" = "proc_macro" ]; then
+    if [ "${name}" = "hello" ]; then
         expected_rc=0
     else
         expected_rc=nonzero

nix/packages.nix

@@ -15,55 +15,6 @@ let
   runClangTidy = pkgs.writeShellScriptBin "run-clang-tidy" ''
     exec ${pkgs.python3}/bin/python3 ${llvmPackages.clang-unwrapped}/bin/run-clang-tidy "$@"
   '';
-
-  rustToolchain = pkgs.rust-bin.fromRustupToolchainFile ../rust-toolchain.toml;
-
-  # Nix wraps its toolchain so that binaries are exposed only under unsuffixed
-  # names (gcc, g++, clang-tidy, ...). Several tools probe for a
-  # version-suffixed name first and fall back to a system binary on the PATH
-  # when Nix doesn't provide it:
-  #   - Conan's Boost recipe looks up `g++-<major>` before plain `g++`.
-  #   - bin/pre-commit/clang_tidy_check.py looks up `run-clang-tidy-<v>` and
-  #     `clang-apply-replacements-<v>` before the unsuffixed names.
-  # On a host that also has the matching system binary (e.g. Ubuntu's
-  # `/usr/bin/g++-15` or `clang-tidy-22`) the probe escapes Nix and mixes a
-  # system tool into the Nix environment. Generate version-suffixed symlinks
-  # next to a package's tools so those probes resolve to the Nix ones.
-  #
-  # Compiler links must point at whichever compiler is active in a given
-  # environment (the plain stdenv compiler in the dev shell, the custom-glibc
-  # wrappers in ci-env.nix), so those callers pass their own `package`; the
-  # clang tooling is environment-independent and is linked in commonPackages.
-  mkVersionedToolLinks =
-    {
-      name,
-      package,
-      version,
-      tools,
-    }:
-    pkgs.linkFarm "${name}-${toString version}-versioned-links" (
-      map (tool: {
-        name = "bin/${tool}-${toString version}";
-        path = "${package}/bin/${tool}";
-      }) tools
-    );
-
-  clangToolLinks = mkVersionedToolLinks {
-    name = "clang-tools";
-    package = clangTools;
-    version = llvmVersion;
-    tools = [
-      "clang-tidy"
-      "clang-apply-replacements"
-      "clang-format"
-    ];
-  };
-  runClangTidyLink = mkVersionedToolLinks {
-    name = "run-clang-tidy";
-    package = runClangTidy;
-    version = llvmVersion;
-    tools = [ "run-clang-tidy" ];
-  };
 in
 {
   inherit
@@ -71,12 +22,9 @@ in
     llvmVersion
     gccPackage
     llvmPackages
-    mkVersionedToolLinks
     ;
 
   commonPackages = with pkgs; [
-    clangToolLinks
-    runClangTidyLink
     ccache
     clangbuildanalyzer
     clangTools
@@ -115,10 +63,14 @@ in
     vim
     zip
     # Rust packages
+    cargo
     cargo-audit
     cargo-llvm-cov
     cargo-nextest
+    clippy
     corrosion
-    rustToolchain
+    rust-analyzer
+    rustc
+    rustfmt
   ];
 }

nix/utils.nix

@@ -1,8 +1,4 @@
-{
-  nixpkgs,
-  nixpkgs-custom-glibc,
-  rust-overlay,
-}:
+{ nixpkgs, nixpkgs-custom-glibc }:
 function:
 nixpkgs.lib.genAttrs
   [
@@ -14,12 +10,7 @@ nixpkgs.lib.genAttrs
   (
     system:
     function {
-      # rust-overlay adds `pkgs.rust-bin`, from which we build the pinned Rust
-      # toolchain (see packages.nix). Consumed by both the CI image and dev shell.
-      pkgs = import nixpkgs {
-        inherit system;
-        overlays = [ (import rust-overlay) ];
-      };
+      pkgs = import nixpkgs { inherit system; };
       # glibc 2.31 — matches the system libc on Ubuntu 20.04 LTS. Sourced
       # from the nixpkgs snapshot pinned via the `nixpkgs-custom-glibc`
       # flake input, so the build uses the compiler from that snapshot

rust-toolchain.toml

@@ -1,4 +1,8 @@
+# Rust toolchain pin for rustup-based CI runners and local development.
+# rustup reads this file and installs the pinned toolchain (see the
+# prepare-runner action in XRPLF/actions, which runs `rustup toolchain install`).
+# NOTE: the Nix CI image and development shell ignore this file; its rustc comes from flake.lock.
 [toolchain]
 channel = "1.95"
-components = ["rustfmt", "clippy", "rust-analyzer"]
+components = ["rustfmt", "clippy"]
 profile = "minimal"

src/test/overlay/base_squelch_test.cpp

@@ -2,26 +2,27 @@
 #include <test/jtx/envconfig.h>
 
 #include <xrpld/app/main/Application.h>
+#include <xrpld/core/Config.h>
 #include <xrpld/overlay/Message.h>
 #include <xrpld/overlay/Peer.h>
 #include <xrpld/overlay/ReduceRelayCommon.h>
 #include <xrpld/overlay/Slot.h>
-#include <xrpld/overlay/Squelch.h>
+#include <xrpld/overlay/SquelchStore.h>
 #include <xrpld/overlay/detail/Handshake.h>
 
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/basics/random.h>
+#include <xrpl/beast/clock/manual_clock.h>
 #include <xrpl/beast/net/IPAddress.h>
 #include <xrpl/beast/net/IPEndpoint.h>
 #include <xrpl/beast/unit_test/suite.h>
 #include <xrpl/beast/utility/Journal.h>
+#include <xrpl/core/ServiceRegistry.h>
 #include <xrpl/json/json_value.h>
 #include <xrpl/protocol/KeyType.h>
 #include <xrpl/protocol/PublicKey.h>
 #include <xrpl/protocol/SecretKey.h>
 
-#include <boost/asio/ip/address.hpp>
-
 #include <xrpl.pb.h>
 
 #include <algorithm>
@@ -35,13 +36,12 @@
 #include <memory>
 #include <numeric>
 #include <optional>
+#include <ostream>
 #include <random>
-#include <ratio>
 #include <set>
 #include <sstream>
 #include <stdexcept>
 #include <string>
-#include <tuple>
 #include <unordered_map>
 #include <utility>
 #include <vector>
@@ -50,6 +50,21 @@ namespace xrpl::test {
 
 using namespace std::chrono;
 
+template <class Clock>
+class ExtendedManualClock : public beast::ManualClock<Clock>
+{
+public:
+    using typename beast::ManualClock<Clock>::duration;
+    using typename beast::ManualClock<Clock>::time_point;
+
+    void
+    randAdvance(std::chrono::milliseconds min, std::chrono::milliseconds max)
+    {
+        auto ms = randInt(min.count(), max.count());
+        this->advance(std::chrono::milliseconds(ms));
+    }
+};
+
 class Link;
 
 using MessageSPtr = std::shared_ptr<Message>;
@@ -59,10 +74,11 @@ using PeerWPtr = std::weak_ptr<Peer>;
 using SquelchCB = std::function<void(PublicKey const&, PeerWPtr const&, std::uint32_t)>;
 using UnsquelchCB = std::function<void(PublicKey const&, PeerWPtr const&)>;
 using LinkIterCB = std::function<void(Link&, MessageSPtr)>;
+using TestStopwatch = ExtendedManualClock<std::chrono::steady_clock>;
 
-static constexpr std::uint32_t kMaxPeers = 10;
-static constexpr std::uint32_t kMaxValidators = 10;
-static constexpr std::uint32_t kMaxMessages = 200000;
+static constexpr std::uint32_t maxPeers = 10;
+static constexpr std::uint32_t maxValidators = 10;
+static constexpr std::uint32_t maxMessages = 200000;
 
 /**
  * Simulate two entities - peer directly connected to the server
@@ -126,6 +142,12 @@ public:
     {
         return {};
     }
+    [[nodiscard]] std::string const&
+    fingerprint() const override
+    {
+        static std::string const kFingerprint{};
+        return kFingerprint;
+    }
     [[nodiscard]] bool
     supportsFeature(ProtocolFeature f) const override
     {
@@ -193,54 +215,6 @@ public:
     }
 };
 
-/**
- * Manually advanced clock.
- */
-class ManualClock
-{
-public:
-    using rep = uint64_t;
-    using period = std::milli;
-    using duration = std::chrono::duration<std::uint32_t, period>;
-    using time_point = std::chrono::time_point<ManualClock>;
-    inline static bool const is_steady = false;  // NOLINT(readability-identifier-naming)
-
-    static void
-    advance(duration d) noexcept
-    {
-        kNow += d;
-    }
-
-    static void
-    randAdvance(milliseconds min, milliseconds max)
-    {
-        kNow += randDuration(min, max);
-    }
-
-    static void
-    reset() noexcept
-    {
-        kNow = time_point(seconds(0));
-    }
-
-    static time_point
-    now() noexcept
-    {
-        return kNow;
-    }
-
-    static duration
-    randDuration(milliseconds min, milliseconds max)
-    {
-        return duration(milliseconds(randInt(min.count(), max.count())));
-    }
-
-    explicit ManualClock() = default;
-
-private:
-    inline static time_point kNow = time_point(seconds(0));
-};
-
 /**
  * Simulate server's OverlayImpl
  */
@@ -255,12 +229,20 @@ public:
         uint256 const& key,
         PublicKey const& validator,
         Peer::id_t id,
-        SquelchCB f,
-        protocol::MessageType type = protocol::mtVALIDATION) = 0;
+        SquelchCB f) = 0;
 
     virtual void deleteIdlePeers(UnsquelchCB) = 0;
 
     virtual void deletePeer(Peer::id_t, UnsquelchCB) = 0;
+
+    TestStopwatch&
+    clock()
+    {
+        return clock_;
+    }
+
+protected:
+    TestStopwatch clock_;
 };
 
 class Validator;
@@ -432,7 +414,7 @@ public:
         return message_;
     }
 
-    [[nodiscard]] std::uint16_t
+    std::uint16_t
     id() const
     {
         return id_;
@@ -462,28 +444,43 @@ private:
     std::uint16_t id_ = 0;
 };
 
+class BaseSquelchingTestSlots : public reduce_relay::Slots
+{
+    using Slots = reduce_relay::Slots;
+
+public:
+    BaseSquelchingTestSlots(
+        ServiceRegistry& registry,
+        reduce_relay::SquelchHandler& handler,
+        Config const& config,
+        reduce_relay::Slots::clock_type& clock)
+        : Slots(registry, handler, config, clock)
+    {
+    }
+
+    Slots::slots_map const&
+    getSlots() const
+    {
+        return trustedSlots_;
+    }
+};
+
 class PeerSim : public PeerPartial, public std::enable_shared_from_this<PeerSim>
 {
 public:
-    using id_t = Peer::id_t;
-    PeerSim(Overlay& overlay, beast::Journal journal) : overlay_(overlay), squelch_(journal)
+    PeerSim(Overlay& overlay, beast::Journal journal)
+        : overlay_(overlay), squelchStore_(journal, overlay_.clock())
     {
     }
 
     ~PeerSim() override = default;
 
-    id_t
+    Peer::id_t
     id() const override
     {
         return id_;
     }
 
-    std::string const&
-    fingerprint() const override
-    {
-        return fingerprint_;
-    }
-
     static void
     resetId()
     {
@@ -498,11 +495,10 @@ public:
     {
         auto validator = m->getValidatorKey();
         assert(validator);
-        if (!squelch_.expireSquelch(*validator))  // NOLINT(bugprone-unchecked-optional-access)
+        if (squelchStore_.isSquelched(*validator))
             return;
 
-        overlay_.updateSlotAndSquelch(
-            {}, *validator, id(), f);  // NOLINT(bugprone-unchecked-optional-access)
+        overlay_.updateSlotAndSquelch({}, *validator, id(), f);
     }
 
     /**
@@ -513,22 +509,15 @@ public:
     {
         auto validator = squelch.validatorpubkey();
         PublicKey const key(Slice(validator.data(), validator.size()));
-        if (squelch.squelch())
-        {
-            squelch_.addSquelch(key, std::chrono::seconds{squelch.squelchduration()});
-        }
-        else
-        {
-            squelch_.removeSquelch(key);
-        }
+        squelchStore_.handleSquelch(
+            key, squelch.squelch(), std::chrono::seconds{squelch.squelchduration()});
     }
 
 private:
-    inline static id_t sid = 0;
-    std::string fingerprint_;
-    id_t id_{sid++};
+    inline static Peer::id_t sid = 0;
+    Peer::id_t id_{sid++};
     Overlay& overlay_;
-    reduce_relay::Squelch<ManualClock> squelch_;
+    reduce_relay::SquelchStore squelchStore_;
 };
 
 class OverlaySim : public Overlay, public reduce_relay::SquelchHandler
@@ -536,9 +525,8 @@ class OverlaySim : public Overlay, public reduce_relay::SquelchHandler
     using Peers = std::unordered_map<Peer::id_t, PeerSPtr>;
 
 public:
-    using id_t = Peer::id_t;
-    using clock_type = ManualClock;
-    OverlaySim(Application& app) : slots_(app, *this, app.config()), registry_(app)
+    using clock_type = TestStopwatch;
+    OverlaySim(Application& app) : slots_(app, *this, app.config(), clock_), logs_(app.getLogs())
     {
     }
 
@@ -548,31 +536,35 @@ public:
     clear()
     {
         peers_.clear();
-        ManualClock::advance(hours(1));
+        clock_.advance(hours(1));
         slots_.deleteIdlePeers();
     }
 
     std::uint16_t
     inState(PublicKey const& validator, reduce_relay::PeerState state)
     {
-        auto res = slots_.inState(validator, state);
-        return res ? *res : 0;
+        auto const& it = slots_.getSlots().find(validator);
+        if (it != slots_.getSlots().end())
+        {
+            return std::count_if(
+                it->second.getPeers().begin(), it->second.getPeers().end(), [&](auto const& it) {
+                    return (it.second.state == state);
+                });
+        }
+
+        return 0;
     }
 
     void
-    updateSlotAndSquelch(
-        uint256 const& key,
-        PublicKey const& validator,
-        Peer::id_t id,
-        SquelchCB f,
-        protocol::MessageType type = protocol::mtVALIDATION) override
+    updateSlotAndSquelch(uint256 const& key, PublicKey const& validator, Peer::id_t id, SquelchCB f)
+        override
     {
         squelch_ = f;
-        slots_.updateSlotAndSquelch(key, validator, id, type);
+        slots_.updateSlotAndSquelch(key, validator, id, true);
     }
 
     void
-    deletePeer(id_t id, UnsquelchCB f) override
+    deletePeer(Peer::id_t id, UnsquelchCB f) override
     {
         unsquelch_ = f;
         slots_.deletePeer(id, true);
@@ -592,7 +584,7 @@ public:
         Peer::id_t id = 0;
         if (peersCache_.empty() || !useCache)
         {
-            peer = std::make_shared<PeerSim>(*this, registry_.getJournal("Squelch"));
+            peer = std::make_shared<PeerSim>(*this, logs_.journal("Squelch"));
             id = peer->id();
         }
         else
@@ -648,36 +640,57 @@ public:
     bool
     isCountingState(PublicKey const& validator)
     {
-        return slots_.inState(validator, reduce_relay::SlotState::Counting);
+        auto const& it = slots_.getSlots().find(validator);
+        if (it != slots_.getSlots().end())
+            return it->second.getState() == reduce_relay::SlotState::Counting;
+
+        return false;
     }
 
-    std::set<id_t>
+    std::set<Peer::id_t>
     getSelected(PublicKey const& validator)
     {
-        return slots_.getSelected(validator);
+        auto const& it = slots_.getSlots().find(validator);
+        if (it == slots_.getSlots().end())
+            return {};
+
+        std::set<Peer::id_t> r;
+        for (auto const& [id, info] : it->second.getPeers())
+        {
+            if (info.state == reduce_relay::PeerState::Selected)
+                r.insert(id);
+        }
+
+        return r;
     }
 
     bool
     isSelected(PublicKey const& validator, Peer::id_t peer)
     {
-        auto selected = slots_.getSelected(validator);
+        auto selected = getSelected(validator);
         return selected.contains(peer);
     }
 
-    id_t
+    Peer::id_t
     getSelectedPeer(PublicKey const& validator)
     {
-        auto selected = slots_.getSelected(validator);
+        auto selected = getSelected(validator);
         assert(!selected.empty());
         return *selected.begin();
     }
 
-    std::unordered_map<
-        id_t,
-        std::tuple<reduce_relay::PeerState, std::uint16_t, std::uint32_t, std::uint32_t>>
+    std::unordered_map<Peer::id_t, reduce_relay::Slot::PeerInfo>
     getPeers(PublicKey const& validator)
     {
-        return slots_.getPeers(validator);
+        auto const& it = slots_.getSlots().find(validator);
+        if (it == slots_.getSlots().end())
+            return {};
+
+        auto r = std::unordered_map<Peer::id_t, reduce_relay::Slot::PeerInfo>();
+        for (auto const& [id, info] : it->second.getPeers())
+            r.emplace(id, info);
+
+        return r;
     }
 
     std::uint16_t
@@ -693,18 +706,32 @@ private:
         if (auto it = peers_.find(id); it != peers_.end())
             squelch_(validator, it->second, squelchDuration);
     }
+
+    void
+    squelchAll(
+        PublicKey const& validator,
+        std::uint32_t duration,
+        std::function<void(Peer::id_t)> callback) override
+    {
+        for (auto const& [id, peer] : peers_)
+        {
+            squelch_(validator, peer, duration);
+            callback(id);
+        }
+    }
     void
     unsquelch(PublicKey const& validator, Peer::id_t id) const override
     {
         if (auto it = peers_.find(id); it != peers_.end())
             unsquelch_(validator, it->second);
     }
+
     SquelchCB squelch_;
     UnsquelchCB unsquelch_;
     Peers peers_;
     Peers peersCache_;
-    reduce_relay::Slots<ManualClock> slots_;
-    ServiceRegistry& registry_;
+    BaseSquelchingTestSlots slots_;
+    Logs& logs_;
 };
 
 class Network
@@ -718,8 +745,8 @@ public:
     void
     init()
     {
-        validators_.resize(kMaxValidators);
-        for (int p = 0; p < kMaxPeers; p++)
+        validators_.resize(maxValidators);
+        for (int p = 0; p < maxPeers; p++)
         {
             auto peer = overlay_.addPeer();
             for (auto& v : validators_)
@@ -763,7 +790,7 @@ public:
     void
     purgePeers()
     {
-        while (overlay_.getNumPeers() > kMaxPeers)
+        while (overlay_.getNumPeers() > maxPeers)
             deleteLastPeer();
     }
 
@@ -820,7 +847,7 @@ public:
     {
         auto size = max - min;
         std::vector<std::uint32_t> s(size);
-        std::iota(s.begin(), s.end(), min);  // NOLINT(modernize-use-ranges)
+        std::ranges::iota(s, min);
         std::random_device d;
         std::mt19937 g(d());
         std::shuffle(s.begin(), s.end(), g);
@@ -831,14 +858,10 @@ public:
     void
     propagate(
         LinkIterCB link,
-        std::uint16_t nValidators = kMaxValidators,
-        std::uint32_t nMessages = kMaxMessages,
-        bool purge = true,
-        bool resetClock = true)
+        std::uint16_t nValidators = maxValidators,
+        std::uint32_t nMessages = maxMessages,
+        bool purge = true)
     {
-        if (resetClock)
-            ManualClock::reset();
-
         if (purge)
         {
             purgePeers();
@@ -847,7 +870,7 @@ public:
 
         for (int m = 0; m < nMessages; ++m)
         {
-            ManualClock::randAdvance(milliseconds(1800), milliseconds(2200));
+            overlay_.clock().randAdvance(milliseconds(1800), milliseconds(2200));
             forRand(0, nValidators, [&](std::uint32_t v) { validators_[v].forLinks(link); });
         }
     }
@@ -878,7 +901,7 @@ public:
             for (auto& [_, v] : peers)
             {
                 (void)_;
-                if (std::get<reduce_relay::PeerState>(v) == reduce_relay::PeerState::Squelched)
+                if (v.state == reduce_relay::PeerState::Squelched)
                     return false;
             }
         }
@@ -890,10 +913,9 @@ private:
     std::vector<Validator> validators_;
 };
 
-class reduce_relay_test : public beast::unit_test::Suite
+class base_squelch_test : public beast::unit_test::Suite
 {
-    using Slot = reduce_relay::Slot<ManualClock>;
-    using id_t = Peer::id_t;
+    using Slot = reduce_relay::Slot;
 
 protected:
     void
@@ -903,7 +925,7 @@ protected:
         std::cout << msg << " "
                   << "num peers " << (int)network_.overlay().getNumPeers() << std::endl;
         for (auto& [k, v] : peers)
-            std::cout << k << ":" << (int)std::get<reduce_relay::PeerState>(v) << " ";
+            std::cout << k << ":" << to_string(v.state) << " ";
         std::cout << std::endl;
     }
 
@@ -943,7 +965,7 @@ protected:
         Peer::id_t peer{};
         std::uint16_t validator{};
         std::optional<PublicKey> key;
-        time_point<ManualClock> time;
+        TestStopwatch::time_point time;
         bool handled = false;
     };
 
@@ -956,12 +978,12 @@ protected:
     {
         std::unordered_map<EventType, Event> events{
             {EventType::LinkDown, {}}, {EventType::PeerDisconnected, {}}};
-        time_point<ManualClock> lastCheck = ManualClock::now();
+        auto lastCheck = network_.overlay().clock().now();
 
         network_.reset();
         network_.propagate([&](Link& link, MessageSPtr m) {
             auto& validator = link.validator();
-            auto now = ManualClock::now();
+            auto const now = network_.overlay().clock().now();
 
             bool squelched = false;
             std::stringstream str;
@@ -982,7 +1004,10 @@ protected:
                     str << s << " ";
                 if (log)
                 {
-                    std::cout << (double)reduce_relay::epoch<milliseconds>(now).count() / 1000.
+                    std::cout << (double)std::chrono::duration_cast<milliseconds>(
+                                     now.time_since_epoch())
+                                     .count() /
+                            1000.
                               << " random, squelched, validator: " << validator.id()
                               << " peers: " << str.str() << std::endl;
                 }
@@ -1011,7 +1036,9 @@ protected:
                     }
                     else
                     {
-                        events[event].isSelected = network_.isSelected(link.peerId());
+                        {
+                            events[event].isSelected = network_.isSelected(link.peerId());
+                        }
                     }
                 };
                 auto r = randInt(0, 1000);
@@ -1063,8 +1090,14 @@ protected:
                 {
                     event.isSelected = network_.overlay().isSelected(*event.key, event.peer);
                     auto peers = network_.overlay().getPeers(*event.key);
-                    auto d = reduce_relay::epoch<milliseconds>(now).count() -
-                        std::get<3>(peers[event.peer]);
+
+                    auto d = std::chrono::duration_cast<std::chrono::milliseconds>(
+                                 now.time_since_epoch())
+                                 .count() -
+                        std::chrono::duration_cast<std::chrono::milliseconds>(
+                            peers[event.peer].lastMessage.time_since_epoch())
+                            .count();
+
                     mustHandle = event.isSelected &&
                         d > milliseconds(reduce_relay::kIdled).count() &&
                         network_.overlay().inState(*event.key, reduce_relay::PeerState::Squelched) >
@@ -1115,7 +1148,6 @@ protected:
     checkCounting(PublicKey const& validator, bool isCountingState)
     {
         auto countingState = network_.overlay().isCountingState(validator);
-        BEAST_EXPECT(countingState == isCountingState);
         return countingState == isCountingState;
     }
 
@@ -1146,7 +1178,7 @@ protected:
     testPeerUnsquelchedTooSoon(bool log)
     {
         doTest("Peer Unsquelched Too Soon", log, [this](bool log) {
-            BEAST_EXPECT(propagateNoSquelch(log, 1, false, false, false));
+            BEAST_EXPECT(propagateNoSquelch(log, 1, false, false));
         });
     }
 
@@ -1157,9 +1189,9 @@ protected:
     void
     testPeerUnsquelched(bool log)
     {
-        ManualClock::advance(seconds(601));
+        network_.overlay().clock().advance(seconds(601));
         doTest("Peer Unsquelched", log, [this](bool log) {
-            BEAST_EXPECT(propagateNoSquelch(log, 2, true, true, false));
+            BEAST_EXPECT(propagateNoSquelch(log, 2, true, true));
         });
     }
 
@@ -1167,9 +1199,9 @@ protected:
      * Propagate enough messages to generate one squelch event
      */
     bool
-    propagateAndSquelch(bool log, bool purge = true, bool resetClock = true)
+    propagateAndSquelch(bool log, bool purge = true)
     {
-        int n = 0;
+        int squelchEvents = 0;
         network_.propagate(
             [&](Link& link, MessageSPtr message) {
                 std::uint16_t squelched = 0;
@@ -1183,32 +1215,27 @@ protected:
                 {
                     BEAST_EXPECT(
                         squelched ==
-                        kMaxPeers - env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
-                    n++;
+                        maxPeers - env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
+                    squelchEvents++;
                 }
             },
             1,
             reduce_relay::kMaxMessageThreshold + 2,
-            purge,
-            resetClock);
+            purge);
         auto selected = network_.overlay().getSelected(network_.validator(0));
         BEAST_EXPECT(selected.size() == env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
-        BEAST_EXPECT(n == 1);  // only one selection round
+
+        BEAST_EXPECT(squelchEvents == 1);  // only one selection round
         auto res = checkCounting(network_.validator(0), false);
         BEAST_EXPECT(res);
-        return n == 1 && res;
+        return squelchEvents == 1 && res;
     }
 
     /**
      * Send fewer message so that squelch event is not generated
      */
     bool
-    propagateNoSquelch(
-        bool log,
-        std::uint16_t nMessages,
-        bool countingState,
-        bool purge = true,
-        bool resetClock = true)
+    propagateNoSquelch(bool log, std::uint16_t nMessages, bool countingState, bool purge = true)
     {
         bool squelched = false;
         network_.propagate(
@@ -1222,9 +1249,9 @@ protected:
             },
             1,
             nMessages,
-            purge,
-            resetClock);
+            purge);
         auto res = checkCounting(network_.validator(0), countingState);
+        BEAST_EXPECT(res);
         return !squelched && res;
     }
 
@@ -1236,9 +1263,9 @@ protected:
     testNewPeer(bool log)
     {
         doTest("New Peer", log, [this](bool log) {
-            BEAST_EXPECT(propagateAndSquelch(log, true, false));
+            BEAST_EXPECT(propagateAndSquelch(log, true));
             network_.addPeer();
-            BEAST_EXPECT(propagateNoSquelch(log, 1, true, false, false));
+            BEAST_EXPECT(propagateNoSquelch(log, 1, true, false));
         });
     }
 
@@ -1250,15 +1277,14 @@ protected:
     testSelectedPeerDisconnects(bool log)
     {
         doTest("Selected Peer Disconnects", log, [this](bool log) {
-            ManualClock::advance(seconds(601));
-            BEAST_EXPECT(propagateAndSquelch(log, true, false));
+            network_.overlay().clock().advance(seconds(601));
+            BEAST_EXPECT(propagateAndSquelch(log, true));
             auto id = network_.overlay().getSelectedPeer(network_.validator(0));
             std::uint16_t unsquelched = 0;
             network_.overlay().deletePeer(
                 id, [&](PublicKey const& key, PeerWPtr const& peer) { unsquelched++; });
             BEAST_EXPECT(
-                unsquelched ==
-                kMaxPeers - env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
+                unsquelched == maxPeers - env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
             BEAST_EXPECT(checkCounting(network_.validator(0), true));
         });
     }
@@ -1271,16 +1297,15 @@ protected:
     testSelectedPeerStopsRelaying(bool log)
     {
         doTest("Selected Peer Stops Relaying", log, [this](bool log) {
-            ManualClock::advance(seconds(601));
-            BEAST_EXPECT(propagateAndSquelch(log, true, false));
-            ManualClock::advance(reduce_relay::kIdled + seconds(1));
+            network_.overlay().clock().advance(seconds(601));
+            BEAST_EXPECT(propagateAndSquelch(log, true));
+            network_.overlay().clock().advance(reduce_relay::kIdled + seconds(1));
             std::uint16_t unsquelched = 0;
             network_.overlay().deleteIdlePeers(
                 [&](PublicKey const& key, PeerWPtr const& peer) { unsquelched++; });
-            auto peers = network_.overlay().getPeers(network_.validator(0));
+
             BEAST_EXPECT(
-                unsquelched ==
-                kMaxPeers - env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
+                unsquelched == maxPeers - env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
             BEAST_EXPECT(checkCounting(network_.validator(0), true));
         });
     }
@@ -1292,12 +1317,11 @@ protected:
     testSquelchedPeerDisconnects(bool log)
     {
         doTest("Squelched Peer Disconnects", log, [this](bool log) {
-            ManualClock::advance(seconds(601));
-            BEAST_EXPECT(propagateAndSquelch(log, true, false));
+            network_.overlay().clock().advance(seconds(601));
+            BEAST_EXPECT(propagateAndSquelch(log, true));
             auto peers = network_.overlay().getPeers(network_.validator(0));
             auto it = std::ranges::find_if(peers, [&](auto it) {
-                return std::get<reduce_relay::PeerState>(it.second) ==
-                    reduce_relay::PeerState::Squelched;
+                return it.second.state == reduce_relay::PeerState::Squelched;
             });
             assert(it != peers.end());
             std::uint16_t unsquelched = 0;
@@ -1389,7 +1413,7 @@ vp_enable=0
             {
                 c.loadFromString(toLoad);
             }
-            catch (std::runtime_error const& e)
+            catch (std::runtime_error& e)
             {
                 error = e.what();
             }
@@ -1433,7 +1457,7 @@ vp_base_squelch_max_selected_peers=2
             {
                 c2.loadFromString(toLoad);
             }
-            catch (std::runtime_error const& e)
+            catch (std::runtime_error& e)
             {
                 error = e.what();
             }
@@ -1446,27 +1470,29 @@ vp_base_squelch_max_selected_peers=2
     testBaseSquelchReady(bool log)
     {
         doTest("BaseSquelchReady", log, [&](bool log) {
-            ManualClock::reset();
-            auto createSlots = [&](bool baseSquelchEnabled) -> reduce_relay::Slots<ManualClock> {
+            auto createSlots = [&](bool baseSquelchEnabled,
+                                   TestStopwatch stopwatch) -> reduce_relay::Slots {
                 env_.app().config().vpReduceRelayBaseSquelchEnable = baseSquelchEnabled;
-                return reduce_relay::Slots<ManualClock>(
-                    env_.app(), network_.overlay(), env_.app().config());
+                return reduce_relay::Slots(
+                    env_.app(), network_.overlay(), env_.app().config(), stopwatch);
             };
+
+            TestStopwatch stopwatch;
             // base squelching must not be ready if squelching is disabled
-            BEAST_EXPECT(!createSlots(false).baseSquelchReady());
+            BEAST_EXPECT(!createSlots(false, stopwatch).baseSquelchReady());
 
             // base squelch must not be ready as not enough time passed from
             // bootup
-            BEAST_EXPECT(!createSlots(true).baseSquelchReady());
+            BEAST_EXPECT(!createSlots(true, stopwatch).baseSquelchReady());
 
-            ManualClock::advance(reduce_relay::kWaitOnBootup + minutes{1});
+            stopwatch.advance(reduce_relay::kWaitOnBootup + minutes{1});
 
             // base squelch enabled and bootup time passed
-            BEAST_EXPECT(createSlots(true).baseSquelchReady());
+            BEAST_EXPECT(createSlots(true, stopwatch).baseSquelchReady());
 
             // even if time passed, base squelching must not be ready if turned
             // off in the config
-            BEAST_EXPECT(!createSlots(false).baseSquelchReady());
+            BEAST_EXPECT(!createSlots(false, stopwatch).baseSquelchReady());
         });
     }
 
@@ -1476,7 +1502,7 @@ vp_base_squelch_max_selected_peers=2
         doTest("Duplicate Message", log, [&](bool log) {
             network_.reset();
             // update message count for the same peer/validator
-            std::int16_t const nMessages = 5;
+            std::int16_t nMessages = 5;
             for (int i = 0; i < nMessages; i++)
             {
                 uint256 const key(i);
@@ -1487,21 +1513,21 @@ vp_base_squelch_max_selected_peers=2
             auto peers = network_.overlay().getPeers(network_.validator(0));
             // first message changes Slot state to Counting and is not counted,
             // hence '-1'.
-            BEAST_EXPECT(std::get<1>(peers[0]) == (nMessages - 1));
+            BEAST_EXPECT(peers[0].count == (nMessages - 1));
             // add duplicate
             uint256 const key(nMessages - 1);
             network_.overlay().updateSlotAndSquelch(
                 key, network_.validator(0), 0, [&](PublicKey const&, PeerWPtr, std::uint32_t) {});
             // confirm the same number of messages
             peers = network_.overlay().getPeers(network_.validator(0));
-            BEAST_EXPECT(std::get<1>(peers[0]) == (nMessages - 1));
+            BEAST_EXPECT(peers[0].count == (nMessages - 1));
             // advance the clock
-            ManualClock::advance(reduce_relay::kIdled + seconds(1));
+            network_.overlay().clock().advance(reduce_relay::kIdled + seconds(1));
             network_.overlay().updateSlotAndSquelch(
                 key, network_.validator(0), 0, [&](PublicKey const&, PeerWPtr, std::uint32_t) {});
             peers = network_.overlay().getPeers(network_.validator(0));
             // confirm message number increased
-            BEAST_EXPECT(std::get<1>(peers[0]) == nMessages);
+            BEAST_EXPECT(peers[0].count == nMessages);
         });
     }
 
@@ -1511,8 +1537,14 @@ vp_base_squelch_max_selected_peers=2
         void
         squelch(PublicKey const&, Peer::id_t, std::uint32_t duration) const override
         {
-            maxDuration = std::max<uint32_t>(duration, maxDuration);
+            maxDuration = std::max<std::uint32_t>(duration, maxDuration);
         }
+
+        void
+        squelchAll(PublicKey const&, std::uint32_t, std::function<void(Peer::id_t)>) override
+        {
+        }
+
         void
         unsquelch(PublicKey const&, Peer::id_t) const override
         {
@@ -1529,10 +1561,11 @@ vp_base_squelch_max_selected_peers=2
 
             auto run = [&](int npeers) {
                 handler.maxDuration = 0;
-                reduce_relay::Slots<ManualClock> slots(env_.app(), handler, env_.app().config());
+                reduce_relay::Slots slots(
+                    env_.app(), handler, env_.app().config(), network_.overlay().clock());
                 // 1st message from a new peer switches the slot
                 // to counting state and resets the counts of all peers +
-                // MAX_MESSAGE_THRESHOLD + 1 messages to reach the threshold
+                // kMaxMessageThreshold + 1 messages to reach the threshold
                 // and switch the slot's state to peer selection.
                 for (int m = 1; m <= reduce_relay::kMaxMessageThreshold + 2; m++)
                 {
@@ -1540,14 +1573,13 @@ vp_base_squelch_max_selected_peers=2
                     {
                         // make unique message hash to make the
                         // slot's internal hash router accept the message
-                        std::uint64_t const mid = (m * 1000) + peer;
+                        std::uint64_t mid = (m * 1000) + peer;
                         uint256 const message{mid};
-                        slots.updateSlotAndSquelch(
-                            message, validator, peer, protocol::MessageType::mtVALIDATION);
+                        slots.updateSlotAndSquelch(message, validator, peer, true);
                     }
                 }
                 // make Slot's internal hash router expire all messages
-                ManualClock::advance(hours(1));
+                network_.overlay().clock().advance(hours(1));
             };
 
             using namespace reduce_relay;
@@ -1613,7 +1645,7 @@ vp_base_squelch_max_selected_peers=2
                 beast::IP::Address const addr = boost::asio::ip::make_address("172.1.1.100");
 
                 setEnv(outboundEnable);
-                auto request = xrpl::makeRequest(
+                auto request = makeRequest(
                     true,
                     env_.app().config().compression,
                     false,
@@ -1632,8 +1664,8 @@ vp_base_squelch_max_selected_peers=2
                 BEAST_EXPECT(!(peerEnabled ^ inboundEnabled));
 
                 setEnv(inboundEnable);
-                auto httpResp = xrpl::makeResponse(
-                    true, httpRequest, addr, addr, uint256{1}, 1, {1, 0}, env_.app());
+                auto httpResp =
+                    makeResponse(true, httpRequest, addr, addr, uint256{1}, 1, {1, 0}, env_.app());
                 // outbound is enabled if the response's header has the feature
                 // enabled and the peer's configuration is enabled
                 auto const outboundEnabled =
@@ -1651,7 +1683,7 @@ vp_base_squelch_max_selected_peers=2
     Network network_;
 
 public:
-    reduce_relay_test()
+    base_squelch_test()
         : env_(*this, jtx::envconfig([](std::unique_ptr<Config> cfg) {
             cfg->vpReduceRelayBaseSquelchEnable = true;
             cfg->vpReduceRelaySquelchMaxSelectedPeers = 6;
@@ -1680,7 +1712,7 @@ public:
     }
 };
 
-class reduce_relay_simulate_test : public reduce_relay_test
+class base_squelch_simulate_test : public base_squelch_test
 {
     void
     testRandom(bool log)
@@ -1696,7 +1728,7 @@ class reduce_relay_simulate_test : public reduce_relay_test
     }
 };
 
-BEAST_DEFINE_TESTSUITE(reduce_relay, overlay, xrpl);
-BEAST_DEFINE_TESTSUITE_MANUAL(reduce_relay_simulate, overlay, xrpl);
+BEAST_DEFINE_TESTSUITE(base_squelch, overlay, xrpl);
+BEAST_DEFINE_TESTSUITE_MANUAL(base_squelch_simulate, overlay, xrpl);
 
 }  // namespace xrpl::test

src/test/overlay/enhanced_squelch_test.cpp

@@ -0,0 +1,874 @@
+#include <test/jtx/Env.h>
+
+#include <xrpld/core/Config.h>
+#include <xrpld/overlay/Peer.h>
+#include <xrpld/overlay/ReduceRelayCommon.h>
+#include <xrpld/overlay/Slot.h>
+
+#include <xrpl/basics/UnorderedContainers.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/chrono.h>
+#include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/core/ServiceRegistry.h>
+#include <xrpl/protocol/KeyType.h>
+#include <xrpl/protocol/PublicKey.h>
+#include <xrpl/protocol/SecretKey.h>
+#include <xrpl/protocol/digest.h>
+
+#include <chrono>
+#include <cstdint>
+#include <functional>
+#include <optional>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+namespace xrpl::test {
+
+class TestHandler : public reduce_relay::SquelchHandler
+{
+public:
+    using squelch_method = std::function<void(PublicKey const&, Peer::id_t, std::uint32_t)>;
+    using squelchAll_method =
+        std::function<void(PublicKey const&, std::uint32_t, std::function<void(Peer::id_t)>)>;
+    using unsquelch_method = std::function<void(PublicKey const&, Peer::id_t)>;
+
+    squelch_method squelchF;
+    squelchAll_method squelchAllF;
+    unsquelch_method unsquelchF;
+
+    TestHandler(squelch_method squelchF, squelchAll_method squelchAllF, unsquelch_method unsquelchF)
+        : squelchF(std::move(squelchF))
+        , squelchAllF(std::move(squelchAllF))
+        , unsquelchF(std::move(unsquelchF))
+    {
+    }
+
+    TestHandler(TestHandler& copy)
+        : squelchF(copy.squelchF), squelchAllF(copy.squelchAllF), unsquelchF(copy.unsquelchF)
+    {
+    }
+
+    void
+    squelch(PublicKey const& validator, Peer::id_t peer, std::uint32_t duration) const override
+    {
+        squelchF(validator, peer, duration);
+    }
+
+    void
+    squelchAll(
+        PublicKey const& validator,
+        std::uint32_t duration,
+        std::function<void(Peer::id_t)> callback) override
+    {
+        squelchAllF(validator, duration, callback);
+    }
+
+    void
+    unsquelch(PublicKey const& validator, Peer::id_t peer) const override
+    {
+        unsquelchF(validator, peer);
+    }
+};
+
+class EnhancedSquelchingTestSlots : public reduce_relay::Slots
+{
+    using Slots = reduce_relay::Slots;
+
+public:
+    EnhancedSquelchingTestSlots(
+        ServiceRegistry& registry,
+        reduce_relay::SquelchHandler& handler,
+        Config const& config,
+        reduce_relay::Slots::clock_type& clock)
+        : Slots(registry, handler, config, clock)
+    {
+    }
+
+    Slots::slots_map const&
+    getSlots(bool trusted) const
+    {
+        if (trusted)
+            return trustedSlots_;
+
+        return untrustedSlots_;
+    }
+
+    hash_map<PublicKey, ValidatorInfo> const&
+    getConsideredValidators()
+    {
+        return consideredValidators_;
+    }
+
+    using Slots::updateConsideredValidator;
+
+    void
+    squelchValidator(PublicKey const& validatorKey, Peer::id_t peerID)
+    {
+        Slots::registerSquelchedValidator(validatorKey, peerID);
+    }
+
+    bool
+    validatorSquelched(PublicKey const& validatorKey)
+    {
+        return Slots::expireAndIsValidatorSquelched(validatorKey);
+    }
+
+    bool
+    peerSquelched(PublicKey const& validatorKey, Peer::id_t peerID)
+    {
+        return Slots::expireAndIsPeerSquelched(validatorKey, peerID);
+    }
+};
+
+class enhanced_squelch_test : public beast::unit_test::Suite
+{
+public:
+    TestHandler::squelch_method noopSquelch = [&](PublicKey const&, Peer::id_t, std::uint32_t) {
+        BEAST_EXPECTS(false, "unexpected call to squelch handler");
+    };
+
+    TestHandler::squelchAll_method noopSquelchAll =
+        [&](PublicKey const&, std::uint32_t, std::function<void(Peer::id_t)>) {
+            BEAST_EXPECTS(false, "unexpected call to squelchAll handler");
+        };
+
+    TestHandler::unsquelch_method noopUnsquelch = [&](PublicKey const&, Peer::id_t) {
+        BEAST_EXPECTS(false, "unexpected call to unsquelch handler");
+    };
+
+    // noop_handler is passed as a place holder Handler to slots
+    TestHandler noopHandler = {
+        noopSquelch,
+        noopSquelchAll,
+        noopUnsquelch,
+    };
+
+    jtx::Env env;
+
+    enhanced_squelch_test() : env(*this)
+    {
+        env.app().config().vpReduceRelayEnhancedSquelchEnable = true;
+    }
+
+    void
+    testConfig()
+    {
+        testcase("Test Config - enabled enhanced squelching");
+        Config c;
+
+        std::string toLoad(R"xrpldConfig(
+[reduce_relay]
+vp_enhanced_squelch_enable=1
+)xrpldConfig");
+
+        c.loadFromString(toLoad);
+        BEAST_EXPECT(c.vpReduceRelayEnhancedSquelchEnable == true);
+
+        toLoad = R"xrpldConfig(
+[reduce_relay]
+vp_enhanced_squelch_enable=0
+)xrpldConfig";
+
+        c.loadFromString(toLoad);
+        BEAST_EXPECT(c.vpReduceRelayEnhancedSquelchEnable == false);
+
+        toLoad = R"xrpldConfig(
+[reduce_relay]
+)xrpldConfig";
+
+        c.loadFromString(toLoad);
+        BEAST_EXPECT(c.vpReduceRelayEnhancedSquelchEnable == false);
+    }
+
+    /**
+     * Tests tracking for squelched validators and peers
+     */
+    void
+    testSquelchTracking()
+    {
+        testcase("squelchTracking");
+        Peer::id_t const squelchedPeerID = 0;
+        Peer::id_t const newPeerID = 1;
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), noopHandler, env.app().config(), stopwatch);
+        auto const publicKey = randomKeyPair(KeyType::Ed25519).first;
+
+        // a new key should not be squelched
+        BEAST_EXPECTS(!slots.validatorSquelched(publicKey), "validator squelched");
+
+        slots.squelchValidator(publicKey, squelchedPeerID);
+
+        // after squelching a peer, the validator must be squelched
+        BEAST_EXPECTS(slots.validatorSquelched(publicKey), "validator not squelched");
+
+        // the peer must also be squelched
+        BEAST_EXPECTS(slots.peerSquelched(publicKey, squelchedPeerID), "peer not squelched");
+
+        // a new peer must not be squelched
+        BEAST_EXPECTS(!slots.peerSquelched(publicKey, newPeerID), "new peer squelched");
+
+        // advance the manual clock to after expiration
+        stopwatch.advance(reduce_relay::kMaxUnsquelchExpireDefault + std::chrono::seconds{11});
+
+        // validator squelch should expire
+        BEAST_EXPECTS(!slots.validatorSquelched(publicKey), "validator squelched after expiry");
+
+        // peer squelch should also expire
+        BEAST_EXPECTS(
+            !slots.peerSquelched(publicKey, squelchedPeerID), "validator squelched after expiry");
+    }
+
+    void
+    testUpdateValidatorSlotNewValidator()
+    {
+        testcase("updateValidatorSlot_newValidator");
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), noopHandler, env.app().config(), stopwatch);
+
+        Peer::id_t const peerID = 1;
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+        uint256 const message{0};
+
+        slots.updateUntrustedValidatorSlot(message, validator, peerID);
+
+        // adding untrusted slot does not effect trusted slots
+        BEAST_EXPECTS(slots.getSlots(true).empty(), "trusted slots changed");
+
+        // we expect that the validator was not added to untrusted slots
+        BEAST_EXPECTS(slots.getSlots(false).empty(), "untrusted slot changed");
+
+        // we expect that the validator was added to th consideration list
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().contains(validator),
+            "new validator was not considered");
+    }
+
+    void
+    testUpdateValidatorSlotSquelchedValidator()
+    {
+        testcase("testUpdateValidatorSlot_squelchedValidator");
+
+        Peer::id_t const squelchedPeerID = 0;
+        Peer::id_t const newPeerID = 1;
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+
+        TestHandler::squelch_method const squelchF =
+            [&](PublicKey const& key, Peer::id_t id, std::uint32_t duration) {
+                BEAST_EXPECTS(key == validator, "squelch called for unknown validator key");
+
+                BEAST_EXPECTS(id == newPeerID, "squelch called for the wrong peer");
+            };
+
+        TestHandler handler{squelchF, noopSquelchAll, noopUnsquelch};
+
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), handler, env.app().config(), stopwatch);
+
+        slots.squelchValidator(validator, squelchedPeerID);
+
+        // this should not trigger squelch assertions, the peer is squelched
+        slots.updateUntrustedValidatorSlot(sha512Half(validator), validator, squelchedPeerID);
+
+        slots.updateUntrustedValidatorSlot(sha512Half(validator), validator, newPeerID);
+
+        // the squelched peer remained squelched
+        BEAST_EXPECTS(slots.peerSquelched(validator, squelchedPeerID), "peer not squelched");
+
+        // because the validator was squelched, the new peer was also squelched
+        BEAST_EXPECTS(slots.peerSquelched(validator, newPeerID), "new peer was not squelched");
+
+        // a squelched validator must not be considered
+        BEAST_EXPECTS(
+            !slots.getConsideredValidators().contains(validator),
+            "squelched validator was added for consideration");
+    }
+
+    void
+    testUpdateValidatorSlotSlotsFull()
+    {
+        testcase("updateValidatorSlot_slotsFull");
+        Peer::id_t const peerID = 1;
+
+        // while there are open untrusted slots, no calls should be made to
+        // squelch any validators
+        TestHandler handler{noopHandler};
+
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), handler, env.app().config(), stopwatch);
+
+        // saturate validator slots
+        auto const validators = fillUntrustedSlots(slots);
+
+        // adding untrusted slot does not effect trusted slots
+        BEAST_EXPECTS(slots.getSlots(true).empty(), "trusted slots changed");
+
+        // simulate additional messages from already selected validators
+        for (auto const& validator : validators)
+        {
+            for (int i = 0; i < reduce_relay::kMaxMessageThreshold; ++i)
+            {
+                slots.updateUntrustedValidatorSlot(
+                    sha512Half(validator) + static_cast<uint256>(i), validator, peerID);
+            }
+        }
+
+        // an untrusted slot was added for each validator
+        BEAST_EXPECT(slots.getSlots(false).size() == reduce_relay::kMaxUntrustedSlots);
+
+        for (auto const& validator : validators)
+            BEAST_EXPECTS(!slots.validatorSquelched(validator), "selected validator was squelched");
+
+        auto const newValidator = randomKeyPair(KeyType::Ed25519).first;
+
+        // once slots are full squelchAll must be called for new peer/validator
+        handler.squelchAllF =
+            [&](PublicKey const& key, std::uint32_t, std::function<void(Peer::id_t)> callback) {
+                BEAST_EXPECTS(key == newValidator, "unexpected validator squelched");
+                callback(peerID);
+            };
+
+        slots.updateUntrustedValidatorSlot(sha512Half(newValidator), newValidator, peerID);
+
+        // Once the slots are saturated every other validator is squelched
+        BEAST_EXPECTS(slots.validatorSquelched(newValidator), "untrusted validator not squelched");
+
+        BEAST_EXPECTS(
+            slots.peerSquelched(newValidator, peerID),
+            "peer for untrusted validator not squelched");
+    }
+
+    void
+    testDeleteIdlePeersDeleteIdleSlots()
+    {
+        testcase("deleteIdlePeers");
+        TestHandler handler{noopHandler};
+        TestStopwatch stopwatch;
+
+        EnhancedSquelchingTestSlots slots(env.app(), handler, env.app().config(), stopwatch);
+        auto keys = fillUntrustedSlots(slots);
+
+        //  verify that squelchAll is called for each idled slot validator
+        handler.squelchAllF = [&](PublicKey const& actualKey,
+                                  std::uint32_t duration,
+                                  std::function<void(Peer::id_t)> callback) {
+            for (auto it = keys.begin(); it != keys.end(); ++it)
+            {
+                if (*it == actualKey)
+                {
+                    keys.erase(it);
+                    return;
+                }
+            }
+            BEAST_EXPECTS(false, "unexpected key passed to squelchAll");
+        };
+
+        BEAST_EXPECTS(
+            slots.getSlots(false).size() == reduce_relay::kMaxUntrustedSlots,
+            "unexpected number of untrusted slots");
+
+        // advance the manual clock to after slot expiration
+        stopwatch.advance(reduce_relay::kMaxUnsquelchExpireDefault + std::chrono::seconds{1});
+
+        slots.deleteIdlePeers();
+
+        BEAST_EXPECTS(slots.getSlots(false).empty(), "unexpected number of untrusted slots");
+
+        BEAST_EXPECTS(keys.empty(), "not all validators were squelched");
+    }
+
+    void
+    testDeleteIdlePeersDeleteIdleUntrustedPeer()
+    {
+        testcase("deleteIdleUntrustedPeer");
+        Peer::id_t const peerID = 1;
+        Peer::id_t const peerID2 = 2;
+        TestStopwatch stopwatch;
+
+        EnhancedSquelchingTestSlots slots(env.app(), noopHandler, env.app().config(), stopwatch);
+
+        // fill one untrusted validator slot
+        auto const validator = fillUntrustedSlots(slots, 1)[0];
+
+        BEAST_EXPECTS(slots.getSlots(false).size() == 1, "unexpected number of untrusted slots");
+
+        slots.updateSlotAndSquelch(
+            sha512Half(validator) + static_cast<uint256>(100), validator, peerID, false);
+
+        slots.updateSlotAndSquelch(
+            sha512Half(validator) + static_cast<uint256>(100), validator, peerID2, false);
+
+        slots.deletePeer(peerID, true);
+
+        auto const slotPeers = getUntrustedSlotPeers(validator, slots);
+        BEAST_EXPECTS(slotPeers.size() == 1, "untrusted validator slot is missing");
+
+        BEAST_EXPECTS(!slotPeers.contains(peerID), "peer was not removed from untrusted slots");
+
+        BEAST_EXPECTS(slotPeers.contains(peerID2), "peer was removed from untrusted slots");
+    }
+
+    /**
+     * Test that untrusted validator slots are correctly updated by
+     * updateSlotAndSquelch
+     */
+    void
+    testUpdateSlotAndSquelchUntrustedValidator()
+    {
+        testcase("updateUntrsutedValidatorSlot");
+        TestHandler handler{noopHandler};
+
+        handler.squelchF = [](PublicKey const&, Peer::id_t, std::uint32_t) {};
+
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), handler, env.app().config(), stopwatch);
+
+        // peers that will be source of validator messages
+        std::vector<Peer::id_t> peers = {};
+
+        // prepare n+1 peers, we expect the n+1st peer will be squelched
+        peers.reserve(env.app().config().vpReduceRelaySquelchMaxSelectedPeers + 1);
+        for (int i = 0; i < env.app().config().vpReduceRelaySquelchMaxSelectedPeers + 1; ++i)
+            peers.push_back(i);
+
+        auto const validator = fillUntrustedSlots(slots, 1)[0];
+
+        // Squelching logic resets all counters each time a new peer is added
+        // Therfore we need to populate counters for each peer before sending
+        // new messages
+        for (auto const& peer : peers)
+        {
+            auto const now = stopwatch.now();
+            slots.updateSlotAndSquelch(
+                sha512Half(validator) + static_cast<uint256>(now.time_since_epoch().count()),
+                validator,
+                peer,
+                false);
+
+            stopwatch.advance(std::chrono::milliseconds{10});
+        }
+
+        // simulate new, unique validator messages sent by peers
+        for (auto const& peer : peers)
+        {
+            for (int i = 0; i < reduce_relay::kMaxMessageThreshold + 1; ++i)
+            {
+                auto const now = stopwatch.now();
+                slots.updateSlotAndSquelch(
+                    sha512Half(validator) + static_cast<uint256>(now.time_since_epoch().count()),
+                    validator,
+                    peer,
+                    false);
+
+                stopwatch.advance(std::chrono::milliseconds{10});
+            }
+        }
+
+        auto const slotPeers = getUntrustedSlotPeers(validator, slots);
+        BEAST_EXPECTS(
+            slotPeers.size() == env.app().config().vpReduceRelaySquelchMaxSelectedPeers + 1,
+            "untrusted validator slot is missing");
+
+        int selected = 0;
+        int squelched = 0;
+        for (auto const& [_, info] : slotPeers)
+        {
+            switch (info.state)
+            {
+                case reduce_relay::PeerState::Selected:
+                    ++selected;
+                    break;
+                case reduce_relay::PeerState::Squelched:
+                    ++squelched;
+                    break;
+                case reduce_relay::PeerState::Counting:
+                    BEAST_EXPECTS(false, "peer should not be in counting state");
+            }
+        }
+
+        BEAST_EXPECTS(squelched == 1, "expected one squelched peer");
+        BEAST_EXPECTS(
+            selected == env.app().config().vpReduceRelaySquelchMaxSelectedPeers,
+            "wrong number of peers selected");
+    }
+
+    void
+    testUpdateConsideredValidatorNew()
+    {
+        testcase("testUpdateConsideredValidator_new");
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), noopHandler, env.app().config(), stopwatch);
+
+        // insert some random validator key
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+        Peer::id_t const peerID = 0;
+        Peer::id_t const peerID2 = 1;
+
+        BEAST_EXPECTS(
+            !slots.updateConsideredValidator(validator, peerID),
+            "validator was selected with insufficient number of peers");
+
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().contains(validator),
+            "new validator was not added for consideration");
+
+        BEAST_EXPECTS(
+            !slots.updateConsideredValidator(validator, peerID),
+            "validator was selected with insufficient number of peers");
+
+        // expect that a peer will be registered once as a message source
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().at(validator).peers.size() == 1,
+            "duplicate peer was registered");
+
+        BEAST_EXPECTS(
+            !slots.updateConsideredValidator(validator, peerID2),
+            "validator was selected with insufficient number of peers");
+
+        // expect that each distinct peer will be registered
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().at(validator).peers.size() == 2,
+            "distinct peers were not registered");
+    }
+
+    void
+    testUpdateConsideredValidatorIdle()
+    {
+        testcase("testUpdateConsideredValidator_idle");
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), noopHandler, env.app().config(), stopwatch);
+
+        // insert some random validator key
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+        Peer::id_t const peerID = 0;
+
+        BEAST_EXPECTS(
+            !slots.updateConsideredValidator(validator, peerID),
+            "validator was selected with insufficient number of peers");
+
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().contains(validator),
+            "new validator was not added for consideration");
+
+        auto const state = slots.getConsideredValidators().at(validator);
+
+        // simulate a validator sending a new message before the idle timer
+        stopwatch.advance(reduce_relay::kIdled - std::chrono::seconds(1));
+
+        BEAST_EXPECTS(
+            !slots.updateConsideredValidator(validator, peerID),
+            "validator was selected with insufficient number of peers");
+        auto const newState = slots.getConsideredValidators().at(validator);
+
+        BEAST_EXPECTS(state.count + 1 == newState.count, "non-idling validator was updated");
+
+        // simulate a validator idling
+        stopwatch.advance(reduce_relay::kIdled + std::chrono::seconds(1));
+
+        BEAST_EXPECTS(
+            !slots.updateConsideredValidator(validator, peerID),
+            "validator was selected with insufficient number of peers");
+    }
+
+    void
+    testUpdateConsideredValidatorSelectQualifying()
+    {
+        testcase("testUpdateConsideredValidator_selectQualifying");
+
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), noopHandler, env.app().config(), stopwatch);
+
+        // insert some random validator key
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+        Peer::id_t const peerID = 0;
+
+        for (int i = 0; i < reduce_relay::kMaxMessageThreshold - 1; ++i)
+        {
+            BEAST_EXPECTS(
+                !slots.updateConsideredValidator(validator, peerID),
+                "validator was selected before reaching message threshold");
+
+            stopwatch.advance(reduce_relay::kIdled - std::chrono::seconds(1));
+        }
+
+        auto const consideredValidator = slots.updateConsideredValidator(validator, peerID);
+        BEAST_EXPECTS(
+            consideredValidator && *consideredValidator == validator,
+            "expected validator was not selected");
+
+        // expect that selected peer was removed
+        BEAST_EXPECTS(
+            !slots.getConsideredValidators().contains(validator),
+            "selected validator was not removed from considered list");
+    }
+
+    void
+    testCleanConsideredValidatorsResetIdle()
+    {
+        testcase("testCleanConsideredValidators_resetIdle");
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+
+        TestStopwatch stopwatch;
+
+        EnhancedSquelchingTestSlots slots(env.app(), noopHandler, env.app().config(), stopwatch);
+
+        // send enough messages for a slot to meet peer requirements
+        for (int i = 0; i < env.app().config().vpReduceRelaySquelchMaxSelectedPeers; ++i)
+        {
+            slots.updateUntrustedValidatorSlot(
+                sha512Half(validator) + static_cast<uint256>(i), validator, i);
+        }
+
+        // send enough messages from some peer to be one message away from
+        // meeting the selection criteria
+        for (int i = 0; i < reduce_relay::kMaxMessageThreshold -
+                 (env.app().config().vpReduceRelaySquelchMaxSelectedPeers + 1);
+             ++i)
+        {
+            slots.updateUntrustedValidatorSlot(
+                sha512Half(validator) + static_cast<uint256>(i), validator, 0);
+        }
+
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().at(validator).count ==
+                reduce_relay::kMaxMessageThreshold - 1,
+            "considered validator information is in an invalid state");
+
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().at(validator).peers.size() ==
+                env.app().config().vpReduceRelaySquelchMaxSelectedPeers,
+            "considered validator information is in an invalid state");
+
+        stopwatch.advance(reduce_relay::kIdled + std::chrono::seconds{1});
+
+        // deleteIdlePeers must reset the progress of a validator that idled
+        slots.deleteIdlePeers();
+
+        slots.updateUntrustedValidatorSlot(
+            sha512Half(validator) + static_cast<uint256>(1), validator, 0);
+
+        // we expect that the validator was not selected
+        BEAST_EXPECTS(slots.getSlots(false).empty(), "untrusted slot was created");
+
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().at(validator).count == 1,
+            "considered validator information is in an invalid state");
+
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().at(validator).peers.size() == 1,
+            "considered validator information is in an invalid state");
+    }
+
+    void
+    testCleanConsideredValidatorsDeletePoorlyConnected()
+    {
+        testcase("cleanConsideredValidators_deletePoorlyConnected");
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+        Peer::id_t const peerID = 0;
+        TestHandler handler{noopHandler};
+
+        //  verify that squelchAll is called for poorly connected validator
+        handler.squelchAllF = [&](PublicKey const& actualKey,
+                                  std::uint32_t duration,
+                                  std::function<void(Peer::id_t)> callback) {
+            BEAST_EXPECTS(actualKey == validator, "unexpected key passed to squelchAll");
+            callback(peerID);
+        };
+
+        TestStopwatch stopwatch;
+
+        EnhancedSquelchingTestSlots slots(env.app(), handler, env.app().config(), stopwatch);
+
+        // send enough messages from a single peer
+        for (int i = 0; i < (2 * reduce_relay::kMaxMessageThreshold) + 1; ++i)
+        {
+            slots.updateUntrustedValidatorSlot(
+                sha512Half(validator) + static_cast<uint256>(i), validator, peerID);
+        }
+
+        stopwatch.advance(reduce_relay::kIdled + std::chrono::seconds{1});
+
+        // deleteIdlePeers must squelch the validator as it failed to reach
+        // peering requirements
+        slots.deleteIdlePeers();
+
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().empty(), "poorly connected validator was not deleted");
+    }
+
+    void
+    testCleanConsideredValidatorsDeleteSilent()
+    {
+        testcase("cleanConsideredValidators_deleteSilent");
+        // insert some random validator key
+        auto const idleValidator = randomKeyPair(KeyType::Ed25519).first;
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+        Peer::id_t const peerID = 0;
+
+        TestHandler handler{noopHandler};
+
+        //  verify that squelchAll is called for idle validator
+        handler.squelchAllF = [&](PublicKey const& actualKey,
+                                  std::uint32_t duration,
+                                  std::function<void(Peer::id_t)> callback) {
+            BEAST_EXPECTS(actualKey == idleValidator, "unexpected key passed to squelchAll");
+            callback(peerID);
+        };
+
+        TestStopwatch stopwatch;
+
+        EnhancedSquelchingTestSlots slots(env.app(), handler, env.app().config(), stopwatch);
+
+        BEAST_EXPECTS(
+            !slots.updateConsideredValidator(idleValidator, peerID),
+            "validator was selected with insufficient number of peers");
+
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().contains(idleValidator),
+            "new validator was not added for consideration");
+
+        // simulate a validator idling
+        stopwatch.advance(reduce_relay::kMaxUntrustedValidatorIdle + std::chrono::seconds(1));
+        BEAST_EXPECTS(
+            !slots.updateConsideredValidator(validator, peerID),
+            "validator was selected with insufficient number of peers");
+
+        slots.deleteIdlePeers();
+
+        BEAST_EXPECTS(
+            !slots.getConsideredValidators().contains(idleValidator),
+            "late validator was not removed");
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().contains(validator), "timely validator was removed");
+    }
+
+    void
+    testSquelchUntrustedValidatorConsideredListCleared()
+    {
+        testcase("testSquelchUntrustedValidator");
+
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+        Peer::id_t const peerID = 0;
+
+        TestHandler handler{noopHandler};
+        //  verify that squelchAll is called for idle validator
+        handler.squelchAllF = [&](PublicKey const& actualKey,
+                                  std::uint32_t duration,
+                                  std::function<void(Peer::id_t)> callback) {
+            BEAST_EXPECTS(actualKey == validator, "unexpected key passed to squelchAll");
+        };
+
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), handler, env.app().config(), stopwatch);
+
+        // add the validator to the considered list
+        slots.updateUntrustedValidatorSlot(sha512Half(validator), validator, peerID);
+
+        BEAST_EXPECTS(
+            slots.getConsideredValidators().contains(validator),
+            "validator was not added to considered list");
+
+        slots.squelchUntrustedValidator(validator);
+
+        BEAST_EXPECTS(
+            !slots.getConsideredValidators().contains(validator),
+            "validator was not removed from considered list");
+    }
+
+    void
+    testSquelchUntrustedValidatorSlotEvicted()
+    {
+        testcase("testSquelchUntrustedValidator_slotEvicted");
+
+        TestHandler handler{noopHandler};
+        TestStopwatch stopwatch;
+        EnhancedSquelchingTestSlots slots(env.app(), handler, env.app().config(), stopwatch);
+
+        // assign a slot to the untrusted validator
+        auto const validators = fillUntrustedSlots(slots, 1);
+
+        //  verify that squelchAll is called for idle validator
+        handler.squelchAllF = [&](PublicKey const& actualKey,
+                                  std::uint32_t duration,
+                                  std::function<void(Peer::id_t)> callback) {
+            BEAST_EXPECTS(actualKey == validators[0], "unexpected key passed to squelchAll");
+        };
+
+        BEAST_EXPECTS(
+            slots.getSlots(false).contains(validators[0]),
+            "a slot was not assigned to a validator");
+
+        slots.squelchUntrustedValidator(validators[0]);
+
+        BEAST_EXPECTS(!slots.getSlots(false).contains(validators[0]), "a slot was not evicted");
+    }
+
+private:
+    /**
+     * A helper method to fill untrusted slots of a given Slots instance
+     * with random validator messages
+     */
+    std::vector<PublicKey>
+    fillUntrustedSlots(
+        EnhancedSquelchingTestSlots& slots,
+        int64_t maxSlots = reduce_relay::kMaxUntrustedSlots)
+    {
+        std::vector<PublicKey> keys;
+        for (int i = 0; i < maxSlots; ++i)
+        {
+            auto const validator = randomKeyPair(KeyType::Ed25519).first;
+            keys.push_back(validator);
+            for (int j = 0; j < env.app().config().vpReduceRelaySquelchMaxSelectedPeers; ++j)
+            {
+                // send enough messages so that a validator slot is selected
+                for (int k = 0; k < reduce_relay::kMaxMessageThreshold; ++k)
+                {
+                    slots.updateUntrustedValidatorSlot(
+                        sha512Half(validator) + static_cast<uint256>(k), validator, j);
+                }
+            }
+        }
+
+        return keys;
+    }
+
+    static std::unordered_map<Peer::id_t, reduce_relay::Slot::PeerInfo>
+    getUntrustedSlotPeers(PublicKey const& validator, EnhancedSquelchingTestSlots const& slots)
+    {
+        auto const& it = slots.getSlots(false).find(validator);
+        if (it == slots.getSlots(false).end())
+            return {};
+
+        auto r = std::unordered_map<Peer::id_t, reduce_relay::Slot::PeerInfo>();
+
+        for (auto const& [id, info] : it->second.getPeers())
+            r.emplace(id, info);
+
+        return r;
+    }
+
+    void
+    run() override
+    {
+        testConfig();
+        testSquelchTracking();
+        testUpdateValidatorSlotNewValidator();
+        testUpdateValidatorSlotSlotsFull();
+        testUpdateValidatorSlotSquelchedValidator();
+        testDeleteIdlePeersDeleteIdleSlots();
+        testDeleteIdlePeersDeleteIdleUntrustedPeer();
+        testUpdateSlotAndSquelchUntrustedValidator();
+        testUpdateConsideredValidatorNew();
+        testUpdateConsideredValidatorIdle();
+        testUpdateConsideredValidatorSelectQualifying();
+        testCleanConsideredValidatorsDeleteSilent();
+        testCleanConsideredValidatorsResetIdle();
+        testCleanConsideredValidatorsDeletePoorlyConnected();
+        testSquelchUntrustedValidatorConsideredListCleared();
+        testSquelchUntrustedValidatorSlotEvicted();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(enhanced_squelch, overlay, xrpl);
+
+}  // namespace xrpl::test

src/test/overlay/squelch_store_test.cpp

@@ -0,0 +1,133 @@
+#include <test/jtx/Env.h>
+
+#include <xrpld/overlay/ReduceRelayCommon.h>
+#include <xrpld/overlay/SquelchStore.h>
+
+#include <xrpl/basics/UnorderedContainers.h>
+#include <xrpl/basics/chrono.h>
+#include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/protocol/KeyType.h>
+#include <xrpl/protocol/PublicKey.h>
+#include <xrpl/protocol/SecretKey.h>
+
+#include <chrono>
+
+namespace xrpl::test {
+
+class TestSquelchStore : public reduce_relay::SquelchStore
+{
+public:
+    TestSquelchStore(beast::Journal journal, TestStopwatch& clock)
+        : reduce_relay::SquelchStore(journal, clock)
+    {
+    }
+
+    hash_map<PublicKey, TestStopwatch::time_point> const&
+    getSquelched() const
+    {
+        return squelched_;
+    }
+};
+
+class squelch_store_test : public beast::unit_test::Suite
+{
+    using seconds = std::chrono::seconds;
+
+public:
+    jtx::Env env;
+
+    squelch_store_test() : env(*this)
+    {
+    }
+
+    void
+    testHandleSquelch()
+    {
+        testcase("SquelchStore handleSquelch");
+
+        TestStopwatch clock;
+        auto store = TestSquelchStore(env.journal, clock);
+
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+
+        // attempt to squelch the peer with a too small duration
+        store.handleSquelch(validator, true, reduce_relay::kMinUnsquelchExpire - seconds{1});
+
+        // the peer must not be squelched
+        BEAST_EXPECTS(!store.isSquelched(validator), "peer is squelched");
+
+        // attempt to squelch the peer with a too big duration
+        store.handleSquelch(validator, true, reduce_relay::kMaxUnsquelchExpirePeers + seconds{1});
+
+        // the peer must not be squelched
+        BEAST_EXPECTS(!store.isSquelched(validator), "peer is squelched");
+
+        // squelch the peer with a good duration
+        store.handleSquelch(validator, true, reduce_relay::kMinUnsquelchExpire + seconds{1});
+
+        // the peer for the validator should be squelched
+        BEAST_EXPECTS(store.isSquelched(validator), "peer and validator are not squelched");
+
+        // unsquelch the validator
+        store.handleSquelch(validator, false, seconds{0});
+
+        BEAST_EXPECTS(!store.isSquelched(validator), "peer is squelched");
+    }
+
+    void
+    testIsSquelched()
+    {
+        testcase("SquelchStore IsSquelched");
+        TestStopwatch clock;
+        auto store = TestSquelchStore(env.journal, clock);
+
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+        auto const duration = reduce_relay::kMinUnsquelchExpire + seconds{1};
+
+        store.handleSquelch(validator, true, reduce_relay::kMinUnsquelchExpire + seconds{1});
+        BEAST_EXPECTS(store.isSquelched(validator), "peer and validator are not squelched");
+
+        clock.advance(duration + seconds{1});
+
+        // the peer with short squelch duration must be not squelched
+        BEAST_EXPECTS(!store.isSquelched(validator), "peer and validator are squelched");
+    }
+
+    void
+    testClearExpiredSquelches()
+    {
+        testcase("SquelchStore testClearExpiredSquelches");
+        TestStopwatch clock;
+        auto store = TestSquelchStore(env.journal, clock);
+
+        auto const validator = randomKeyPair(KeyType::Ed25519).first;
+        auto const duration = reduce_relay::kMinUnsquelchExpire + seconds{1};
+        store.handleSquelch(validator, true, duration);
+        BEAST_EXPECTS(
+            store.getSquelched().size() == 1, "validators were not registered in the store");
+
+        clock.advance(duration + seconds{1});
+
+        auto const validator2 = randomKeyPair(KeyType::Ed25519).first;
+        auto const duration2 = reduce_relay::kMinUnsquelchExpire + seconds{2};
+        store.handleSquelch(validator2, true, duration2);
+
+        BEAST_EXPECTS(!store.getSquelched().contains(validator), "expired squelch was not deleted");
+
+        BEAST_EXPECTS(
+            store.getSquelched().contains(validator2),
+            "validators were not registered in the store");
+    }
+    void
+    run() override
+    {
+        testHandleSquelch();
+        testIsSquelched();
+        testClearExpiredSquelches();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(squelch_store, overlay, xrpl);
+
+}  // namespace xrpl::test

src/xrpld/core/Config.h

@@ -260,6 +260,9 @@ public:
      * //////////////    END OF TEMPORARY CODE BLOCK    /////////////////////
      */
 
+    // Enable enhanced squelching of unique untrusted validator messages
+    bool vpReduceRelayEnhancedSquelchEnable = false;
+
     // Transaction reduce-relay feature
     bool txReduceRelayEnable = false;
     // If tx reduce-relay feature is disabled

src/xrpld/core/detail/Config.cpp

@@ -845,6 +845,8 @@ Config::loadFromString(std::string const& fileContents)
          * //////////////  !!END OF TEMPORARY CODE BLOCK!! /////////////////////
          */
 
+        vpReduceRelayEnhancedSquelchEnable = sec.valueOr(Keys::kVpEnhancedSquelchEnable, false);
+
         txReduceRelayEnable = sec.valueOr(Keys::kTxEnable, false);
         txReduceRelayMetrics = sec.valueOr(Keys::kTxMetrics, false);
         txReduceRelayMinPeers = sec.valueOr(Keys::kTxMinPeers, 20);

src/xrpld/overlay/README.md

@@ -233,7 +233,7 @@ synchronization software. For more on this topic, please visit [ntp.org](http://
 The mandatory `Public-Key` field identifies the sending server's public key,
 encoded in base58 using the standard encoding for node public keys.
 
-See: https://xrpl.org/base58-encodings.html
+See: <https://xrpl.org/base58-encodings.html>
 
 | Field Name      |      Request       |      Response      |
 | --------------- | :----------------: | :----------------: |
@@ -277,7 +277,7 @@ The field can take two values:
 - **`Private`**: The server's IP address and port should not be included in
   crawl reports. _This is the default, if the field is omitted._
 
-For more on the Peer Crawler, please visit https://xrpl.org/peer-crawler.html.
+For more on the Peer Crawler, please visit <https://xrpl.org/peer-crawler.html>.
 
 | Field Name      |      Request       |      Response      |
 | --------------- | :----------------: | :----------------: |
@@ -354,9 +354,52 @@ transferred between A and B and will not be able to intelligently tamper with th
 message stream between Alice and Bob, although she may be still be able to inject
 delays or terminate the link.
 
-# XRPL clustering
+## Peer-to-Peer Traffic Routing
 
-A cluster consists of more than one XRPL server under common
+### Squelching
+
+Validator Squelching is a network feature that reduces redundant message traffic by intelligently selecting a small subset of peers to listen to for each validator. Messages from non-selected peers are temporarily ignored, or "squelched." This process significantly cuts down on processing overhead and provides dynamic fault tolerance by allowing the network to ignore misbehaving peers without a permanent ban. The system continuously re-evaluates peer performance to adapt to changing network conditions.
+
+### Components
+
+The squelching architecture is built on five key classes:
+
+- `SquelchStore.h`: A low-level, timed key-value store that maps a validator to its squelch expiration timestamp.
+
+- `Slot.h/Slot`: Manages the state for a single validator, tracking all peers that relay its messages. It operates in a Counting state to gather peer performance data and a Selected state after choosing the best peers and squelching the rest. It handles peer disconnections and idleness to keep the selection optimal.
+
+- `Slot.h/Slots`: The central container that manages all active Slot instances. It applies different policies for trusted and untrusted validators, evaluates candidates for the limited untrusted slots, and runs periodic cleanup routines.
+
+- `OverlayImpl.h`: Integrates the squelching system with the network, capturing events and dispatching them to the Slots container in a thread-safe manner.
+
+- `PeerImp.h` - Handles squelch messages, and calls `OverlayImpl.h` when it received proposal or validation messages.
+
+### Component Dependency
+
+The component dependencies follow a clear hierarchy from the network layer down to individual peers:
+
+- `OverlayImpl`: The top-level component that owns a single instance of Slots, and a currently connected Peers.
+- `Slots`: This central orchestrator owns and manages a collection of many Slot instances.
+- `Slot`: Each Slot represents a single validator and manages the state of all PeerImp instances that relay messages for it.
+- `PeerImp`: Represents a connected peer and owns its own instance of SquelchStore to manage its local squelch state for various validators.
+
+### The Squelching Lifecycle
+
+When a message from a validator arrives, it is dispatched to the appropriate Slot. The Slot, initially in a Counting state, tracks message volume from each peer. Once enough data is gathered, it triggers a selection, randomly choosing a small number of the best-performing peers. The Slot then instructs a SquelchHandler to squelch all non-selected peers for a calculated duration and transitions to a Selected state. The system continuously monitors for network changes, such as a selected peer disconnecting, which causes the Slot to reset to the Counting state and begin a new evaluation.
+
+### Trusted vs. Untrusted Validators
+
+The system applies different policies based on validator trust status:
+
+- **Trusted Validators**: Are granted a Slot immediately to optimize traffic from known-good sources.
+
+- **Untrusted Validators**: Are handled more cautiously, especially when the Enhanced Squelching feature is enabled. They must compete for a fixed number of limited slots by first proving their reliability in a "consideration pool." Validators that fail to gain a slot or become idle are aggressively squelched across all peers. This can also be triggered by a network-wide consensus to ignore a specific untrusted validator.
+
+This dual-policy approach optimizes trusted traffic while robustly protecting the network from potentially malicious or unknown validators.
+
+## XRP Ledger Clustering
+
+A cluster consists of more than one XRP Ledger server under common
 administration that share load information, distribute cryptography
 operations, and provide greater response consistency.
 
@@ -366,7 +409,7 @@ Cluster nodes share information about their internal load status. Cluster
 nodes do not have to verify the cryptographic signatures on messages
 received from other cluster nodes.
 
-## Configuration
+### Configuration
 
 A server's public key can be determined from the output of the `server_info`
 command. The key is in the `pubkey_node` value, and is a text string
@@ -392,7 +435,7 @@ New spokes can be added as follows:
 - Restart each hub, one by one
 - Restart the spoke
 
-## Transaction Behavior
+### Transaction Behavior
 
 When a transaction is received from a cluster member, several normal checks
 are bypassed:
@@ -408,7 +451,7 @@ does not meet its current relay fee. It is preferable to keep the cluster
 in agreement and permit confirmation from one cluster member to more
 reliably indicate the transaction's acceptance by the cluster.
 
-## Server Load Information
+### Server Load Information
 
 Cluster members exchange information on their server's load level. The load
 level is essentially the amount by which the normal fee levels are multiplied
@@ -419,7 +462,7 @@ fee, is the highest of its local load level, the network load level, and the
 cluster load level. The cluster load level is the median load level reported
 by a cluster member.
 
-## Gossip
+### Gossip
 
 Gossip is the mechanism by which cluster members share information about
 endpoints (typically IPv4 addresses) that are imposing unusually high load
@@ -434,7 +477,7 @@ the servers in a cluster. With gossip, if he chooses to use the same IP
 address to impose load on more than one server, he will find that the amount
 of load he can impose before getting disconnected is much lower.
 
-## Monitoring
+### Monitoring
 
 The `peers` command will report on the status of the cluster. The `cluster`
 object will contain one entry for each member of the cluster (either configured

src/xrpld/overlay/ReduceRelayCommon.h

@@ -11,10 +11,10 @@
 namespace xrpl::reduce_relay {
 
 // Peer's squelch is limited in time to
-// rand{MIN_UNSQUELCH_EXPIRE, max_squelch},
+// rand{kMinUnsquelchExpire, max_squelch},
 // where max_squelch is
-// min(max(MAX_UNSQUELCH_EXPIRE_DEFAULT, SQUELCH_PER_PEER * number_of_peers),
-//     MAX_UNSQUELCH_EXPIRE_PEERS)
+// min(max(kMaxUnsquelchExpireDefault, kSquelchPerPeer * number_of_peers),
+//     kMaxUnsquelchExpirePeers)
 static constexpr auto kMinUnsquelchExpire = std::chrono::seconds{300};
 static constexpr auto kMaxUnsquelchExpireDefault = std::chrono::seconds{600};
 static constexpr auto kSquelchPerPeer = std::chrono::seconds(10);
@@ -29,6 +29,14 @@ static constexpr uint16_t kMinMessageThreshold = 19;
 static constexpr uint16_t kMaxMessageThreshold = 20;
 // Max selected peers to choose as the source of messages from validator
 static constexpr uint16_t kMaxSelectedPeers = 5;
+
+// Max number of untrusted slots the server will maintain
+static constexpr uint16_t kMaxUntrustedSlots = 30;
+
+// The maximum of seconds an untrusted validator can go without sending a
+// validation message. After this, a validator may be squelched
+static constexpr auto kMaxUntrustedValidatorIdle = std::chrono::seconds{30};
+
 // Wait before reduce-relay feature is enabled on boot up to let
 // the server establish peer connections
 static constexpr auto kWaitOnBootup = std::chrono::minutes{10};

src/xrpld/overlay/Squelch.h

@@ -1,106 +0,0 @@
-#pragma once
-
-#include <xrpld/overlay/ReduceRelayCommon.h>
-
-#include <xrpl/basics/Log.h>
-#include <xrpl/basics/UnorderedContainers.h>
-#include <xrpl/beast/utility/Journal.h>
-#include <xrpl/protocol/PublicKey.h>
-
-#include <chrono>
-
-namespace xrpl::reduce_relay {
-
-/**
- * Maintains squelching of relaying messages from validators
- */
-template <typename ClockType>
-class Squelch
-{
-    using time_point = ClockType::time_point;
-
-public:
-    explicit Squelch(beast::Journal journal) : journal_(journal)
-    {
-    }
-    virtual ~Squelch() = default;
-
-    /**
-     * Squelch validation/proposal relaying for the validator
-     * @param validator The validator's public key
-     * @param squelchDuration Squelch duration in seconds
-     * @return false if invalid squelch duration
-     */
-    bool
-    addSquelch(PublicKey const& validator, std::chrono::seconds const& squelchDuration);
-
-    /**
-     * Remove the squelch
-     * @param validator The validator's public key
-     */
-    void
-    removeSquelch(PublicKey const& validator);
-
-    /**
-     * Remove expired squelch
-     * @param validator Validator's public key
-     * @return true if removed or doesn't exist, false if still active
-     */
-    bool
-    expireSquelch(PublicKey const& validator);
-
-private:
-    /**
-     * Maintains the list of squelched relaying to downstream peers.
-     * Expiration time is included in the TMSquelch message.
-     */
-    hash_map<PublicKey, time_point> squelched_;
-    beast::Journal const journal_;
-};
-
-template <typename ClockType>
-bool
-Squelch<ClockType>::addSquelch(
-    PublicKey const& validator,
-    std::chrono::seconds const& squelchDuration)
-{
-    if (squelchDuration >= kMinUnsquelchExpire && squelchDuration <= kMaxUnsquelchExpirePeers)
-    {
-        squelched_[validator] = ClockType::now() + squelchDuration;
-        return true;
-    }
-
-    JLOG(journal_.error()) << "squelch: invalid squelch duration " << squelchDuration.count();
-
-    // unsquelch if invalid duration
-    removeSquelch(validator);
-
-    return false;
-}
-
-template <typename ClockType>
-void
-Squelch<ClockType>::removeSquelch(PublicKey const& validator)
-{
-    squelched_.erase(validator);
-}
-
-template <typename ClockType>
-bool
-Squelch<ClockType>::expireSquelch(PublicKey const& validator)
-{
-    auto now = ClockType::now();
-
-    auto const& it = squelched_.find(validator);
-    if (it == squelched_.end())
-        return true;
-    if (it->second > now)
-        return false;
-
-    // squelch expired
-    squelched_.erase(it);
-
-    return true;
-}
-
-}  // namespace xrpl::reduce_relay

src/xrpld/overlay/SquelchStore.h

@@ -0,0 +1,117 @@
+#pragma once
+
+#include <xrpl/basics/UnorderedContainers.h>
+#include <xrpl/beast/clock/abstract_clock.h>
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/protocol/PublicKey.h>
+
+#include <chrono>
+
+namespace xrpl::reduce_relay {
+
+/**
+ * @brief Manages the temporary suppression ("squelching") of validators.
+ *
+ * @details This class provides a mechanism to temporarily ignore messages from
+ * specific validators for a defined duration. It tracks which
+ * validators are currently squelched and handles the
+ * expiration of the squelch period. The use of an
+ * abstract clock allows for deterministic testing of time-based
+ * squelch logic.
+ */
+class SquelchStore
+{
+    using clock_type = beast::AbstractClock<std::chrono::steady_clock>;
+    using time_point = clock_type::time_point;
+
+public:
+    explicit SquelchStore(beast::Journal journal, clock_type& clock)
+        : journal_(journal), clock_(clock)
+    {
+    }
+    virtual ~SquelchStore() = default;
+
+    /**
+     * @brief Manages the squelch status of a validator.
+     *
+     * @details This function acts as the primary public interface for
+     * controlling a validator's squelch state. Based on the `squelch` flag, it
+     * either adds a new squelch entry for the specified duration or removes an
+     * existing one. This function also clears all expired squelches.
+     *
+     * @param validator The public key of the validator to manage.
+     * @param squelch If `true`, the validator will be squelched. If `false`,
+     * any existing squelch will be removed.
+     * @param duration The duration in seconds for the squelch. This value is
+     * only used when `squelch` is `true`.
+     */
+    void
+    handleSquelch(PublicKey const& validator, bool squelch, std::chrono::seconds duration);
+
+    /**
+     * @brief Checks if a validator is currently squelched.
+     *
+     * @details This function checks if the validator's squelch has expired.
+     *
+     * @param validator The public key of the validator to check.
+     * @return `true` if a non-expired squelch entry exists for the
+     * validator, `false` otherwise.
+     */
+    bool
+    isSquelched(PublicKey const& validator) const;
+
+    // The following field is protected for unit tests.
+protected:
+    /**
+     * @brief The core data structure mapping a validator's public key to the
+     * time point when their squelch expires.
+     */
+    hash_map<PublicKey, time_point> squelched_;
+
+private:
+    /**
+     * @brief Internal implementation to add or update a squelch entry.
+     *
+     * @details Calculates the expiration time point by adding the duration to
+     * the current time and inserts or overwrites the entry for the validator in
+     * the `squelched_` map.
+     *
+     * @param validator The public key of the validator to squelch.
+     * @param squelchDuration The duration for which the validator should be
+     * squelched.
+     */
+    void
+    add(PublicKey const& validator, std::chrono::seconds const& squelchDuration);
+
+    /**
+     * @brief Internal implementation to remove a squelch entry.
+     *
+     * @details Erases the squelch entry for the given validator from the
+     * `squelched_` map, effectively unsquelching it.
+     *
+     * @param validator The public key of the validator to unsquelch.
+     */
+    void
+    remove(PublicKey const& validator);
+
+    /**
+     * @brief Internal implementation to remove all expired squelches.
+     *
+     * @details Erases all squelch entries whose expiration is in the past.
+     */
+    void
+    removeExpired();
+
+    /**
+     * @brief The logging interface used by this store.
+     */
+    beast::Journal const journal_;
+
+    /**
+     * @brief A reference to the clock used for all time-based operations,
+     * allowing for deterministic testing via dependency injection.
+     */
+    clock_type& clock_;
+};
+
+}  // namespace xrpl::reduce_relay

src/xrpld/overlay/detail/OverlayImpl.cpp

@@ -188,7 +188,7 @@ OverlayImpl::OverlayImpl(
               collector))
     , resolver_(resolver)
     , nextId_(1)
-    , slots_(app, *this, app.config())
+    , slots_(app, *this, app.config(), stopwatch())
     , stats_(
           [this] { collectMetrics(); },
           collector,
@@ -610,17 +610,21 @@ OverlayImpl::stop()
 void
 OverlayImpl::onWrite(beast::PropertyStream::Map& stream)
 {
-    beast::PropertyStream::Set set("traffic", stream);
-    auto const stats = traffic_.getCounts();
-    for (auto const& pair : stats)
     {
-        beast::PropertyStream::Map item(set);
-        item["category"] = pair.second.name;
-        item["bytes_in"] = std::to_string(pair.second.bytesIn.load());
-        item["messages_in"] = std::to_string(pair.second.messagesIn.load());
-        item["bytes_out"] = std::to_string(pair.second.bytesOut.load());
-        item["messages_out"] = std::to_string(pair.second.messagesOut.load());
+        beast::PropertyStream::Set set("traffic", stream);
+        auto const stats = traffic_.getCounts();
+        for (auto const& pair : stats)
+        {
+            beast::PropertyStream::Map item(set);
+            item["category"] = pair.second.name;
+            item["bytes_in"] = std::to_string(pair.second.bytesIn.load());
+            item["messages_in"] = std::to_string(pair.second.messagesIn.load());
+            item["bytes_out"] = std::to_string(pair.second.bytesOut.load());
+            item["messages_out"] = std::to_string(pair.second.messagesOut.load());
+        }
     }
+
+    slots_.onWrite(stream);
 }
 
 //------------------------------------------------------------------------------
@@ -1429,12 +1433,24 @@ OverlayImpl::squelch(PublicKey const& validator, Peer::id_t id, uint32_t squelch
     }
 }
 
+void
+OverlayImpl::squelchAll(
+    PublicKey const& validator,
+    std::uint32_t squelchDuration,
+    std::function<void(Peer::id_t)> report)
+{
+    forEach([&](std::shared_ptr<PeerImp> const& p) {
+        p->send(makeSquelchMessage(validator, true, squelchDuration));
+        report(p->id());
+    });
+}
+
 void
 OverlayImpl::updateSlotAndSquelch(
     uint256 const& key,
     PublicKey const& validator,
     std::set<Peer::id_t>&& peers,
-    protocol::MessageType type)
+    bool isTrusted)
 {
     if (!slots_.baseSquelchReady())
         return;
@@ -1444,8 +1460,12 @@ OverlayImpl::updateSlotAndSquelch(
         post(
             strand_,
             // Must capture copies of reference parameters (i.e. key, validator)
-            [this, key = key, validator = validator, peers = std::move(peers), type]() mutable {
-                updateSlotAndSquelch(key, validator, std::move(peers), type);
+            [this,
+             key = key,
+             validator = validator,
+             peers = std::move(peers),
+             isTrusted]() mutable {
+                updateSlotAndSquelch(key, validator, std::move(peers), isTrusted);
             });
 
         return;
@@ -1453,9 +1473,12 @@ OverlayImpl::updateSlotAndSquelch(
 
     for (auto id : peers)
     {
-        slots_.updateSlotAndSquelch(key, validator, id, type, [&]() {
-            reportInboundTraffic(TrafficCount::Category::SquelchIgnored, 0);
-        });
+        slots_.updateSlotAndSquelch(
+            key,
+            validator,
+            id,
+            [&]() { reportInboundTraffic(TrafficCount::Category::SquelchIgnored, 0); },
+            isTrusted);
     }
 }
 
@@ -1464,7 +1487,7 @@ OverlayImpl::updateSlotAndSquelch(
     uint256 const& key,
     PublicKey const& validator,
     Peer::id_t peer,
-    protocol::MessageType type)
+    bool isTrusted)
 {
     if (!slots_.baseSquelchReady())
         return;
@@ -1475,18 +1498,71 @@ OverlayImpl::updateSlotAndSquelch(
             post(
                 strand_,
                 // Must capture copies of reference parameters (i.e. key, validator)
-                [this, key = key, validator = validator, peer, type]() {
-                    updateSlotAndSquelch(key, validator, peer, type);
+                [this, key = key, validator = validator, peer, isTrusted]() {
+                    updateSlotAndSquelch(key, validator, peer, isTrusted);
                 });
         }
         return;
     }
 
-    slots_.updateSlotAndSquelch(key, validator, peer, type, [&]() {
+    slots_.updateSlotAndSquelch(
+        key,
+        validator,
+        peer,
+        [&]() { reportInboundTraffic(TrafficCount::Category::SquelchIgnored, 0); },
+        isTrusted);
+}
+
+void
+OverlayImpl::updateUntrustedValidatorSlot(
+    uint256 const& key,
+    PublicKey const& validator,
+    Peer::id_t peer)
+{
+    if (!slots_.enhancedSquelchReady())
+        return;
+
+    if (!strand_.running_in_this_thread())
+    {
+        post(
+            strand_,
+            // Must capture copies of reference parameters (i.e. key, validator)
+            [this, key = key, validator = validator, peer]() {
+                updateUntrustedValidatorSlot(key, validator, peer);
+            });
+        return;
+    }
+
+    slots_.updateUntrustedValidatorSlot(key, validator, peer, [&]() {
         reportInboundTraffic(TrafficCount::Category::SquelchIgnored, 0);
     });
 }
 
+void
+OverlayImpl::handleUntrustedSquelch(PublicKey const& validator)
+{
+    if (!strand_.running_in_this_thread())
+    {
+        post(strand_, [this, validator] { handleUntrustedSquelch(validator); });
+        return;
+    }
+
+    auto count = 0;
+    // we can get the total number of peers with size(), however that would have
+    // to acquire another lock on peers. Instead, count the number of peers in
+    // the same loop, as we're already iterating all peers.
+    auto total = 0;
+    forEach([&](std::shared_ptr<PeerImp> const& p) {
+        ++total;
+        if (p->isSquelched(validator))
+            ++count;
+    });
+
+    // if majority of peers squelched the validator
+    if (count >= total - 1)
+        slots_.squelchUntrustedValidator(validator);
+}
+
 void
 OverlayImpl::deletePeer(Peer::id_t id)
 {

src/xrpld/overlay/detail/OverlayImpl.h

@@ -14,7 +14,6 @@
 
 #include <xrpl/basics/Resolver.h>
 #include <xrpl/basics/UnorderedContainers.h>
-#include <xrpl/basics/UptimeClock.h>
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/beast/insight/Collector.h>
 #include <xrpl/beast/insight/Gauge.h>
@@ -120,7 +119,7 @@ private:
     std::atomic<uint64_t> peerDisconnects_{0};
     std::atomic<uint64_t> peerDisconnectsCharges_{0};
 
-    reduce_relay::Slots<UptimeClock> slots_;
+    reduce_relay::Slots slots_;
 
     // Transaction reduce-relay metrics
     metrics::TxMetrics txMetrics_;
@@ -385,37 +384,86 @@ public:
     }
 
     /**
-     * Updates message count for validator/peer. Sends TMSquelch if the number
-     * of messages for N peers reaches threshold T. A message is counted
-     * if a peer receives the message for the first time and if
-     * the message has been  relayed.
-     * @param key Unique message's key
-     * @param validator Validator's public key
-     * @param peers Peers' id to update the slots for
-     * @param type Received protocol message type
+     * @brief Processes a message from a validator received via multiple peers.
+     *
+     * @details This function serves as a thread-safe entry point to the
+     * squelching system.
+     *
+     * @param key The unique hash of the message.
+     * @param validator The public key of the validator.
+     * @param peers A set of peer IDs that relayed this message.
+     * @param isTrusted `true` if the message is from a trusted validator.
      */
     void
     updateSlotAndSquelch(
         uint256 const& key,
         PublicKey const& validator,
         std::set<Peer::id_t>&& peers,
-        protocol::MessageType type);
+        bool isTrusted);
 
     /**
-     * Overload to reduce allocation in case of single peer
+     * @brief Processes a message from a validator received via a single peer.
+     *
+     * @details This function is a thread-safe entry point for handling a
+     * message from a single peer. It ensures the squelching feature is ready
+     * and serializes the call onto the `strand_`. It then invokes the
+     * underlying `Slots::updateSlotAndSquelch` method to process the message.
+     *
+     * @param key The unique hash of the message.
+     * @param validator The public key of the validator.
+     * @param peer The ID of the peer that relayed this message.
+     * @param isTrusted `true` if the message is from a trusted validator.
      */
     void
     updateSlotAndSquelch(
         uint256 const& key,
         PublicKey const& validator,
         Peer::id_t peer,
-        protocol::MessageType type);
+        bool isTrusted);
 
     /**
-     * Called when the peer is deleted. If the peer was selected to be the
-     * source of messages from the validator then squelched peers have to be
-     * unsquelched.
-     * @param id Peer's id
+     * @brief Processes a message specifically for the untrusted validator slot
+     * logic.
+     *
+     * @details This function is the thread-safe entry point for the enhanced
+     * squelching feature, which manages a limited number of slots for
+     * untrusted validators. It ensures the feature is ready, posts the work to
+     * the `strand_`, and then calls the underlying
+     * `Slots::updateUntrustedValidatorSlot` to handle the slot admission and
+     * evaluation logic.
+     *
+     * @param key The unique hash of the message.
+     * @param validator The public key of the untrusted validator.
+     * @param peer The ID of the peer that relayed this message.
+     */
+    void
+    updateUntrustedValidatorSlot(uint256 const& key, PublicKey const& validator, Peer::id_t peer);
+
+    /**
+     * @brief Handles a squelch message for an untrusted validator.
+     *
+     * @details This function is called when this node receives a message
+     * indicating that a peer is squelching an untrusted validator. It
+     * tallies how many of its own connected peers have also squelched the
+     * validator. If a majority of peers agree, this node takes definitive local
+     * action by calling `Slots::squelchUntrustedValidator`, effectively joining
+     * the consensus to silence the validator.
+     *
+     * @param validator The public key of the untrusted validator being
+     * squelched.
+     */
+    void
+    handleUntrustedSquelch(PublicKey const& validator);
+
+    /**
+     * @brief Handles the deletion of a peer from the overlay network.
+     *
+     * @details This function provides a thread-safe entry point for removing a
+     * peer. It ensures the operation is executed on the correct strand and
+     * then delegates the logic to `Slots::deletePeer`, which notifies all
+     * active slots about the peer's removal.
+     *
+     * @param id The ID of the peer to be deleted.
      */
     void
     deletePeer(Peer::id_t id);
@@ -444,6 +492,12 @@ private:
     squelch(PublicKey const& validator, Peer::id_t const id, std::uint32_t squelchDuration)
         const override;
 
+    void
+    squelchAll(
+        PublicKey const& validator,
+        std::uint32_t squelchDuration,
+        std::function<void(Peer::id_t)>) override;
+
     void
     unsquelch(PublicKey const& validator, Peer::id_t id) const override;
 

src/xrpld/overlay/detail/PeerImp.cpp

@@ -153,7 +153,7 @@ PeerImp::PeerImp(
     , publicKey_(publicKey)
     , lastPingTime_(clock_type::now())
     , creationTime_(clock_type::now())
-    , squelch_(app_.getJournal("Squelch"))
+    , squelchStore_(app_.getJournal("SquelchStore"), stopwatch())
     , usage_(consumer)
     , fee_{.fee = Resource::kFeeTrivialPeer, .context = ""}
     , slot_(slot)
@@ -280,8 +280,8 @@ PeerImp::send(std::shared_ptr<Message> const& m)
         if (!self->socket_.is_open())
             return;
 
-        auto validator = m->getValidatorKey();
-        if (validator && !self->squelch_.expireSquelch(*validator))
+        auto const validator = m->getValidatorKey();
+        if (validator && self->isSquelched(*validator))
         {
             self->overlay_.reportOutboundTraffic(
                 TrafficCount::Category::SquelchSuppressed,
@@ -598,6 +598,12 @@ PeerImp::hasRange(std::uint32_t uMin, std::uint32_t uMax)
     return (tracking_ != Tracking::Diverged) && (uMin >= minLedger_) && (uMax <= maxLedger_);
 }
 
+bool
+PeerImp::isSquelched(PublicKey const& validator) const
+{
+    return squelchStore_.isSquelched(validator);
+}
+
 //------------------------------------------------------------------------------
 
 void
@@ -1744,20 +1750,6 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMProposeSet> const& m)
     // suppression for 30 seconds to avoid doing a relatively expensive lookup
     // every time a spam packet is received
     PublicKey const publicKey{makeSlice(set.nodepubkey())};
-    auto const isTrusted = app_.getValidators().trusted(publicKey);
-
-    // If the operator has specified that untrusted proposals be dropped then
-    // this happens here I.e. before further wasting CPU verifying the signature
-    // of an untrusted key
-    if (!isTrusted)
-    {
-        // report untrusted proposal messages
-        overlay_.reportInboundTraffic(
-            TrafficCount::Category::ProposalUntrusted, Message::messageSize(*m));
-
-        if (app_.config().relayUntrustedProposals == -1)
-            return;
-    }
 
     uint256 const proposeHash = uint256::fromRaw(set.currenttxhash());
     uint256 const prevLedger = uint256::fromRaw(set.previousledger());
@@ -1767,13 +1759,16 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMProposeSet> const& m)
     uint256 const suppression = proposalUniqueId(
         proposeHash, prevLedger, set.proposeseq(), closeTime, publicKey.slice(), sig);
 
-    if (auto [added, relayed] = app_.getHashRouter().addSuppressionPeerWithStatus(suppression, id_);
+    auto const isTrusted = app_.getValidators().trusted(publicKey);
+
+    if (auto const& [added, relayed] =
+            app_.getHashRouter().addSuppressionPeerWithStatus(suppression, id_);
         !added)
     {
         // Count unique messages (Slots has it's own 'HashRouter'), which a peer
         // receives within IDLED seconds since the message has been relayed.
         if (relayed && (stopwatch().now() - *relayed) < reduce_relay::kIdled)
-            overlay_.updateSlotAndSquelch(suppression, publicKey, id_, protocol::mtPROPOSE_LEDGER);
+            overlay_.updateSlotAndSquelch(suppression, publicKey, id_, isTrusted);
 
         // report duplicate proposal messages
         overlay_.reportInboundTraffic(
@@ -1786,6 +1781,15 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMProposeSet> const& m)
 
     if (!isTrusted)
     {
+        overlay_.reportInboundTraffic(
+            TrafficCount::Category::ProposalUntrusted, Message::messageSize(*m));
+
+        // If the operator has specified that untrusted proposals be dropped
+        // then this happens here I.e. before further wasting CPU verifying the
+        // signature of an untrusted key
+        if (app_.config().relayUntrustedProposals == -1)
+            return;
+
         if (tracking_.load() == Tracking::Diverged)
         {
             JLOG(pJournal_.debug()) << "Proposal: Dropping untrusted (peer divergence)";
@@ -2354,19 +2358,6 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMValidation> const& m)
         // lookup every time a spam packet is received
         auto const isTrusted = app_.getValidators().trusted(val->getSignerPublic());
 
-        // If the operator has specified that untrusted validations be
-        // dropped then this happens here I.e. before further wasting CPU
-        // verifying the signature of an untrusted key
-        if (!isTrusted)
-        {
-            // increase untrusted validations received
-            overlay_.reportInboundTraffic(
-                TrafficCount::Category::ValidationUntrusted, Message::messageSize(*m));
-
-            if (app_.config().relayUntrustedValidations == -1)
-                return;
-        }
-
         auto key = sha512Half(makeSlice(m->validation()));
 
         auto [added, relayed] = app_.getHashRouter().addSuppressionPeerWithStatus(key, id_);
@@ -2378,8 +2369,7 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMValidation> const& m)
             // relayed.
             if (relayed && (stopwatch().now() - *relayed) < reduce_relay::kIdled)
             {
-                overlay_.updateSlotAndSquelch(
-                    key, val->getSignerPublic(), id_, protocol::mtVALIDATION);
+                overlay_.updateSlotAndSquelch(key, val->getSignerPublic(), id_, isTrusted);
             }
 
             // increase duplicate validations received
@@ -2390,6 +2380,23 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMValidation> const& m)
             return;
         }
 
+        // at this point the message is guaranteed to be unique
+        if (!isTrusted)
+        {
+            overlay_.reportInboundTraffic(
+                TrafficCount::Category::ValidationUntrusted, Message::messageSize(*m));
+
+            // If the operator has specified that untrusted validations be
+            // dropped then this happens here I.e. before further wasting CPU
+            // verifying the signature of an untrusted key
+            // TODO: Deprecate relayUntrustedValidations config once enhanced
+            // squelching is the defacto routing algorithm.
+            if (app_.config().relayUntrustedValidations == -1)
+                return;
+
+            overlay_.updateUntrustedValidatorSlot(key, val->getSignerPublic(), id_);
+        }
+
         if (!isTrusted && (tracking_.load() == Tracking::Diverged))
         {
             JLOG(pJournal_.debug()) << "Dropping untrusted validation from diverged peer";
@@ -2746,18 +2753,27 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMSquelch> const& m)
             return;
         }
 
-        std::uint32_t const duration = m->has_squelchduration() ? m->squelchduration() : 0;
-        if (!m->squelch())
-        {
-            self->squelch_.removeSquelch(key);
-        }
-        else if (!self->squelch_.addSquelch(key, std::chrono::seconds{duration}))
+        auto const duration =
+            std::chrono::seconds{m->has_squelchduration() ? m->squelchduration() : 0};
+
+        if (m->squelch() &&
+            (duration < reduce_relay::kMinUnsquelchExpire ||
+             duration > reduce_relay::kMaxUnsquelchExpirePeers))
         {
             self->fee_.update(Resource::kFeeInvalidData, "squelch duration");
+            return;
         }
 
         JLOG(self->pJournal_.debug())
-            << "onMessage: TMSquelch " << slice << " " << self->id() << " " << duration;
+            << "onMessage: TMSquelch " << (!m->squelch() ? "un" : "")
+            << "squelch message; validator: " << slice << "peer: " << self->id()
+            << " duration: " << duration.count();
+
+        self->squelchStore_.handleSquelch(key, m->squelch(), duration);
+
+        // if the squelch is for an untrusted validator
+        if (m->squelch() && !self->app_.getValidators().trusted(key))
+            self->overlay_.handleUntrustedSquelch(key);
     });
 }
 
@@ -3034,10 +3050,7 @@ PeerImp::checkPropose(
         if (!haveMessage.empty())
         {
             overlay_.updateSlotAndSquelch(
-                peerPos.suppressionID(),
-                peerPos.publicKey(),
-                std::move(haveMessage),
-                protocol::mtPROPOSE_LEDGER);
+                peerPos.suppressionID(), peerPos.publicKey(), std::move(haveMessage), isTrusted);
         }
     }
 }
@@ -3069,7 +3082,7 @@ PeerImp::checkValidation(
             if (!haveMessage.empty())
             {
                 overlay_.updateSlotAndSquelch(
-                    key, val->getSignerPublic(), std::move(haveMessage), protocol::mtVALIDATION);
+                    key, val->getSignerPublic(), std::move(haveMessage), val->isTrusted());
             }
         }
     }

src/xrpld/overlay/detail/PeerImp.h

@@ -6,7 +6,7 @@
 #include <xrpld/overlay/Compression.h>
 #include <xrpld/overlay/Message.h>
 #include <xrpld/overlay/Peer.h>
-#include <xrpld/overlay/Squelch.h>
+#include <xrpld/overlay/SquelchStore.h>
 #include <xrpld/overlay/detail/OverlayImpl.h>
 #include <xrpld/overlay/detail/ProtocolVersion.h>
 #include <xrpld/peerfinder/PeerfinderManager.h>
@@ -15,8 +15,8 @@
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/Number.h>
 #include <xrpl/basics/UnorderedContainers.h>
-#include <xrpl/basics/UptimeClock.h>
 #include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/chrono.h>
 #include <xrpl/beast/net/IPEndpoint.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/utility/WrappedSink.h>
@@ -129,7 +129,7 @@ private:
     clock_type::time_point lastPingTime_;
     clock_type::time_point const creationTime_;
 
-    reduce_relay::Squelch<UptimeClock> squelch_;
+    reduce_relay::SquelchStore squelchStore_;
 
     // Notes on thread locking:
     //
@@ -469,6 +469,14 @@ public:
         return txReduceRelayEnabled_;
     }
 
+    /**
+     * Check if a given validator is squelched.
+     * @param validator Validator's public key
+     * @return true if squelch exists and it is not expired. False otherwise.
+     */
+    bool
+    isSquelched(PublicKey const& validator) const;
+
 private:
     void
     close();
@@ -783,7 +791,7 @@ PeerImp::PeerImp(
     , publicKey_(publicKey)
     , lastPingTime_(clock_type::now())
     , creationTime_(clock_type::now())
-    , squelch_(app_.getJournal("Squelch"))
+    , squelchStore_(app_.getJournal("SquelchStore"), stopwatch())
     , usage_(usage)
     , fee_{.fee = Resource::kFeeTrivialPeer}
     , slot_(std::move(slot))

src/xrpld/overlay/detail/Slot.cpp

@@ -0,0 +1,669 @@
+#include <xrpld/overlay/Slot.h>
+
+#include <xrpld/overlay/Peer.h>
+#include <xrpld/overlay/ReduceRelayCommon.h>
+
+#include <xrpl/basics/Log.h>
+#include <xrpl/basics/UnorderedContainers.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/random.h>
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/beast/utility/PropertyStream.h>
+#include <xrpl/beast/utility/instrumentation.h>
+#include <xrpl/protocol/PublicKey.h>
+#include <xrpl/protocol/tokens.h>
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <iterator>
+#include <optional>
+#include <sstream>
+#include <string>
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
+namespace xrpl::reduce_relay {
+
+void
+Slot::deleteIdlePeer(PublicKey const& validator)
+{
+    using namespace std::chrono;
+    auto const now = clock_.now();
+    for (auto it = peers_.begin(); it != peers_.end();)
+    {
+        auto const& peer = it->second;
+        auto const id = it->first;
+        ++it;
+        if (now - peer.lastMessage > reduce_relay::kIdled)
+        {
+            JLOG(journal_.trace())
+                << "deleteIdlePeer: deleting idle peer " << formatLogMessage(validator, id)
+                << " peer_state: " << to_string(peer.state)
+                << " idle for: " << (now - peer.lastMessage).count();
+            deletePeer(validator, id, false);
+        }
+    }
+}
+
+void
+Slot::update(PublicKey const& validator, Peer::id_t id, ignored_squelch_callback report)
+{
+    using namespace std::chrono;
+    auto const now = clock_.now();
+    auto const it = peers_.find(id);
+
+    // First message from this peer
+    if (it == peers_.end())
+    {
+        JLOG(journal_.trace()) << "update: adding new slot" << formatLogMessage(validator, id);
+        peers_.emplace(
+            id,
+            PeerInfo{
+                .state = PeerState::Counting,
+                .count = 0,
+                .expire = now,
+                .lastMessage = now,
+                .timesSelected = 0});
+        initCounting();
+        return;
+    }
+    // Message from a peer with expired squelch
+    if (it->second.state == PeerState::Squelched && now > it->second.expire)
+    {
+        JLOG(journal_.trace()) << "update: squelch expired" << formatLogMessage(validator, id);
+        it->second.state = PeerState::Counting;
+        it->second.lastMessage = now;
+        initCounting();
+        return;
+    }
+
+    auto& peer = it->second;
+
+    peer.lastMessage = now;
+
+    // report if we received a message from a squelched peer
+    if (peer.state == PeerState::Squelched)
+        report();
+
+    if (getState() != SlotState::Counting || peer.state == PeerState::Squelched)
+        return;
+
+    if (++peer.count > reduce_relay::kMinMessageThreshold)
+        considered_.insert(id);
+    if (peer.count == (reduce_relay::kMaxMessageThreshold + 1))
+        ++reachedThreshold_;
+
+    if (now - lastSelected_ > 2 * reduce_relay::kMaxUnsquelchExpireDefault)
+    {
+        JLOG(journal_.warn()) << "update: resetting due to inactivity"
+                              << formatLogMessage(validator, id) << " inactive for: "
+                              << duration_cast<seconds>(now - lastSelected_).count();
+        initCounting();
+        return;
+    }
+
+    if (reachedThreshold_ == maxSelectedPeers_)
+    {
+        // Randomly select maxSelectedPeers_ peers from considered.
+        // Exclude peers that have been idling > kIdled -
+        // it's possible that deleteIdlePeer() has not been called yet.
+        // If number of remaining peers != maxSelectedPeers_
+        // then reset the Counting state and let deleteIdlePeer() handle
+        // idled peers.
+        std::unordered_set<Peer::id_t> selected;
+        std::stringstream str;
+        while (selected.size() != maxSelectedPeers_ && !considered_.empty())
+        {
+            auto const i = considered_.size() == 1 ? 0 : randInt(considered_.size() - 1);
+            auto const it = std::next(considered_.begin(), i);
+            auto const id = *it;
+            considered_.erase(it);
+
+            auto const& peersIt = peers_.find(id);
+            if (peersIt == peers_.end())
+            {
+                JLOG(journal_.error())
+                    << "update: peer not found" << formatLogMessage(validator, id);
+
+                continue;
+            }
+            if (now - peersIt->second.lastMessage < reduce_relay::kIdled)
+            {
+                selected.insert(id);
+                str << id << " ";
+            }
+        }
+
+        if (selected.size() != maxSelectedPeers_)
+        {
+            JLOG(journal_.error())
+                << "update: selection failed" << formatLogMessage(validator, std::nullopt);
+
+            initCounting();
+            return;
+        }
+
+        lastSelected_ = now;
+
+        JLOG(journal_.trace()) << "update: selected peers "
+                               << formatLogMessage(validator, std::nullopt)
+                               << " peers: " << str.str();
+
+        XRPL_ASSERT(
+            peers_.size() >= maxSelectedPeers_, "xrpl::reduce_relay::Slot::update : minimum peers");
+
+        // squelch peers which are not selected and
+        // not already squelched
+        str.clear();
+        for (auto& [k, v] : peers_)
+        {
+            v.count = 0;
+
+            if (selected.contains(k))
+            {
+                v.state = PeerState::Selected;
+                ++v.timesSelected;
+            }
+
+            else if (v.state != PeerState::Squelched)
+            {
+                if (journal_.trace())
+                    str << k << " ";
+                v.state = PeerState::Squelched;
+                std::chrono::seconds const duration =
+                    getSquelchDuration(peers_.size() - maxSelectedPeers_);
+                v.expire = now + duration;
+                handler_.squelch(validator, k, duration.count());
+            }
+        }
+        JLOG(journal_.trace()) << "update: squelched peers "
+                               << formatLogMessage(validator, std::nullopt)
+                               << " peers: " << str.str();
+        considered_.clear();
+        reachedThreshold_ = 0;
+        state_ = SlotState::Selected;
+    }
+}
+
+std::chrono::seconds
+Slot::getSquelchDuration(std::size_t npeers) const
+{
+    using namespace std::chrono;
+    auto m = std::max(
+        reduce_relay::kMaxUnsquelchExpireDefault, seconds{reduce_relay::kSquelchPerPeer * npeers});
+    if (m > reduce_relay::kMaxUnsquelchExpirePeers)
+    {
+        m = reduce_relay::kMaxUnsquelchExpirePeers;
+        JLOG(journal_.warn()) << "getSquelchDuration: unexpected squelch duration " << npeers;
+    }
+    return seconds{randInt(reduce_relay::kMinUnsquelchExpire / 1s, m / 1s)};
+}
+
+void
+Slot::deletePeer(PublicKey const& validator, Peer::id_t id, bool erase)
+{
+    auto it = peers_.find(id);
+    if (it == peers_.end())
+        return;
+
+    std::vector<Peer::id_t> toUnsquelch;
+    auto const now = clock_.now();
+    if (it->second.state == PeerState::Selected)
+    {
+        JLOG(journal_.debug()) << "deletePeer: unsquelching selected peer "
+                               << formatLogMessage(validator, id)
+                               << " peer_state: " << to_string(it->second.state)
+                               << " considered: " << (considered_.contains(id))
+                               << " erase: " << erase;
+
+        for (auto& [k, v] : peers_)
+        {
+            if (v.state == PeerState::Squelched)
+                toUnsquelch.push_back(k);
+            v.state = PeerState::Counting;
+            v.count = 0;
+            v.expire = now;
+        }
+
+        considered_.clear();
+        reachedThreshold_ = 0;
+        state_ = SlotState::Counting;
+    }
+    else if (considered_.contains(id))
+    {
+        if (it->second.count > reduce_relay::kMaxMessageThreshold)
+            --reachedThreshold_;
+        considered_.erase(id);
+    }
+
+    it->second.lastMessage = now;
+    it->second.count = 0;
+
+    if (erase)
+        peers_.erase(it);
+
+    // Must be after peers_.erase(it)
+    for (auto const& k : toUnsquelch)
+        handler_.unsquelch(validator, k);
+}
+
+void
+Slot::onWrite(beast::PropertyStream::Map& stream) const
+{
+    using namespace std::chrono;
+    auto const now = clock_.now();
+    stream["state"] = to_string(getState());
+    stream["reachedThreshold"] = reachedThreshold_;
+    stream["considered"] = considered_.size();
+    stream["lastSelected"] = duration_cast<std::chrono::seconds>(now - lastSelected_).count();
+    stream["isTrusted"] = isTrusted_;
+
+    beast::PropertyStream::Set peers("peers", stream);
+
+    for (auto const& [id, info] : peers_)
+    {
+        beast::PropertyStream::Map item(peers);
+        item["id"] = id;
+        item["count"] = info.count;
+        item["expire"] = duration_cast<std::chrono::seconds>(info.expire - now).count();
+        item["lastMessage"] = duration_cast<std::chrono::seconds>(now - info.lastMessage).count();
+        item["timesSelected"] = info.timesSelected;
+        item["state"] = to_string(info.state);
+    }
+}
+
+void
+Slot::initCounting()
+{
+    state_ = SlotState::Counting;
+    considered_.clear();
+    reachedThreshold_ = 0;
+    for (auto& [_, peer] : peers_)
+    {
+        (void)_;
+        peer.count = 0;
+    }
+}
+
+std::string
+Slot::formatLogMessage(PublicKey const& validator, std::optional<Peer::id_t> id) const
+{
+    std::stringstream ss;
+    ss << "validator: " << toBase58(TokenType::NodePublic, validator);
+    if (id)
+        ss << " peer: " << *id;
+    ss << " trusted: " << isTrusted_;
+    ss << " slot_state: " << to_string(getState());
+    return ss.str();
+}
+
+// --------------------------------- Slots --------------------------------- //
+
+bool
+Slots::reduceRelayReady()
+{
+    if (!reduceRelayReady_)
+    {
+        reduceRelayReady_ = std::chrono::duration_cast<std::chrono::minutes>(
+                                clock_.now().time_since_epoch()) > reduce_relay::kWaitOnBootup;
+    }
+
+    return reduceRelayReady_;
+}
+
+void
+Slots::registerSquelchedValidator(PublicKey const& validatorKey, Peer::id_t peerID)
+{
+    peersWithSquelchedValidators_[validatorKey].insert(peerID);
+}
+
+bool
+Slots::expireAndIsValidatorSquelched(PublicKey const& validatorKey)
+{
+    beast::expire(peersWithSquelchedValidators_, reduce_relay::kMaxUnsquelchExpireDefault);
+
+    return peersWithSquelchedValidators_.find(validatorKey) != peersWithSquelchedValidators_.end();
+}
+
+bool
+Slots::expireAndIsPeerSquelched(PublicKey const& validatorKey, Peer::id_t peerID)
+{
+    beast::expire(peersWithSquelchedValidators_, reduce_relay::kMaxUnsquelchExpireDefault);
+
+    auto const it = peersWithSquelchedValidators_.find(validatorKey);
+
+    // if validator was not squelched, the peer was also not squelched
+    if (it == peersWithSquelchedValidators_.end())
+        return false;
+
+    // if a peer is found the squelch for it has not expired
+    return it->second.contains(peerID);
+}
+
+bool
+Slots::expireAndIsPeerMessageCached(uint256 const& key, Peer::id_t id)
+{
+    beast::expire(peersWithMessage_, reduce_relay::kIdled);
+
+    // return false if the ID was not inserted
+    if (key.isNonZero())
+        return !peersWithMessage_[key].insert(id).second;
+
+    return false;
+}
+
+void
+Slots::updateSlotAndSquelch(
+    uint256 const& key,
+    PublicKey const& validator,
+    Peer::id_t id,
+    Slot::ignored_squelch_callback report,
+    bool isTrusted)
+{
+    if (expireAndIsPeerMessageCached(key, id))
+        return;
+
+    // If we receive a message from a trusted validator either update an
+    // existing slot or insert a new one. If we are not running enhanced
+    // squelching also deduplicate untrusted validator messages
+    if (isTrusted || !enhancedSquelchEnabled_)
+    {
+        // if enhanced squelching is disabled, keep untrusted validator slots
+        // separately from trusted ones
+        auto it =
+            (isTrusted ? trustedSlots_ : untrustedSlots_)
+                .emplace(
+                    std::make_pair(
+                        validator,
+                        Slot(
+                            handler_, logs_.journal("Slot"), maxSelectedPeers_, isTrusted, clock_)))
+                .first;
+
+        it->second.update(validator, id, report);
+    }
+    else
+    {
+        auto it = untrustedSlots_.find(validator);
+        // If we received a message from a validator that is not
+        // selected, and is not squelched, there is nothing to do. It
+        // will be squelched later when `updateValidatorSlot` is called.
+        if (it == untrustedSlots_.end())
+            return;
+
+        it->second.update(validator, id, report);
+    }
+}
+
+void
+Slots::updateUntrustedValidatorSlot(
+    uint256 const& key,
+    PublicKey const& validator,
+    Peer::id_t id,
+    Slot::ignored_squelch_callback report)
+{
+    // We received a message from an already selected validator
+    // we can ignore this message
+    if (untrustedSlots_.contains(validator))
+        return;
+
+    // Did we receive a message from an already squelched validator?
+    // This could happen in few cases:
+    //      1. It happened so that the squelch for a particular peer expired
+    //      before our local squelch.
+    //      2. We receive a message from a new peer that did not receive the
+    //      squelch request.
+    //      3. The peer is ignoring our squelch request and we have not sent
+    //      the control message in a while.
+    // In all of these cases we can only send them a squelch request again.
+    if (expireAndIsValidatorSquelched(validator))
+    {
+        if (!expireAndIsPeerSquelched(validator, id))
+        {
+            JLOG(journal_.debug())
+                << "updateUntrustedValidatorSlot: received a message from a "
+                   "squelched validator "
+                << "validator: " << toBase58(TokenType::NodePublic, validator) << " peer: " << id;
+
+            registerSquelchedValidator(validator, id);
+            handler_.squelch(validator, id, reduce_relay::kMaxUnsquelchExpireDefault.count());
+        }
+        return;
+    }
+
+    // Do we have any available slots for additional untrusted validators?
+    // This could happen in few cases:
+    //      1. We received a message from a new untrusted validator, but we
+    //      are at capacity.
+    //      2. We received a message from a previously squelched validator.
+    // In all of these cases we send a squelch message to all peers.
+    // The validator may still  be considered by the selector. However, it
+    // will be eventually cleaned and squelched
+    if (untrustedSlots_.size() == reduce_relay::kMaxUntrustedSlots)
+    {
+        JLOG(journal_.debug()) << "updateUntrustedValidatorSlot: slots full squelching validator "
+                               << "validator: " << toBase58(TokenType::NodePublic, validator);
+
+        handler_.squelchAll(validator, kMaxUnsquelchExpireDefault.count(), [&](Peer::id_t id) {
+            registerSquelchedValidator(validator, id);
+        });
+        return;
+    }
+
+    if (auto const v = updateConsideredValidator(validator, id))
+    {
+        JLOG(journal_.debug()) << "updateUntrustedValidatorSlot: selected untrusted validator "
+                               << "validator: " << toBase58(TokenType::NodePublic, *v);
+
+        untrustedSlots_.emplace(
+            std::make_pair(
+                *v, Slot(handler_, logs_.journal("Slot"), maxSelectedPeers_, false, clock_)));
+    }
+    // When we reach kMaxUntrustedSlots, don't  explicitly clean them.
+    // Since we stop updating their counters, they will idle, and will be
+    // removed and squelched.
+}
+
+std::optional<PublicKey>
+Slots::updateConsideredValidator(PublicKey const& validator, Peer::id_t peer)
+{
+    auto const now = clock_.now();
+
+    auto it = consideredValidators_.find(validator);
+    if (it == consideredValidators_.end())
+    {
+        consideredValidators_.emplace(
+            std::make_pair(
+                validator,
+                ValidatorInfo{
+                    .count = 1,
+                    .lastMessage = now,
+                    .peers = {peer},
+                }));
+
+        return std::nullopt;
+    }
+
+    it->second.peers.insert(peer);
+    it->second.lastMessage = now;
+    ++it->second.count;
+
+    // if the validator has not met selection criteria yet
+    if (it->second.count < reduce_relay::kMaxMessageThreshold)
+        return std::nullopt;
+
+    auto key = it->first;
+    consideredValidators_.erase(it);
+
+    return key;
+}
+
+void
+Slots::squelchUntrustedValidator(PublicKey const& validator)
+{
+    JLOG(journal_.info()) << "squelchUntrustedValidator: squelching untrusted validator: "
+                          << toBase58(TokenType::NodePublic, validator);
+    // to prevent the validator from being reinserted squelch the validator
+    // before removing the validator from consideration and slots
+    handler_.squelchAll(validator, kMaxUnsquelchExpireDefault.count(), [&](Peer::id_t id) {
+        registerSquelchedValidator(validator, id);
+    });
+
+    consideredValidators_.erase(validator);
+    untrustedSlots_.erase(validator);
+}
+
+void
+Slots::deletePeer(Peer::id_t id, bool erase)
+{
+    auto const f = [&](slots_map& slots) {
+        for (auto& [validator, slot] : slots)
+            slot.deletePeer(validator, id, erase);
+    };
+
+    f(trustedSlots_);
+    f(untrustedSlots_);
+}
+
+void
+Slots::deleteIdlePeers()
+{
+    auto const f = [&](slots_map& slots) {
+        auto const now = clock_.now();
+
+        for (auto it = slots.begin(); it != slots.end();)
+        {
+            auto const& validator = it->first;
+            auto& slot = it->second;
+            slot.deleteIdlePeer(validator);
+
+            // delete the slot if the untrusted slot no longer meets the
+            // selection critera or it has not been selected for a while
+            if ((!slot.isTrusted_ && slot.getPeers().size() < maxSelectedPeers_) ||
+                now - it->second.getLastSelected() > reduce_relay::kMaxUnsquelchExpireDefault)
+            {
+                JLOG(journal_.trace())
+                    << "deleteIdlePeers: deleting " << (slot.isTrusted_ ? "trusted" : "untrusted")
+                    << " slot " << toBase58(TokenType::NodePublic, it->first) << " reason: "
+                    << (now - it->second.getLastSelected() >
+                                reduce_relay::kMaxUnsquelchExpireDefault
+                            ? " inactive "
+                            : " insufficient peers");
+
+                // if an untrusted validator slot idled - peers stopped
+                // sending messages for this validator squelch it
+                if (!it->second.isTrusted_)
+                {
+                    handler_.squelchAll(
+                        it->first,
+                        reduce_relay::kMaxUnsquelchExpireDefault.count(),
+                        [&](Peer::id_t id) { registerSquelchedValidator(it->first, id); });
+                }
+
+                it = slots.erase(it);
+            }
+            else
+            {
+                {
+                    ++it;
+                }
+            }
+        }
+    };
+
+    f(trustedSlots_);
+    f(untrustedSlots_);
+
+    // remove and squelch all validators that the selector deemed unsuitable
+    // there might be some good validators in this set that "lapsed".
+    // However, since these are untrusted validators we're not concerned
+    for (auto const& validator : cleanConsideredValidators())
+    {
+        handler_.squelchAll(
+            validator, reduce_relay::kMaxUnsquelchExpireDefault.count(), [&](Peer::id_t id) {
+                registerSquelchedValidator(validator, id);
+            });
+    }
+}
+
+std::vector<PublicKey>
+Slots::cleanConsideredValidators()
+{
+    auto const now = clock_.now();
+
+    std::vector<PublicKey> keys;
+    std::stringstream ss;
+    for (auto it = consideredValidators_.begin(); it != consideredValidators_.end();)
+    {
+        if (now - it->second.lastMessage > reduce_relay::kMaxUntrustedValidatorIdle)
+        {
+            keys.push_back(it->first);
+            ss << " " << toBase58(TokenType::NodePublic, it->first);
+            it = consideredValidators_.erase(it);
+        }
+        // Due to some reason the validator idled, reset their progress
+        else if (now - it->second.lastMessage > reduce_relay::kIdled)
+        {
+            it->second.reset();
+            ++it;
+        }
+        else
+        {
+            {
+                ++it;
+            }
+        }
+    }
+
+    if (!keys.empty())
+    {
+        JLOG(journal_.info()) << "cleanConsideredValidators: removed considered validators "
+                              << ss.str();
+    }
+    return keys;
+}
+
+void
+Slots::onWrite(beast::PropertyStream::Map& stream) const
+{
+    auto const writeSlot = [](beast::PropertyStream::Set& set,
+                              hash_map<PublicKey, Slot> const& slots) {
+        for (auto const& [validator, slot] : slots)
+        {
+            beast::PropertyStream::Map item(set);
+            item["validator"] = toBase58(TokenType::NodePublic, validator);
+            slot.onWrite(item);
+        }
+    };
+
+    beast::PropertyStream::Map slots("slots", stream);
+
+    {
+        beast::PropertyStream::Set set("trusted", slots);
+        writeSlot(set, trustedSlots_);
+    }
+
+    {
+        beast::PropertyStream::Set set("untrusted", slots);
+        writeSlot(set, untrustedSlots_);
+    }
+
+    {
+        beast::PropertyStream::Set set("considered", slots);
+
+        auto const now = clock_.now();
+
+        for (auto const& [validator, info] : consideredValidators_)
+        {
+            beast::PropertyStream::Map item(set);
+            item["validator"] = toBase58(TokenType::NodePublic, validator);
+            item["lastMessage"] =
+                std::chrono::duration_cast<std::chrono::seconds>(now - info.lastMessage).count();
+            item["messageCount"] = info.count;
+            item["peers"] = info.peers.size();
+        }
+    }
+}
+
+}  // namespace xrpl::reduce_relay

src/xrpld/overlay/detail/SquelchStore.cpp

@@ -0,0 +1,74 @@
+#include <xrpld/overlay/SquelchStore.h>
+
+#include <xrpld/overlay/ReduceRelayCommon.h>
+
+#include <xrpl/basics/Log.h>
+#include <xrpl/basics/Slice.h>
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/protocol/PublicKey.h>
+
+#include <chrono>
+#include <unordered_map>
+
+namespace xrpl::reduce_relay {
+
+void
+SquelchStore::handleSquelch(PublicKey const& validator, bool squelch, std::chrono::seconds duration)
+{
+    // Remove all expired squelches. This call is here, as it is on the least
+    // critical execution path, that does not require periodic cleanup calls.
+    removeExpired();
+
+    if (squelch)
+    {
+        // This should never trigger. The squelch duration is validated in
+        // PeerImp.onMessage(TMSquelch). However, if somehow invalid duration is
+        // passed, log is as an error
+        if ((duration < reduce_relay::kMinUnsquelchExpire ||
+             duration > reduce_relay::kMaxUnsquelchExpirePeers))
+        {
+            JLOG(journal_.error())
+                << "SquelchStore: invalid squelch duration validator: " << Slice(validator)
+                << " duration: " << duration.count();
+            return;
+        }
+
+        add(validator, duration);
+        return;
+    }
+
+    remove(validator);
+}
+
+bool
+SquelchStore::isSquelched(PublicKey const& validator) const
+{
+    auto const now = clock_.now();
+
+    auto const it = squelched_.find(validator);
+    if (it == squelched_.end())
+        return false;
+
+    return it->second > now;
+}
+
+void
+SquelchStore::add(PublicKey const& validator, std::chrono::seconds const& duration)
+{
+    squelched_[validator] = clock_.now() + duration;
+}
+
+void
+SquelchStore::remove(PublicKey const& validator)
+{
+    squelched_.erase(validator);
+}
+
+void
+SquelchStore::removeExpired()
+{
+    auto const now = clock_.now();
+    std::erase_if(squelched_, [&](auto const& entry) { return entry.second < now; });
+}
+
+}  // namespace xrpl::reduce_relay