test: pin revocation response refusals

The original design sketch had honest nodes informing senders of fresher
manifests upon receipt of stale; it was parked as an amplification
surface pending an explicit bounded one-shot model. The repair ledger is
that model, so the lane lands as one more caller of sendManifestRepair
at the singleton global-sequence guard, reusing the snapshot the guard
already fetches.

Strictly-stale only: equal-sequence arrivals are ordinary always-send
traffic and draw nothing. The trigger needs no signature verification;
safety comes from bounds, not authentication - answers are limited to
masters this node retains, once per sequence per connection via the
shared ledger, and monotone sequences make correction exchanges
converge in one round trip. A retained revocation is itself the answer:
a stale normal manifest for a revoked master draws the tombstone, giving
revocations a demand-driven re-supply path to any peer that demonstrably
still uses the dead identity. Retention, not listing, qualifies a master
for an answer.

Deferred: the legacy batch lane does not correct (singletons only); no
live-scenario control yet (the mixed old-binary connect dump is the
natural trigger); design-doc sync for the reopened corrections language
left to the integration pass.

.gitignore

@@ -129,3 +129,8 @@ generated
 
 # Suggested in-tree build directory
 /.build/
+
+# x-testnet runtime state; scenario definitions remain tracked.
+/.testnet/output/
+/.testnet/nodes/
+/testnet/

.testnet/scenarios/manifest_revocation_live.py

@@ -0,0 +1,108 @@
+"""Revoke one of five validators live and watch the terminal update propagate.
+
+Five validators are used deliberately: revoking one leaves four of five —
+exactly the 80% quorum — so the network keeps closing ledgers and the
+scenario can assert liveness with the revoked member excluded, not merely a
+halt. (On a three-node UNL the same revocation stops the network: small-net
+quorums round up to everyone.)
+
+The revocation is installed by config on the OLD release node deliberately:
+under the new transport slice a config-loaded revocation on an upgraded node
+sits in its durable cache and is never gossiped (there is no connect-time
+dump and no ambient manifest relay), while the old binary still dumps its
+manifest cache on every fresh connection and relays what it accepts. The old
+node is therefore the only in-topology seeder for a config-injected
+revocation — that asymmetry is itself part of the documented durability
+boundary.
+"""
+
+from xahaud_scripts.testnet.scenario import (
+    AssertionError as ScenarioAssertion,
+)
+
+
+async def _await_log(ctx, log, pattern, *, nodes, name, since=None, deadline=36):
+    # Catch the runner's ScenarioAssertion, not the builtin. The scenario
+    # module shadows AssertionError without subclassing it.
+    for attempt in range(deadline):
+        try:
+            ctx.assert_log(pattern, nodes=nodes, **({"since": since} if since else {}))
+            return
+        except ScenarioAssertion:
+            if attempt == deadline - 1:
+                raise
+            if attempt % 6 == 5:
+                log(f"{name}: attempt {attempt + 1}/{deadline}")
+            await ctx.sleep(5, name=name)
+
+
+async def scenario(ctx, log):
+    nodes = [0, 1, 2, 3, 4, 5]
+    # Validators 0-4 meet at n0. The old release node n5 is a pendant on n4
+    # so it can seed the revocation without joining the UNL mesh.
+    expected = ctx.topology_edges(
+        [(0, 1), (0, 2), (0, 3), (0, 4), (4, 5)]
+    )
+
+    await ctx.apply_topology(expected, nodes=nodes, exact=False)
+
+    # Normal operation first: everyone validates and validator 4's manifest
+    # is durably known, so the revocation supersedes real retained state.
+    await ctx.wait_for_ledgers(2, node_id=0, timeout=180)
+
+    revoked = ctx.mark("revoked")
+    result = await ctx.revoke_validator(4, 5)
+    log(
+        "revoked validator n4 master via the old relay n5: "
+        f"{result['public_key']}"
+    )
+
+    # The restarted old node loads the revocation and seeds it through its
+    # legacy connect-time dump and accept-relay; its peer n4 spreads it into
+    # the mesh through the normal revocation relay lane. The old node's
+    # reconnect interval dominates the latency, so wait on the log fact.
+    await _await_log(
+        ctx,
+        log,
+        "Revoked",
+        nodes=[4],
+        name="await-revocation-arrival",
+        since=revoked,
+        deadline=36,
+    )
+
+    # Terminal manifest applied and relayed onward by upgraded nodes: first
+    # at the old seeder's direct peer, then across the mesh.
+    ctx.assert_log(
+        "manifest_revocation accepted_for_relay",
+        since=revoked,
+        nodes=[4],
+    )
+    await _await_log(
+        ctx,
+        log,
+        "Revoked",
+        nodes=[0],
+        name="await-mesh-revocation",
+        since=revoked,
+        deadline=36,
+    )
+    ctx.assert_log(
+        "manifest_revocation accepted_for_relay",
+        since=revoked,
+        nodes=[0],
+    )
+
+    # Liveness with the revoked member excluded: four of five is exactly
+    # quorum, so ledgers keep closing.
+    await ctx.wait_for_ledgers(2, node_id=0, timeout=180)
+
+    ctx.assert_not_log("Validation forwarded by peer is invalid", nodes=nodes)
+    ctx.assert_not_log("Validation: Too small", nodes=nodes)
+
+    log(
+        "PASS: a config-injected revocation seeded by the old relay reached"
+        " the mesh, applied terminally on upgraded nodes, relayed onward,"
+        " and the remaining four validators kept the network live at exact"
+        " quorum"
+    )

.testnet/scenarios/manifest_rotation_propagation.py

@@ -0,0 +1,113 @@
+"""Rotate a validator's manifest mid-run and watch the bump propagate.
+
+"Heals on a sequence bump" is the load-bearing healing claim of the
+manifest-before-validation transport slice. This exercises it live for the
+first time: the validator mints sequence 2 and restarts; its validations then
+carry the new manifest; an old release relay forwards the existing envelopes;
+the upgraded observer supersedes its retained knowledge and admits sequence 2;
+and the observer's repair lane re-arms at the newer sequence when the old
+relay's later validations arrive naked.
+"""
+
+from xahaud_scripts.testnet.scenario import (
+    AssertionError as ScenarioAssertion,
+)
+
+
+async def _await_log(ctx, log, pattern, *, nodes, name, since=None, deadline=36):
+    # Catch the runner's ScenarioAssertion, not the builtin. The scenario
+    # module shadows AssertionError without subclassing it.
+    for attempt in range(deadline):
+        try:
+            ctx.assert_log(pattern, nodes=nodes, **({"since": since} if since else {}))
+            return
+        except ScenarioAssertion:
+            if attempt == deadline - 1:
+                raise
+            if attempt % 6 == 5:
+                log(f"{name}: attempt {attempt + 1}/{deadline}")
+            await ctx.sleep(5, name=name)
+
+
+async def scenario(ctx, log):
+    nodes = [0, 1, 2]
+    expected = ctx.topology_edges([(0, 1), (1, 2)])
+
+    await ctx.apply_topology(expected, nodes=nodes, exact=False)
+
+    # Baseline: sequence 1 propagates through the old relay and is admitted
+    # by the upgraded observer before we rotate anything. The validator can
+    # race several ledgers ahead of propagation under fast bootstrap, so
+    # wait on the log fact itself.
+    await ctx.wait_for_ledgers(2, node_id=0, timeout=120)
+    await _await_log(
+        ctx,
+        log,
+        "manifest_validation single_manifest_processed .*sequence=1",
+        nodes=[2],
+        name="await-baseline-admission",
+        deadline=24,
+    )
+
+    rotated = ctx.mark("rotated")
+    rotation = await ctx.rotate_validator_manifest(0)
+    assert rotation["sequence"] == 2, rotation
+    log(f"rotated validator n0 to manifest sequence {rotation['sequence']}")
+
+    # The restarted validator re-joins and validates under the new signing
+    # key; always-send carries the sequence-2 prerequisite with each
+    # validation, and the old relay's one-shot manifest forward plus later
+    # naked relays exercise both the supersede and the repair re-arm. Wait
+    # on the terminal log fact rather than ledger counts: in this topology
+    # only the validator advances its ledger, and its own restart closed the
+    # node-0 WebSocket ledger feed. The repair re-arm is the last event in
+    # the causal chain, so everything else must precede it.
+    # polls at 5s => 180s budget at ~16s consensus rounds
+    await _await_log(
+        ctx,
+        log,
+        "manifest_validation repair_sent .*sequence=2",
+        nodes=[2],
+        name="await-repair-rearm",
+        since=rotated,
+        deadline=36,
+    )
+
+    ctx.assert_log(
+        "manifest_validation pair_enqueued .*sequence=2",
+        since=rotated,
+        nodes=[0],
+    )
+    ctx.assert_log(
+        "manifest_validation candidate_staged .*sequence=2",
+        since=rotated,
+        nodes=[2],
+    )
+    ctx.assert_log(
+        "manifest_validation candidate_matched",
+        since=rotated,
+        nodes=[2],
+    )
+    ctx.assert_log(
+        "manifest_validation single_manifest_processed .*sequence=2"
+        " .*disposition=accepted",
+        since=rotated,
+        nodes=[2],
+    )
+    ctx.assert_log_order(
+        [
+            "manifest_validation single_manifest_processed .*sequence=2",
+            "manifest_validation repair_sent .*sequence=2",
+        ],
+        since=rotated,
+        nodes=[2],
+    )
+
+    ctx.assert_not_log("Validation forwarded by peer is invalid", nodes=nodes)
+    ctx.assert_not_log("Validation: Too small", nodes=nodes)
+
+    log(
+        "PASS: mid-run rotation to sequence 2 propagated through an old"
+        " relay; the upgraded observer superseded and admitted the new"
+        " manifest and re-armed its repair lane at the new sequence"
+    )

.testnet/scenarios/manifest_validation_mixed_binaries.py

@@ -0,0 +1,54 @@
+"""Exercise manifest/validation traffic across old and new binaries."""
+
+
+async def scenario(ctx, log):
+    nodes = [0, 1, 2]
+    expected = ctx.topology_edges([(0, 1), (1, 2)])
+
+    await ctx.apply_topology(expected, nodes=nodes, exact=False)
+    # The first close produces the validator traffic under test; later closes
+    # produce naked relays from the old middle node (it forwards a manifest
+    # singleton at most once), which must draw the bounded repair. These
+    # binaries are different product revisions, so multi-ledger convergence is
+    # deliberately not used as the protocol-compatibility oracle.
+    await ctx.wait_for_ledgers(3, node_id=0, timeout=180)
+
+    # n0 is the sole validator and uses the new ordered prerequisite path
+    # toward the old middle node. The old node processes and relays the two
+    # existing envelope types normally. n2 is a new observer and must retain
+    # the relayed candidate across unrelated traffic until its validation.
+    ctx.assert_log("manifest_validation pair_enqueued", nodes=[0])
+    ctx.assert_log("manifest_validation candidate_staged", nodes=[2])
+    ctx.assert_log("manifest_validation candidate_matched", nodes=[2])
+    ctx.assert_log("manifest_validation validation_parsed", nodes=[2])
+    ctx.assert_log("manifest_validation single_manifest_processed", nodes=[2])
+    ctx.assert_log_order(
+        [
+            "manifest_validation candidate_staged",
+            "manifest_validation candidate_matched",
+            "manifest_validation single_manifest_processed",
+        ],
+        nodes=[2],
+    )
+
+    # After durable admission, the old relay's later validations arrive
+    # naked; an authenticated naked validation is an implicit request, so
+    # the upgraded observer repairs its sender once per master/sequence.
+    ctx.assert_log("manifest_validation repair_sent", nodes=[2])
+    ctx.assert_log_order(
+        [
+            "manifest_validation single_manifest_processed",
+            "manifest_validation repair_sent",
+        ],
+        nodes=[2],
+    )
+
+    ctx.assert_not_log("Validation forwarded by peer is invalid", nodes=nodes)
+    ctx.assert_not_log("Validation: Too small", nodes=nodes)
+
+    log(
+        "PASS: a release relay accepted the upgraded sender's existing "
+        "manifest/validation envelopes; the upgraded observer retained, "
+        "matched, verified, and admitted the relayed prerequisite, then "
+        "repaired the old middle node's later naked validations"
+    )

.testnet/scenarios/manifest_validation_order.py

@@ -0,0 +1,69 @@
+"""Manifest packets precede validations across a real two-hop relay."""
+
+
+async def scenario(ctx, log):
+    nodes = [0, 1, 2]
+    expected = ctx.topology_edges([(0, 1), (1, 2)])
+
+    ctx.mark("before-manifest-validation-connect")
+    # A live TCP peer session is visible from both endpoints, so require the
+    # two links without treating their reverse views as extra connections.
+    await ctx.apply_topology(expected, nodes=nodes, exact=False)
+    await ctx.wait_for_ledgers(3, timeout=90)
+    ctx.mark("after-manifest-validation-ledgers")
+
+    send_pattern = r"manifest_validation send_prerequisite"
+    pair_pattern = r"manifest_validation pair_enqueued .*order=manifest,validation"
+    staged_pattern = r"manifest_validation candidate_staged"
+    matched_pattern = r"manifest_validation candidate_matched"
+    processed_pattern = r"manifest_validation single_manifest_processed"
+    validation_pattern = r"manifest_validation validation_parsed"
+
+    # This is a fresh per-test network, so the complete node logs are the
+    # scenario range. Avoid timestamp filtering: xahaud's custom local-time
+    # prefix is not yet understood by x-testnet's Marker parser.
+    for node in nodes:
+        sent = ctx.assert_log(send_pattern, nodes=[node])
+        paired = ctx.assert_log(pair_pattern, nodes=[node])
+        assert paired.count == sent.count, (
+            f"n{node} logged {sent.count} prerequisite sends but "
+            f"{paired.count} completed pairs"
+        )
+        ctx.assert_log_order([send_pattern, pair_pattern], nodes=[node])
+
+        staged = ctx.assert_log(staged_pattern, nodes=[node])
+        matched = ctx.assert_log(matched_pattern, nodes=[node])
+        assert matched.count <= staged.count, (
+            f"n{node} matched {matched.count} candidates after staging only "
+            f"{staged.count}"
+        )
+        ctx.assert_log(validation_pattern, nodes=[node])
+        ctx.assert_log_order(
+            [
+                staged_pattern,
+                validation_pattern,
+                matched_pattern,
+            ],
+            nodes=[node],
+        )
+
+    # n0 originates the sole listed validator's manifest and already owns it.
+    # The relay and observer must each admit that manifest; other valid pairs
+    # may remain ephemeral, so matched and durably processed counts are not
+    # expected to be equal.
+    for node in [1, 2]:
+        processed = ctx.assert_log(processed_pattern, nodes=[node])
+        matched = ctx.assert_log(matched_pattern, nodes=[node])
+        assert processed.count <= matched.count, (
+            f"n{node} processed {processed.count} manifests after matching "
+            f"only {matched.count} candidates"
+        )
+        ctx.assert_log_order(
+            [staged_pattern, matched_pattern, processed_pattern], nodes=[node]
+        )
+
+    log(
+        "PASS: every live peer enqueued manifest before validation and every "
+        "receiver staged and matched its one candidate, then applied it only "
+        "after the associated validation passed signature verification"
+    )

.testnet/scenarios/manifest_validation_order.yml

@@ -0,0 +1,54 @@
+defaults:
+  network:
+    node_count: 3
+    validators: 1
+    launcher: tmux
+    fixed_peers: false
+    log_levels:
+      Protocol: debug
+
+tests:
+  - name: manifest_validation_order
+    script: .testnet/scenarios/manifest_validation_order.py
+
+  - name: manifest_validation_mixed_binaries
+    script: .testnet/scenarios/manifest_validation_mixed_binaries.py
+    network:
+      validators: 1
+      node_binaries:
+        1: "@release-3350"
+      log_levels:
+        Protocol: debug
+        Validations: debug
+
+  - name: manifest_rotation_propagation
+    script: .testnet/scenarios/manifest_rotation_propagation.py
+    network:
+      validators: 1
+      node_binaries:
+        1: "@release-3350"
+      log_levels:
+        Protocol: debug
+        Validations: debug
+
+  - name: manifest_revocation_live
+    script: .testnet/scenarios/manifest_revocation_live.py
+    network:
+      node_count: 6
+      validators: 5
+      node_binaries:
+        5: "@release-3350"
+      log_levels:
+        Protocol: debug
+        Validations: debug
+
+  - name: manifest_wipe_recovery
+    script: .testnet/scenarios/manifest_wipe_recovery.py
+    network:
+      node_count: 4
+      validators: 1
+      node_binaries:
+        1: "@release-3350"
+      log_levels:
+        Protocol: debug
+        Validations: debug

.testnet/scenarios/manifest_wipe_recovery.py

@@ -0,0 +1,99 @@
+"""Wipe a mid-chain node's wallet and watch both generations heal it.
+
+The forever cache's informal replicated history is gone from new nodes, so
+this pins what replaces it. A wiped upgraded node reconnects with empty
+durable state; its old-release upstream re-seeds it through the legacy
+connect-time dump lane, its own validation traffic re-pairs through
+always-send, and it resumes forwarding paired prerequisites downstream. The
+recovery, not a starvation, is the honest live boundary: every reachable
+topology heals, because old peers dump at connect and new peers re-send the
+prerequisite with every validation.
+"""
+
+from xahaud_scripts.testnet.scenario import (
+    AssertionError as ScenarioAssertion,
+)
+
+
+async def _await_log(ctx, log, pattern, *, nodes, name, since=None, deadline=36):
+    # Catch the runner's ScenarioAssertion, not the builtin. The scenario
+    # module shadows AssertionError without subclassing it.
+    for attempt in range(deadline):
+        try:
+            ctx.assert_log(pattern, nodes=nodes, **({"since": since} if since else {}))
+            return
+        except ScenarioAssertion:
+            if attempt == deadline - 1:
+                raise
+            if attempt % 6 == 5:
+                log(f"{name}: attempt {attempt + 1}/{deadline}")
+            await ctx.sleep(5, name=name)
+
+
+async def scenario(ctx, log):
+    nodes = [0, 1, 2, 3]
+    expected = ctx.topology_edges([(0, 1), (1, 2), (2, 3)])
+
+    await ctx.apply_topology(expected, nodes=nodes, exact=False)
+
+    # Baseline: knowledge reaches the end of the chain. The tail observer
+    # receives paired traffic from the upgraded mid-chain node. Admission at
+    # the tail implies the whole upstream chain, so wait on that log fact —
+    # three hops can trail the validator's fast-bootstrap ledger count.
+    await ctx.wait_for_ledgers(2, node_id=0, timeout=120)
+    await _await_log(
+        ctx,
+        log,
+        "manifest_validation single_manifest_processed .*sequence=1",
+        nodes=[3],
+        name="await-baseline-chain",
+        deadline=24,
+    )
+    ctx.assert_log(
+        "manifest_validation single_manifest_processed .*sequence=1",
+        nodes=[2],
+    )
+
+    wiped = ctx.mark("wiped")
+    await ctx.restart_node(2, wipe_wallet_db=True)
+    log("restarted n2 with a wiped wallet database")
+
+    await ctx.wait_for_ledgers(3, node_id=0, timeout=180)
+
+    # Wait on the terminal log fact: resumed paired forwarding downstream is
+    # the last event in the recovery chain, so re-learning and re-admission
+    # must precede it.
+    await _await_log(
+        ctx,
+        log,
+        "manifest_validation send_prerequisite .*sequence=1",
+        nodes=[2],
+        name="await-resumed-forwarding",
+        since=wiped,
+        deadline=36,
+    )
+
+    # Recovery: the wiped node re-learned the validator identity from live
+    # traffic (the old upstream's connect-time dump arrives as a singleton
+    # candidate; the next validation proves it) and admitted it durably
+    # again.
+    ctx.assert_log(
+        "manifest_validation candidate_staged .*sequence=1",
+        since=wiped,
+        nodes=[2],
+    )
+    ctx.assert_log(
+        "manifest_validation single_manifest_processed .*sequence=1"
+        " .*disposition=accepted",
+        since=wiped,
+        nodes=[2],
+    )
+
+    ctx.assert_not_log("Validation forwarded by peer is invalid", nodes=nodes)
+    ctx.assert_not_log("Validation: Too small", nodes=nodes)
+
+    log(
+        "PASS: a wallet-wiped mid-chain node re-learned the validator"
+        " identity from live traffic through an old upstream, re-admitted it"
+        " durably, and resumed paired forwarding to the tail observer"
+    )

Builds/levelization/results/ordering.txt

@@ -77,6 +77,11 @@ test.ledger > xrpld.app
 test.ledger > xrpld.core
 test.ledger > xrpld.ledger
 test.ledger > xrpl.protocol
+test.net > test.toplevel
+test.net > xrpl.basics
+test.net > xrpld.core
+test.net > xrpld.net
+test.net > xrpl.json
 test.nodestore > test.jtx
 test.nodestore > test.toplevel
 test.nodestore > test.unit_test

cmake/RippledCore.cmake

@@ -179,121 +179,7 @@ if(xrpld)
     file(GLOB_RECURSE sources CONFIGURE_DEPENDS
       "${CMAKE_CURRENT_SOURCE_DIR}/src/test/*.cpp"
     )
-    if(HOOKS_TEST_ONLY OR DEFINED ENV{HOOKS_TEST_ONLY})
-      # Keep test infra but drop the individual *_test.cpp files
-      list(FILTER sources EXCLUDE REGEX "_test\\.cpp$")
-      message(STATUS "HOOKS_TEST_ONLY: excluded *_test.cpp from src/test/")
-    endif()
     target_sources(rippled PRIVATE ${sources})
-
-    # Optional: include external hook test sources from another directory.
-    # Set via -DHOOKS_TEST_DIR=/path/to/tests or env HOOKS_TEST_DIR.
-    # Optionally set HOOKS_C_DIR to pass --hooks-c-dir args to the compiler
-    # (e.g. "tipbot=/path/to/hooks" — multiple values separated by ";").
-    #
-    # hookz build-test-hooks must be on PATH. It auto-compiles hooks referenced
-    # in each *_test.cpp and generates *_test_hooks.h next to the test file.
-    if(NOT HOOKS_TEST_DIR AND DEFINED ENV{HOOKS_TEST_DIR})
-      set(HOOKS_TEST_DIR $ENV{HOOKS_TEST_DIR})
-    endif()
-    if(NOT HOOKS_C_DIR AND DEFINED ENV{HOOKS_C_DIR})
-      set(HOOKS_C_DIR $ENV{HOOKS_C_DIR})
-    endif()
-    if(HOOKS_TEST_DIR AND EXISTS "${HOOKS_TEST_DIR}")
-      file(GLOB EXTERNAL_HOOK_TESTS CONFIGURE_DEPENDS
-        "${HOOKS_TEST_DIR}/*_test.cpp"
-      )
-      if(EXTERNAL_HOOK_TESTS)
-        # Build extra args for hookz build-test-hooks
-        set(_hooks_extra_args "")
-        set(_hooks_source_deps "")
-        if(HOOKS_C_DIR)
-          foreach(_dir ${HOOKS_C_DIR})
-            list(APPEND _hooks_extra_args "--hooks-c-dir" "${_dir}")
-
-            string(REGEX REPLACE "^[^=]+=" "" _hook_dir "${_dir}")
-            if(EXISTS "${_hook_dir}")
-              file(GLOB_RECURSE _hook_dir_deps CONFIGURE_DEPENDS
-                "${_hook_dir}/*.c"
-                "${_hook_dir}/*.h"
-              )
-              if(HOOKS_TEST_DIR)
-                list(FILTER _hook_dir_deps EXCLUDE REGEX "^${HOOKS_TEST_DIR}/")
-              endif()
-              list(APPEND _hooks_source_deps ${_hook_dir_deps})
-            endif()
-          endforeach()
-          list(REMOVE_DUPLICATES _hooks_source_deps)
-        endif()
-        if(HOOKS_COVERAGE OR DEFINED ENV{HOOKS_COVERAGE})
-          list(APPEND _hooks_extra_args "--hook-coverage")
-          message(STATUS "Hook coverage enabled: compiling hooks with hookz")
-        endif()
-        set(_hooks_always_run OFF)
-        if(HOOKS_FORCE_RECOMPILE OR DEFINED ENV{HOOKS_FORCE_RECOMPILE})
-          list(APPEND _hooks_extra_args "--force-write" "--no-cache")
-          set(_hooks_always_run ON)
-          message(STATUS "Hook force recompile enabled (cache bypassed)")
-        endif()
-        if(HOOKZ_BUILDBOX OR "$ENV{HOOKZ_BUILDBOX}" STREQUAL "1")
-          # Remote evidence must observe the service on every build. An
-          # OUTPUT-cached header may otherwise preserve locally built WASM
-          # after the compiler mode changes.
-          list(APPEND _hooks_extra_args "--buildbox")
-          if(NOT _hooks_always_run)
-            list(APPEND _hooks_extra_args "--force-write")
-          endif()
-          set(_hooks_always_run ON)
-          message(STATUS "Canonical buildbox enabled (always recompiled)")
-        endif()
-
-        # Run hookz build-test-hooks on each test file before compilation
-        foreach(_test_file ${EXTERNAL_HOOK_TESTS})
-          get_filename_component(_stem ${_test_file} NAME_WE)
-          set(_hooks_header "${HOOKS_TEST_DIR}/${_stem}_hooks.h")
-          if(_hooks_always_run)
-            # Always run — no DEPENDS, no OUTPUT caching
-            add_custom_target(compile_hooks_${_stem} ALL
-              COMMAND hookz build-test-hooks "${_test_file}" ${_hooks_extra_args}
-              WORKING_DIRECTORY "${CMAKE_SOURCE_DIR}"
-              COMMENT "Compiling hooks for ${_stem} (forced)"
-              VERBATIM
-            )
-            list(APPEND EXTERNAL_HOOK_TARGETS compile_hooks_${_stem})
-          else()
-            add_custom_command(
-              OUTPUT "${_hooks_header}"
-              COMMAND hookz build-test-hooks "${_test_file}" ${_hooks_extra_args}
-              DEPENDS "${_test_file}" ${_hooks_source_deps}
-              WORKING_DIRECTORY "${CMAKE_SOURCE_DIR}"
-              COMMENT "Compiling hooks for ${_stem}"
-              VERBATIM
-            )
-            list(APPEND EXTERNAL_HOOK_HEADERS "${_hooks_header}")
-          endif()
-        endforeach()
-
-        # Ensure headers are generated before rippled compiles
-        if(_hooks_always_run)
-          foreach(_tgt ${EXTERNAL_HOOK_TARGETS})
-            add_dependencies(rippled ${_tgt})
-          endforeach()
-        else()
-          add_custom_target(compile_external_hooks DEPENDS ${EXTERNAL_HOOK_HEADERS})
-          add_dependencies(rippled compile_external_hooks)
-        endif()
-
-        target_sources(rippled PRIVATE ${EXTERNAL_HOOK_TESTS})
-        # Keep the generated hook-header include path scoped to the external
-        # test sources so changing HOOKS_TEST_DIR doesn't invalidate the
-        # compile command for the rest of rippled.
-        set_property(
-          SOURCE ${EXTERNAL_HOOK_TESTS}
-          APPEND PROPERTY INCLUDE_DIRECTORIES "${HOOKS_TEST_DIR}"
-        )
-        message(STATUS "Including external hook tests from: ${HOOKS_TEST_DIR}")
-      endif()
-    endif()
   endif()
 
   target_link_libraries(rippled

include/xrpl/basics/Log.h

@@ -25,7 +25,6 @@
 #include <boost/beast/core/string.hpp>
 #include <boost/filesystem.hpp>
 #include <fstream>
-#include <functional>
 #include <map>
 #include <memory>
 #include <mutex>
@@ -166,7 +165,6 @@ private:
     beast::severities::Severity thresh_;
     File file_;
     bool silent_ = false;
-    std::function<std::string(std::string const&)> transform_;
 
 public:
     Logs(beast::severities::Severity level);
@@ -205,33 +203,6 @@ public:
         std::string const& text,
         bool console);
 
-    /** Set a transform applied to every log message before output.
-     *  Useful in tests to replace raw account IDs with human-readable names.
-     *  Pass nullptr to clear.
-     *
-     *  TODO: This is test-only infrastructure (used by TestEnv). Consider
-     *  moving to SuiteLogs or a test-specific subclass if the Logs interface
-     *  needs to stay clean for production.
-     */
-    void
-    setTransform(std::function<std::string(std::string const&)> fn)
-    {
-        std::lock_guard lock(mutex_);
-        transform_ = std::move(fn);
-    }
-
-    /** Apply the current transform to text (or return as-is if none set). */
-    std::string const&
-    applyTransform(std::string const& text) const
-    {
-        if (!transform_)
-            return text;
-        // Store in thread_local to return a const ref
-        thread_local std::string buf;
-        buf = transform_(text);
-        return buf;
-    }
-
     std::string
     rotate();
 

include/xrpl/hook/Enum.h

@@ -417,7 +417,6 @@ getImportWhitelist(Rules const& rules)
 #define int64_t 0x7EU
 #define int32_t 0x7FU
 #define uint32_t 0x7FU
-#define void_t 0x00U
 
 #define HOOK_WRAP_PARAMS(...) __VA_ARGS__
 
@@ -429,15 +428,11 @@ getImportWhitelist(Rules const& rules)
 
 #include "hook_api.macro"
 
-    // Coverage callback: void __on_source_line(uint32_t line, uint32_t col)
-    whitelist["__on_source_line"] = {void_t, uint32_t, uint32_t};
-
 #undef HOOK_API_DEFINITION
 #undef HOOK_WRAP_PARAMS
 #undef int64_t
 #undef int32_t
 #undef uint32_t
-#undef void_t
 #pragma pop_macro("HOOK_API_DEFINITION")
 
     return whitelist;

include/xrpl/hook/Guard.h

@@ -1383,51 +1383,21 @@ validateGuards(
                 int result_count = parseLeb128(wasm, i, &i);
                 CHECK_SHORT_HOOK();
 
-                if (j == hook_type_idx)
+                // this needs a reliable hook cleaner otherwise it will catch
+                // most compilers out
+                if (result_count != 1)
                 {
-                    // hook/cbak must return exactly one value (i64)
-                    if (result_count != 1)
-                    {
-                        GUARDLOG(hook::log::FUNC_RETURN_COUNT)
-                            << "Malformed transaction. "
-                            << "hook/cbak function type must return exactly "
-                               "one value. "
-                            << "\n";
-                        return {};
-                    }
-                }
-                else if (first_signature)
-                {
-                    // For whitelisted imports, check expected return count.
-                    // void_t (0x00) means 0 return values.
-                    uint8_t expected_return = (*first_signature).get()[0];
-                    int expected_result_count =
-                        (expected_return == 0x00U) ? 0 : 1;
-                    if (result_count != expected_result_count)
-                    {
-                        GUARDLOG(hook::log::FUNC_RETURN_COUNT)
-                            << "Malformed transaction. "
-                            << "Hook API: " << *first_name
-                            << " has wrong return count "
-                            << "(expected " << expected_result_count << ", got "
-                            << result_count << ")."
-                            << "\n";
-                        return {};
-                    }
-                }
-                else
-                {
-                    if (result_count != 1)
-                    {
-                        GUARDLOG(hook::log::FUNC_RETURN_COUNT)
-                            << "Malformed transaction. "
-                            << "Hook declares a function type that returns "
-                               "fewer or more than one value. "
-                            << "\n";
-                        return {};
-                    }
+                    GUARDLOG(hook::log::FUNC_RETURN_COUNT)
+                        << "Malformed transaction. "
+                        << "Hook declares a function type that returns fewer "
+                           "or more than one value. "
+                        << "\n";
+                    return {};
                 }
 
+                // this can only ever be 1 in production, but in testing it may
+                // also be 0 or >1 so for completeness this loop is here but can
+                // be taken out in prod
                 for (int k = 0; k < result_count; ++k)
                 {
                     int result_type = parseLeb128(wasm, i, &i);

include/xrpl/hook/Macro.h

@@ -150,23 +150,22 @@
         WasmEdge_CallingFrameContext const& frameCtx __VA_OPT__(               \
             COMMA __VA_ARGS__))
 
-#define HOOK_SETUP()                                                   \
-    using enum hook_api::hook_return_code;                             \
-    try                                                                \
-    {                                                                  \
-        [[maybe_unused]] ApplyContext& applyCtx = hookCtx.applyCtx;    \
-        [[maybe_unused]] auto& view = applyCtx.view();                 \
-        [[maybe_unused]] auto j = applyCtx.app.journal("View");        \
-        [[maybe_unused]] auto jh = applyCtx.app.journal("HooksTrace"); \
-        [[maybe_unused]] WasmEdge_MemoryInstanceContext* memoryCtx =   \
-            WasmEdge_CallingFrameGetMemoryInstance(&frameCtx, 0);      \
-        [[maybe_unused]] unsigned char* memory =                       \
-            WasmEdge_MemoryInstanceGetPointer(memoryCtx, 0, 0);        \
-        [[maybe_unused]] const uint64_t memory_length =                \
-            WasmEdge_MemoryInstanceGetPageSize(memoryCtx) *            \
-            WasmEdge_kPageSize;                                        \
-        [[maybe_unused]] auto& api = hookCtx.api();                    \
-        if (!memoryCtx || !memory || !memory_length)                   \
+#define HOOK_SETUP()                                                 \
+    using enum hook_api::hook_return_code;                           \
+    try                                                              \
+    {                                                                \
+        [[maybe_unused]] ApplyContext& applyCtx = hookCtx.applyCtx;  \
+        [[maybe_unused]] auto& view = applyCtx.view();               \
+        [[maybe_unused]] auto j = applyCtx.app.journal("View");      \
+        [[maybe_unused]] WasmEdge_MemoryInstanceContext* memoryCtx = \
+            WasmEdge_CallingFrameGetMemoryInstance(&frameCtx, 0);    \
+        [[maybe_unused]] unsigned char* memory =                     \
+            WasmEdge_MemoryInstanceGetPointer(memoryCtx, 0, 0);      \
+        [[maybe_unused]] const uint64_t memory_length =              \
+            WasmEdge_MemoryInstanceGetPageSize(memoryCtx) *          \
+            WasmEdge_kPageSize;                                      \
+        [[maybe_unused]] auto& api = hookCtx.api();                  \
+        if (!memoryCtx || !memory || !memory_length)                 \
             return INTERNAL_ERROR;
 
 #define HOOK_TEARDOWN()                                        \

src/libxrpl/basics/Log.cpp

@@ -196,10 +196,9 @@ Logs::write(
     std::string const& text,
     bool console)
 {
-    std::lock_guard lock(mutex_);
-    std::string const& transformed = transform_ ? transform_(text) : text;
     std::string s;
-    format(s, transformed, level, partition);
+    format(s, text, level, partition);
+    std::lock_guard lock(mutex_);
     file_.writeln(s);
     if (!silent_)
         std::cerr << s << '\n';

src/test/app/Manifest_test.cpp

@@ -454,6 +454,8 @@ public:
         auto const sk = randomSecretKey();
         auto const pk = derivePublicKey(KeyType::ed25519, sk);
 
+        BEAST_EXPECT(!cache.getManifestSnapshot(pk));
+
         // getSigningKey should return same key if there is no manifest
         BEAST_EXPECT(cache.getSigningKey(pk) == pk);
 
@@ -468,6 +470,17 @@ public:
                 sk, KeyType::ed25519, kp0.second, KeyType::secp256k1, 0)));
         BEAST_EXPECT(cache.getSigningKey(pk) == kp0.first);
         BEAST_EXPECT(cache.getMasterKey(kp0.first) == pk);
+        if (auto const snapshot = cache.getManifestSnapshot(kp0.first))
+        {
+            BEAST_EXPECT(snapshot->masterKey == pk);
+            BEAST_EXPECT(snapshot->signingKey == kp0.first);
+            BEAST_EXPECT(snapshot->sequence == 0);
+            BEAST_EXPECT(!snapshot->revoked());
+        }
+        else
+        {
+            fail("current signing key resolves its manifest snapshot");
+        }
 
         // getSigningKey should return the latest ephemeral public key
         // for the listed validator master public key
@@ -481,6 +494,16 @@ public:
         BEAST_EXPECT(cache.getSigningKey(pk) == kp1.first);
         BEAST_EXPECT(cache.getMasterKey(kp1.first) == pk);
         BEAST_EXPECT(cache.getMasterKey(kp0.first) == kp0.first);
+        BEAST_EXPECT(!cache.getManifestSnapshot(kp0.first));
+        if (auto const snapshot = cache.getManifestSnapshot(pk))
+        {
+            BEAST_EXPECT(snapshot->signingKey == kp1.first);
+            BEAST_EXPECT(snapshot->sequence == 1);
+        }
+        else
+        {
+            fail("master key resolves its current manifest snapshot");
+        }
 
         // getSigningKey and getMasterKey should fail if a new manifest is
         // applied with the same signing key but a higher sequence
@@ -502,6 +525,16 @@ public:
         BEAST_EXPECT(cache.getSigningKey(pk) == pk);
         BEAST_EXPECT(cache.getMasterKey(kp0.first) == kp0.first);
         BEAST_EXPECT(cache.getMasterKey(kp1.first) == kp1.first);
+        if (auto const snapshot = cache.getManifestSnapshot(pk))
+        {
+            BEAST_EXPECT(snapshot->revoked());
+            BEAST_EXPECT(!snapshot->signingKey);
+        }
+        else
+        {
+            fail("master key resolves its revocation snapshot");
+        }
+        BEAST_EXPECT(!cache.getManifestSnapshot(kp1.first));
     }
 
     void

src/test/jtx/Env.h

@@ -106,8 +106,7 @@ public:
         std::string const& partition,
         beast::severities::Severity threshold) override
     {
-        return std::make_unique<SuiteJournalSink>(
-            partition, threshold, suite_, this);
+        return std::make_unique<SuiteJournalSink>(partition, threshold, suite_);
     }
 };
 

src/test/jtx/TestEnv.h

@@ -1,148 +0,0 @@
-#ifndef TEST_JTX_TESTENV_H_INCLUDED
-#define TEST_JTX_TESTENV_H_INCLUDED
-
-#include <test/jtx/Env.h>
-#include <xrpl/basics/Log.h>
-#include <xrpl/protocol/AccountID.h>
-#include <cstdlib>
-#include <cstring>
-#include <map>
-#include <sstream>
-#include <string>
-
-namespace ripple {
-namespace test {
-namespace jtx {
-
-/**
- * TestEnv wraps Env with:
- *   - Named account registry: env.account("alice")
- *   - Auto log transform: replaces r-addresses with Account(name) in log output
- *   - Env-var driven per-partition log levels via TESTENV_LOGGING
- *
- * Usage:
- *   TestEnv env{suite, features};
- *   auto const& alice = env.account("alice");
- *   auto const& bob = env.account("bob");
- *   env.fund(XRP(10000), alice, bob);
- *   // Logs now show Account(alice), Account(bob) instead of r-addresses
- *
- * Log levels via env var:
- *   TESTENV_LOGGING="HooksTrace=trace,View=debug"
- *
- * Valid levels: trace, debug, info, warning, error, fatal
- */
-class TestEnv : public Env
-{
-    std::map<std::string, Account> accounts_;
-    std::string prefix_;
-
-public:
-    TestEnv(beast::unit_test::suite& suite, FeatureBitset features)
-        : Env(suite, features)
-    {
-        installTransform();
-        applyLoggingEnvVar();
-    }
-
-    TestEnv(
-        beast::unit_test::suite& suite,
-        std::unique_ptr<Config> config,
-        FeatureBitset features,
-        std::unique_ptr<Logs> logs = nullptr,
-        beast::severities::Severity thresh = beast::severities::kError)
-        : Env(suite, std::move(config), features, std::move(logs), thresh)
-    {
-        installTransform();
-        applyLoggingEnvVar();
-    }
-
-    ~TestEnv()
-    {
-        app().logs().setTransform(nullptr);
-    }
-
-    /// Get or create a named account.
-    /// First call creates the Account; subsequent calls return the same one.
-    Account const&
-    account(std::string const& name)
-    {
-        auto [it, inserted] = accounts_.try_emplace(name, name);
-        return it->second;
-    }
-
-    /// Set a prefix that appears at the start of every log line.
-    /// Useful for visually separating test phases in trace output.
-    /// Pass empty string to clear.
-    void
-    setPrefix(std::string const& prefix)
-    {
-        prefix_ = prefix.empty() ? "" : "[" + prefix + "] ";
-    }
-
-private:
-    static beast::severities::Severity
-    parseSeverity(std::string const& s)
-    {
-        if (s == "trace")
-            return beast::severities::kTrace;
-        if (s == "debug")
-            return beast::severities::kDebug;
-        if (s == "info")
-            return beast::severities::kInfo;
-        if (s == "warning")
-            return beast::severities::kWarning;
-        if (s == "error")
-            return beast::severities::kError;
-        if (s == "fatal")
-            return beast::severities::kFatal;
-        return beast::severities::kError;
-    }
-
-    void
-    applyLoggingEnvVar()
-    {
-        // Parse TESTENV_LOGGING="Partition1=level,Partition2=level"
-        auto const* envVal = std::getenv("TESTENV_LOGGING");
-        if (!envVal || !envVal[0])
-            return;
-
-        std::istringstream ss(envVal);
-        std::string pair;
-        while (std::getline(ss, pair, ','))
-        {
-            auto eq = pair.find('=');
-            if (eq == std::string::npos)
-                continue;
-            auto partition = pair.substr(0, eq);
-            auto level = pair.substr(eq + 1);
-            app().logs().get(partition).threshold(parseSeverity(level));
-        }
-    }
-
-    void
-    installTransform()
-    {
-        app().logs().setTransform([this](std::string const& text) {
-            std::string out = prefix_ + text;
-            for (auto const& [name, acc] : accounts_)
-            {
-                auto raddr = toBase58(acc.id());
-                std::string::size_type pos = 0;
-                std::string replacement = "Account(" + name + ")";
-                while ((pos = out.find(raddr, pos)) != std::string::npos)
-                {
-                    out.replace(pos, raddr.size(), replacement);
-                    pos += replacement.size();
-                }
-            }
-            return out;
-        });
-    }
-};
-
-}  // namespace jtx
-}  // namespace test
-}  // namespace ripple
-
-#endif

src/test/net/RPCSub_test.cpp

@@ -0,0 +1,351 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2024 Ripple Labs Inc.
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include <test/jtx.h>
+#include <xrpld/core/Job.h>
+#include <xrpld/core/JobQueue.h>
+#include <xrpld/net/RPCSub.h>
+#include <xrpl/json/json_value.h>
+
+#include <boost/asio.hpp>
+#include <boost/asio/ip/tcp.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <memory>
+#include <string>
+#include <thread>
+
+namespace ripple {
+namespace test {
+
+// Minimal HTTP endpoint that counts received webhook POSTs and replies
+// with a configurable status. Responses are EOF-delimited (no
+// Content-Length) and the socket is closed right after writing — the
+// exact shape that triggered the original handleData EOF-completion
+// leak. So these tests exercise RPCSub flow control AND the HTTPClient
+// EOF fix end to end: if either regressed, delivery would stall and the
+// expected count would never be reached within the timeout.
+class MockWebhookEndpoint
+{
+    boost::asio::io_service ios_;
+    std::unique_ptr<boost::asio::io_service::work> work_;
+    boost::asio::ip::tcp::acceptor acceptor_;
+    std::thread thread_;
+    unsigned short port_;
+
+    std::atomic<int> received_{0};
+    std::atomic<int> status_{200};
+    std::atomic<int> delayMs_{0};
+
+public:
+    MockWebhookEndpoint()
+        : work_(std::make_unique<boost::asio::io_service::work>(ios_))
+        , acceptor_(
+              ios_,
+              boost::asio::ip::tcp::endpoint(
+                  boost::asio::ip::address::from_string("127.0.0.1"),
+                  0))
+    {
+        port_ = acceptor_.local_endpoint().port();
+        accept();
+        thread_ = std::thread([this] { ios_.run(); });
+    }
+
+    ~MockWebhookEndpoint()
+    {
+        work_.reset();
+        boost::system::error_code ec;
+        acceptor_.close(ec);
+        ios_.stop();
+        if (thread_.joinable())
+            thread_.join();
+    }
+
+    unsigned short
+    port() const
+    {
+        return port_;
+    }
+
+    int
+    received() const
+    {
+        return received_;
+    }
+
+    void
+    setStatus(int s)
+    {
+        status_ = s;
+    }
+
+    // Delay each reply so delivery is deterministically slower than the
+    // microsecond-fast enqueue loop — keeps the deque full for the
+    // queue-cap drop test regardless of scheduling.
+    void
+    setResponseDelay(int ms)
+    {
+        delayMs_ = ms;
+    }
+
+private:
+    void
+    accept()
+    {
+        auto sock = std::make_shared<boost::asio::ip::tcp::socket>(ios_);
+        acceptor_.async_accept(*sock, [this, sock](auto ec) {
+            if (ec)
+                return;
+            handle(sock);
+            accept();
+        });
+    }
+
+    void
+    handle(std::shared_ptr<boost::asio::ip::tcp::socket> sock)
+    {
+        auto buf = std::make_shared<boost::asio::streambuf>();
+        boost::asio::async_read_until(
+            *sock, *buf, "\r\n\r\n", [this, sock, buf](auto ec, std::size_t) {
+                if (ec)
+                    return;
+
+                ++received_;
+
+                auto const delay = delayMs_.load();
+                if (delay > 0)
+                {
+                    auto timer =
+                        std::make_shared<boost::asio::steady_timer>(ios_);
+                    timer->expires_from_now(std::chrono::milliseconds(delay));
+                    timer->async_wait(
+                        [this, sock, timer](auto) { reply(sock); });
+                }
+                else
+                {
+                    reply(sock);
+                }
+            });
+    }
+
+    void
+    reply(std::shared_ptr<boost::asio::ip::tcp::socket> sock)
+    {
+        // EOF-delimited reply: no Content-Length, close after writing.
+        // This is the realistic failing-webhook shape.
+        auto resp = std::make_shared<std::string>(
+            "HTTP/1.0 " + std::to_string(status_.load()) +
+            " Reply\r\n\r\n{\"result\":{}}");
+        boost::asio::async_write(
+            *sock, boost::asio::buffer(*resp), [sock, resp](auto, std::size_t) {
+                boost::system::error_code ig;
+                sock->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ig);
+                sock->close(ig);
+            });
+    }
+};
+
+//------------------------------------------------------------------------------
+
+class RPCSub_test : public beast::unit_test::suite
+{
+    // Generous ceiling: the instrumented Debug (coverage) build is much
+    // slower than Release, so timeouts are sized for that, not Release.
+    template <class Cond>
+    bool
+    waitFor(Cond cond, std::chrono::seconds timeout = std::chrono::seconds{30})
+    {
+        auto const deadline = std::chrono::steady_clock::now() + timeout;
+        while (!cond() && std::chrono::steady_clock::now() < deadline)
+            std::this_thread::sleep_for(std::chrono::milliseconds(10));
+        return cond();
+    }
+
+    std::shared_ptr<RPCSub>
+    makeSub(
+        jtx::Env& env,
+        MockWebhookEndpoint& ep,
+        std::size_t maxQueueSize = 16384)
+    {
+        return make_RPCSub(
+            env.app().getOPs(),
+            env.app().getJobQueue(),
+            "http://127.0.0.1:" + std::to_string(ep.port()) + "/",
+            "",
+            "",
+            env.app().logs(),
+            maxQueueSize);
+    }
+
+    // True once no RPCSub sending job is queued or running. sendThread
+    // captures a raw `this`, so the RPCSub must not be destroyed while a
+    // job is still in flight — wait on this before letting the sub die.
+    bool
+    sendingIdle(jtx::Env& env)
+    {
+        return env.app().getJobQueue().getJobCountTotal(jtCLIENT_SUBSCRIBE) ==
+            0;
+    }
+
+    // Wait for all events to reach the endpoint AND the sending job to
+    // finish, so the sub can be torn down without racing sendThread.
+    void
+    drainAndSettle(jtx::Env& env, MockWebhookEndpoint& ep, int expected)
+    {
+        bool const delivered =
+            waitFor([&] { return ep.received() >= expected; });
+        bool const idle = waitFor([&] { return sendingIdle(env); });
+        log << "  drainAndSettle: received=" << ep.received() << "/" << expected
+            << " idle=" << idle << std::endl;
+        BEAST_EXPECT(delivered);
+        BEAST_EXPECT(idle);
+    }
+
+    void
+    send(std::shared_ptr<RPCSub> const& sub, int n)
+    {
+        Json::Value ev(Json::objectValue);
+        ev["n"] = n;
+        sub->send(ev, false);
+    }
+
+    void
+    testDelivery()
+    {
+        testcase("Webhook events are delivered");
+
+        using namespace jtx;
+        Env env{*this};
+        MockWebhookEndpoint ep;
+
+        static constexpr int N = 10;
+        {
+            auto sub = makeSub(env, ep);
+            for (int i = 0; i < N; ++i)
+                send(sub, i);
+            drainAndSettle(env, ep, N);
+        }
+
+        BEAST_EXPECT(ep.received() == N);
+    }
+
+    void
+    testErrorsDoNotStall()
+    {
+        testcase("Delivery continues when endpoint returns HTTP 500");
+
+        // The original bug (xrpld #6341): an endpoint returning errors
+        // without Content-Length never completed, stalling delivery to
+        // ALL subscribers. Here every response is a 500 with no
+        // Content-Length (EOF-delimited) — all N must still arrive.
+        using namespace jtx;
+        Env env{*this};
+        MockWebhookEndpoint ep;
+        ep.setStatus(500);
+
+        static constexpr int N = 10;
+        {
+            auto sub = makeSub(env, ep);
+            for (int i = 0; i < N; ++i)
+                send(sub, i);
+            drainAndSettle(env, ep, N);
+        }
+
+        BEAST_EXPECT(ep.received() == N);
+    }
+
+    void
+    testRestartAfterDrain()
+    {
+        testcase("Sending restarts after the queue drains");
+
+        // After a batch drains, sendThread clears mSending and returns.
+        // A later send() must start a fresh sending job; if mSending were
+        // left set (the #6341 failure mode) the second burst would never
+        // be delivered.
+        using namespace jtx;
+        Env env{*this};
+        MockWebhookEndpoint ep;
+
+        {
+            auto sub = makeSub(env, ep);
+
+            // First burst, then wait for the sending job to fully drain
+            // and exit (mSending cleared) — deterministically, not via a
+            // sleep.
+            for (int i = 0; i < 5; ++i)
+                send(sub, i);
+            drainAndSettle(env, ep, 5);
+
+            // Second burst must start a fresh sending job.
+            for (int i = 5; i < 10; ++i)
+                send(sub, i);
+            drainAndSettle(env, ep, 10);
+        }
+
+        BEAST_EXPECT(ep.received() == 10);
+    }
+
+    void
+    testQueueCapDrops()
+    {
+        testcase("Events past the queue cap are dropped");
+
+        // With a tiny cap, pushing far more events than delivery can keep
+        // up with forces send() down the drop path: enqueue is microsecond
+        // -fast while each (delayed) HTTP delivery is a full round-trip, so
+        // the deque sits at the cap and excess events are dropped. The
+        // delay makes "delivery slower than enqueue" hold regardless of
+        // scheduling, so this isn't timing-dependent. We just need some
+        // delivered (cap works) and some dropped (drop path exercised).
+        using namespace jtx;
+        Env env{*this};
+        MockWebhookEndpoint ep;
+        ep.setResponseDelay(50);
+
+        static constexpr int pushed = 50;
+        {
+            auto sub = makeSub(env, ep, /*maxQueueSize*/ 2);
+            for (int i = 0; i < pushed; ++i)
+                send(sub, i);
+            BEAST_EXPECT(waitFor([&] { return sendingIdle(env); }));
+        }
+
+        log << "  queue cap: received " << ep.received() << "/" << pushed
+            << std::endl;
+        BEAST_EXPECT(ep.received() > 0);
+        BEAST_EXPECT(ep.received() < pushed);
+    }
+
+public:
+    void
+    run() override
+    {
+        testDelivery();
+        testErrorsDoNotStall();
+        testRestartAfterDrain();
+        testQueueCapDrops();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(RPCSub, net, ripple);
+
+}  // namespace test
+}  // namespace ripple

src/test/unit_test/SuiteJournal.h

@@ -19,7 +19,6 @@
 
 #ifndef TEST_UNIT_TEST_SUITE_JOURNAL_H
 #define TEST_UNIT_TEST_SUITE_JOURNAL_H
-#include <xrpl/basics/Log.h>
 #include <xrpl/beast/unit_test.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <mutex>
@@ -32,18 +31,13 @@ class SuiteJournalSink : public beast::Journal::Sink
 {
     std::string partition_;
     beast::unit_test::suite& suite_;
-    Logs* logs_ = nullptr;
 
 public:
     SuiteJournalSink(
         std::string const& partition,
         beast::severities::Severity threshold,
-        beast::unit_test::suite& suite,
-        Logs* logs = nullptr)
-        : Sink(threshold, false)
-        , partition_(partition + " ")
-        , suite_(suite)
-        , logs_(logs)
+        beast::unit_test::suite& suite)
+        : Sink(threshold, false), partition_(partition + " "), suite_(suite)
     {
     }
 
@@ -103,12 +97,11 @@ SuiteJournalSink::writeAlways(
     // Only write the string if the level at least equals the threshold.
     if (level >= threshold())
     {
-        std::string const& output = logs_ ? logs_->applyTransform(text) : text;
         // std::endl flushes → sync() → str()/str("") race in shared buffer →
         // crashes
         static std::mutex log_mutex;
         std::lock_guard lock(log_mutex);
-        suite_.log << s << partition_ << output << std::endl;
+        suite_.log << s << partition_ << text << std::endl;
     }
 }
 

src/xrpld/app/consensus/RCLConsensus.cpp

@@ -903,7 +903,7 @@ RCLConsensus::Adaptor::validate(
     // Broadcast to all our peers:
     protocol::TMValidation val;
     val.set_validation(serialized.data(), serialized.size());
-    app_.overlay().broadcast(val);
+    app_.overlay().broadcast(val, v->getSignerPublic());
 
     // Publish to all our subscribers:
     app_.getOPs().pubValidation(v);

src/xrpld/app/hook/applyHook.h

@@ -12,11 +12,9 @@
 #include <xrpl/protocol/TER.h>
 #include <xrpl/protocol/digest.h>
 #include <any>
-#include <fstream>
 #include <memory>
 #include <optional>
 #include <queue>
-#include <set>
 #include <utility>
 #include <vector>
 #include <wasmedge/wasmedge.h>
@@ -307,130 +305,6 @@ static WasmEdge_String hookFunctionName =
 // see: lib/system/allocator.cpp
 #define WasmEdge_kPageSize 65536ULL
 
-// --- Coverage infrastructure ---
-//
-// Global coverage accumulator keyed by hook hash. Persists across all hook
-// executions in the process. Each __on_source_line call records a (line, col)
-// pair under the executing hook's hash.
-//
-// Test API:
-//   hook::coverageReset()            — clear all accumulated data
-//   hook::coverageHits(hookHash)     — get hits for a specific hook
-//   hook::coverageLabel(hash, label) — register a human-readable label
-//   hook::coverageDump(path)         — write all data to a file
-//
-// The dump file format is:
-//   [label or hash]
-//   hits=<line:col>,<line:col>,...
-
-struct CoverageData
-{
-    std::set<uint32_t> hits{};
-};
-
-// Global accumulator — survives across HookContext lifetimes
-inline std::map<ripple::uint256, CoverageData>&
-coverageMap()
-{
-    static std::map<ripple::uint256, CoverageData> map;
-    return map;
-}
-
-// Hash → label mapping (e.g. hash → "file:tipbot/tip.c")
-inline std::map<ripple::uint256, std::string>&
-coverageLabels()
-{
-    static std::map<ripple::uint256, std::string> labels;
-    return labels;
-}
-
-inline void
-coverageReset()
-{
-    coverageMap().clear();
-    coverageLabels().clear();
-}
-
-inline void
-coverageLabel(ripple::uint256 const& hookHash, std::string const& label)
-{
-    coverageLabels()[hookHash] = label;
-}
-
-inline std::set<uint32_t> const*
-coverageHits(ripple::uint256 const& hookHash)
-{
-    auto& map = coverageMap();
-    auto it = map.find(hookHash);
-    if (it == map.end())
-        return nullptr;
-    return &it->second.hits;
-}
-
-inline bool
-coverageDump(std::string const& path)
-{
-    auto& map = coverageMap();
-    if (map.empty())
-        return false;
-
-    auto& labels = coverageLabels();
-
-    std::ofstream out(path);
-    if (!out)
-        return false;
-
-    for (auto const& [hash, data] : map)
-    {
-        auto it = labels.find(hash);
-        if (it != labels.end())
-            out << "[" << it->second << "]\n";
-        else
-            out << "[" << to_string(hash) << "]\n";
-
-        out << "hits=";
-        bool first = true;
-        for (auto key : data.hits)
-        {
-            if (!first)
-                out << ",";
-            out << (key >> 16) << ":" << (key & 0xFFFF);
-            first = false;
-        }
-        out << "\n\n";
-    }
-
-    return true;
-}
-
-// --- Coverage host callback ---
-
-inline WasmEdge_Result
-onSourceLine(
-    void* data_ptr,
-    const WasmEdge_CallingFrameContext* frameCtx,
-    const WasmEdge_Value* in,
-    WasmEdge_Value* out)
-{
-    // Called by hookz-instrumented WASM at each DWARF source location.
-    // in[0] = line number, in[1] = column number.
-    (void)out;
-    (void)frameCtx;
-    auto* hookCtx = reinterpret_cast<HookContext*>(data_ptr);
-    if (!hookCtx)
-        return WasmEdge_Result_Success;
-
-    uint32_t line = WasmEdge_ValueGetI32(in[0]);
-    uint32_t col = WasmEdge_ValueGetI32(in[1]);
-
-    // Pack (line, col) into a single uint32_t key.
-    // Limits: line < 65536, col < 65536 — more than sufficient for hooks.
-    uint32_t key = (line << 16) | (col & 0xFFFF);
-    coverageMap()[hookCtx->result.hookHash].hits.insert(key);
-
-    return WasmEdge_Result_Success;
-}
-
 /**
  * HookExecutor is effectively a two-part function:
  * The first part sets up the Hook Api inside the wasm import, ready for use
@@ -609,22 +483,6 @@ public:
 #undef HOOK_WRAP_PARAMS
 #pragma pop_macro("HOOK_API_DEFINITION")
 
-        // Coverage callback: void __on_source_line(i32 line, i32 col)
-        // Registered unconditionally — production hooks don't import it,
-        // so it's harmless. Instrumented hooks call it at each DWARF
-        // source location to record line:col coverage hits.
-        {
-            static WasmEdge_ValType paramsOSL[] = {
-                WasmEdge_ValType_I32, WasmEdge_ValType_I32};
-            static auto* ftOSL =
-                WasmEdge_FunctionTypeCreate(paramsOSL, 2, nullptr, 0);
-            auto* hfOSL = WasmEdge_FunctionInstanceCreate(
-                ftOSL, hook::onSourceLine, (void*)(&ctx), 0);
-            static auto nameOSL =
-                WasmEdge_StringCreateByCString("__on_source_line");
-            WasmEdge_ModuleInstanceAddFunction(importObj, nameOSL, hfOSL);
-        }
-
         WasmEdge_TableInstanceContext* hostTable =
             WasmEdge_TableInstanceCreate(tableType);
         WasmEdge_ModuleInstanceAddTable(importObj, tableName, hostTable);

src/xrpld/app/hook/detail/applyHook.cpp

@@ -1111,7 +1111,7 @@ DEFINE_HOOK_FUNCTION(
     if (NOT_IN_BOUNDS(read_ptr, read_len, memory_length))
         return OUT_OF_BOUNDS;
 
-    if (!jh.trace())
+    if (!j.trace())
         return 0ULL;
 
     if (read_len > 128)
@@ -1125,16 +1125,16 @@ DEFINE_HOOK_FUNCTION(
 
         if (read_len > 0)
         {
-            JLOG(jh.trace())
-                << "HookTrace[" << HC_ACC() << "]: "
-                << std::string_view((const char*)memory + read_ptr, read_len)
-                << ": " << number;
+            j.trace() << "HookTrace[" << HC_ACC() << "]: "
+                      << std::string_view(
+                             (const char*)memory + read_ptr, read_len)
+                      << ": " << number;
 
             return 0ULL;
         }
     }
 
-    JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]: " << number;
+    j.trace() << "HookTrace[" << HC_ACC() << "]: " << number;
     return 0ULL;
     HOOK_TEARDOWN();
 }
@@ -1154,7 +1154,7 @@ DEFINE_HOOK_FUNCTION(
         NOT_IN_BOUNDS(dread_ptr, dread_len, memory_length))
         return OUT_OF_BOUNDS;
 
-    if (!jh.trace())
+    if (!j.trace())
         return 0ULL;
 
     if (mread_len > 128)
@@ -1214,9 +1214,8 @@ DEFINE_HOOK_FUNCTION(
 
     if (out_len > 0)
     {
-        JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]: "
-                         << std::string_view(
-                                (const char*)output_storage, out_len);
+        j.trace() << "HookTrace[" << HC_ACC() << "]: "
+                  << std::string_view((const char*)output_storage, out_len);
     }
 
     return 0ULL;
@@ -3404,7 +3403,7 @@ DEFINE_HOOK_FUNCTION(
     if (NOT_IN_BOUNDS(read_ptr, read_len, memory_length))
         return OUT_OF_BOUNDS;
 
-    if (!jh.trace())
+    if (!j.trace())
         return 0ULL;
 
     if (read_len > 128)
@@ -3421,8 +3420,8 @@ DEFINE_HOOK_FUNCTION(
 
     if (float1 == 0)
     {
-        JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]: " << messageKey
-                         << ": Float 0*10^(0) <ZERO>";
+        j.trace() << "HookTrace[" << HC_ACC() << "]: " << messageKey
+                  << ": Float 0*10^(0) <ZERO>";
         return 0ULL;
     }
 
@@ -3433,14 +3432,14 @@ DEFINE_HOOK_FUNCTION(
         man.value() > maxMantissa || exp.value() < minExponent ||
         exp.value() > maxExponent)
     {
-        JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]: " << messageKey
-                         << ": Float <INVALID>";
+        j.trace() << "HookTrace[" << HC_ACC() << "]: " << messageKey
+                  << ": Float <INVALID>";
         return 0ULL;
     }
 
-    JLOG(jh.trace()) << "HookTrace[" << HC_ACC() << "]:" << messageKey
-                     << ": Float " << (neg ? "-" : "") << man.value() << "*10^("
-                     << exp.value() << ")";
+    j.trace() << "HookTrace[" << HC_ACC() << "]:" << messageKey << ": Float "
+              << (neg ? "-" : "") << man.value() << "*10^(" << exp.value()
+              << ")";
     return 0ULL;
 
     HOOK_TEARDOWN();

src/xrpld/app/misc/Manifest.h

@@ -25,6 +25,7 @@
 #include <xrpl/protocol/PublicKey.h>
 #include <xrpl/protocol/SecretKey.h>
 
+#include <cstdint>
 #include <optional>
 #include <shared_mutex>
 #include <string>
@@ -254,6 +255,26 @@ class DatabaseCon;
 /** Remembers manifests with the highest sequence number. */
 class ManifestCache
 {
+public:
+    /** An atomic read of one cached validator manifest.
+
+        The snapshot may represent a revocation. If a signing key is supplied,
+        it resolves only while that key is the current key for the manifest.
+    */
+    struct Snapshot
+    {
+        PublicKey masterKey;
+        std::optional<PublicKey> signingKey;
+        std::uint32_t sequence;
+        std::string serialized;
+
+        bool
+        revoked() const
+        {
+            return Manifest::revoked(sequence);
+        }
+    };
+
 private:
     beast::Journal j_;
     std::shared_mutex mutable mutex_;
@@ -266,6 +287,9 @@ private:
 
     std::atomic<std::uint32_t> seq_{0};
 
+    std::optional<ManifestDisposition>
+    checkKeyRolesUnlocked(Manifest const& m) const;
+
 public:
     explicit ManifestCache(
         beast::Journal j = beast::Journal(beast::Journal::getNullSink()))
@@ -330,6 +354,17 @@ public:
     std::optional<std::string>
     getManifest(PublicKey const& pk) const;
 
+    /** Return one internally consistent view of the current manifest.
+
+        @param pk A master key or its current ephemeral signing key.
+
+        Unlike getManifest(), revocations are returned. This is used by the
+        overlay when ordering a manifest immediately before a validation and
+        when repairing a peer that sent an authenticated naked validation.
+    */
+    std::optional<Snapshot>
+    getManifestSnapshot(PublicKey const& pk) const;
+
     /** Returns `true` if master key has been revoked in a manifest.
 
         @param pk Master public key
@@ -341,6 +376,22 @@ public:
     bool
     revoked(PublicKey const& pk) const;
 
+    /** Check whether a manifest's keys conflict with retained key roles.
+
+        This does not verify signatures, compare sequences, or mutate the
+        cache. Transport uses it before relaying an unlisted pair: an
+        ephemeral path must not endorse an association the authoritative
+        cache would reject.
+
+        @return A key-role disposition, or `std::nullopt` when admissible.
+
+        @par Thread Safety
+
+        May be called concurrently.
+    */
+    std::optional<ManifestDisposition>
+    checkKeyRoles(Manifest const& m) const;
+
     /** Add manifest to cache.
 
         @param m Manifest to add

src/xrpld/app/misc/detail/Manifest.cpp

@@ -354,6 +354,28 @@ ManifestCache::getManifest(PublicKey const& pk) const
     return std::nullopt;
 }
 
+std::optional<ManifestCache::Snapshot>
+ManifestCache::getManifestSnapshot(PublicKey const& pk) const
+{
+    std::shared_lock lock{mutex_};
+
+    auto masterKey = pk;
+    if (auto const signing = signingToMasterKeys_.find(pk);
+        signing != signingToMasterKeys_.end())
+        masterKey = signing->second;
+
+    auto const manifest = map_.find(masterKey);
+    if (manifest == map_.end())
+        return std::nullopt;
+
+    auto const& current = manifest->second;
+    return Snapshot{
+        current.masterKey,
+        current.signingKey,
+        current.sequence,
+        current.serialized};
+}
+
 bool
 ManifestCache::revoked(PublicKey const& pk) const
 {
@@ -366,6 +388,56 @@ ManifestCache::revoked(PublicKey const& pk) const
     return false;
 }
 
+std::optional<ManifestDisposition>
+ManifestCache::checkKeyRolesUnlocked(Manifest const& m) const
+{
+    if (auto const x = signingToMasterKeys_.find(m.masterKey);
+        x != signingToMasterKeys_.end())
+    {
+        JLOG(j_.warn()) << to_string(m)
+                        << ": Master key already used as ephemeral key for "
+                        << toBase58(TokenType::NodePublic, x->second);
+        return ManifestDisposition::badMasterKey;
+    }
+
+    if (m.revoked())
+        return std::nullopt;
+
+    if (!m.signingKey)
+    {
+        JLOG(j_.warn()) << to_string(m)
+                        << ": is not revoked and the manifest has no signing "
+                           "key. Hence, the manifest is invalid";
+        return ManifestDisposition::invalid;
+    }
+
+    if (auto const x = signingToMasterKeys_.find(*m.signingKey);
+        x != signingToMasterKeys_.end())
+    {
+        JLOG(j_.warn()) << to_string(m)
+                        << ": Ephemeral key already used as ephemeral key for "
+                        << toBase58(TokenType::NodePublic, x->second);
+        return ManifestDisposition::badEphemeralKey;
+    }
+
+    if (auto const x = map_.find(*m.signingKey); x != map_.end())
+    {
+        JLOG(j_.warn()) << to_string(m)
+                        << ": Ephemeral key used as master key for "
+                        << to_string(x->second);
+        return ManifestDisposition::badEphemeralKey;
+    }
+
+    return std::nullopt;
+}
+
+std::optional<ManifestDisposition>
+ManifestCache::checkKeyRoles(Manifest const& m) const
+{
+    std::shared_lock lock{mutex_};
+    return checkKeyRolesUnlocked(m);
+}
+
 ManifestDisposition
 ManifestCache::applyManifest(Manifest m)
 {
@@ -418,51 +490,8 @@ ManifestCache::applyManifest(Manifest m)
         if (auto stream = j_.warn(); stream && revoked)
             LOG_MANIFEST_ACTION(stream, "Revoked", m.masterKey, m.sequence);
 
-        // Sanity check: the master key of this manifest should not be used as
-        // the ephemeral key of another manifest:
-        if (auto const x = signingToMasterKeys_.find(m.masterKey);
-            x != signingToMasterKeys_.end())
-        {
-            JLOG(j_.warn()) << to_string(m)
-                            << ": Master key already used as ephemeral key for "
-                            << toBase58(TokenType::NodePublic, x->second);
-
-            return ManifestDisposition::badMasterKey;
-        }
-
-        if (!revoked)
-        {
-            if (!m.signingKey)
-            {
-                JLOG(j_.warn()) << to_string(m)
-                                << ": is not revoked and the manifest has no "
-                                   "signing key. Hence, the manifest is "
-                                   "invalid";
-                return ManifestDisposition::invalid;
-            }
-
-            // Sanity check: the ephemeral key of this manifest should not be
-            // used as the master or ephemeral key of another manifest:
-            if (auto const x = signingToMasterKeys_.find(*m.signingKey);
-                x != signingToMasterKeys_.end())
-            {
-                JLOG(j_.warn())
-                    << to_string(m)
-                    << ": Ephemeral key already used as ephemeral key for "
-                    << toBase58(TokenType::NodePublic, x->second);
-
-                return ManifestDisposition::badEphemeralKey;
-            }
-
-            if (auto const x = map_.find(*m.signingKey); x != map_.end())
-            {
-                JLOG(j_.warn())
-                    << to_string(m) << ": Ephemeral key used as master key for "
-                    << to_string(x->second);
-
-                return ManifestDisposition::badEphemeralKey;
-            }
-        }
+        if (auto const disposition = checkKeyRolesUnlocked(m))
+            return *disposition;
 
         return std::nullopt;
     };
@@ -524,7 +553,6 @@ ManifestCache::applyManifest(Manifest m)
         signingToMasterKeys_.emplace(*m.signingKey, m.masterKey);
 
     iter->second = std::move(m);
-
     // Something has changed. Keep track of it.
     seq_++;
 
@@ -584,8 +612,13 @@ ManifestCache::load(
 
         auto mo = deserializeManifest(base64_decode(revocationStr));
 
-        if (!mo || !mo->revoked() ||
-            applyManifest(std::move(*mo)) == ManifestDisposition::invalid)
+        if (!mo || !mo->revoked())
+        {
+            JLOG(j_.error()) << "Invalid validator key revocation in config";
+            return false;
+        }
+
+        if (applyManifest(std::move(*mo)) == ManifestDisposition::invalid)
         {
             JLOG(j_.error()) << "Invalid validator key revocation in config";
             return false;

src/xrpld/app/tx/detail/SetHook.cpp

@@ -546,7 +546,7 @@ SetHook::validateHookSetEntry(SetHookCtx& ctx, STObject const& hookSetObj)
                 }
 
                 auto result = validateGuards(
-                    hook,
+                    hook,  // wasm to verify
                     logger,
                     hsacc,
                     hook_api::getImportWhitelist(ctx.rules),

src/xrpld/net/RPCSub.h

@@ -22,7 +22,6 @@
 
 #include <xrpld/core/JobQueue.h>
 #include <xrpld/net/InfoSub.h>
-#include <boost/asio/io_service.hpp>
 
 namespace ripple {
 
@@ -39,16 +38,17 @@ protected:
     explicit RPCSub(InfoSub::Source& source);
 };
 
-// VFALCO Why is the io_service needed?
 std::shared_ptr<RPCSub>
 make_RPCSub(
     InfoSub::Source& source,
-    boost::asio::io_service& io_service,
     JobQueue& jobQueue,
     std::string const& strUrl,
     std::string const& strUsername,
     std::string const& strPassword,
-    Logs& logs);
+    Logs& logs,
+    // Max events buffered before new ones are dropped. Configurable so
+    // tests can exercise the drop path without queueing the full default.
+    std::size_t maxQueueSize = 16384);
 
 }  // namespace ripple
 

src/xrpld/net/detail/HTTPClient.cpp

@@ -122,12 +122,20 @@ public:
         mComplete = complete;
         mTimeout = timeout;
 
+        // Bind a non-owning `this` (not shared_from_this()) into mBuild.
+        // mBuild is a member, so capturing a shared_ptr to self here would
+        // form a reference cycle (this -> mBuild -> shared_ptr<this>) that
+        // never breaks, leaking the object and its socket FD after the
+        // request completes. mBuild is only ever invoked from
+        // handleRequest(), which always runs inside an async handler that
+        // already holds a shared_from_this(), so the object is guaranteed
+        // alive whenever mBuild fires — a raw `this` is safe.
         request(
             bSSL,
             deqSites,
             std::bind(
                 &HTTPClientImp::makeGet,
-                shared_from_this(),
+                this,
                 strPath,
                 std::placeholders::_1,
                 std::placeholders::_2),
@@ -393,8 +401,12 @@ public:
         if (boost::regex_match(strHeader, smMatch, reBody))  // we got some body
             mBody = smMatch[1];
 
+        bool const hasContentLength =
+            boost::regex_match(strHeader, smMatch, reSize);
+        mReceivedContentLength = hasContentLength;
+
         std::size_t const responseSize = [&] {
-            if (boost::regex_match(strHeader, smMatch, reSize))
+            if (hasContentLength)
                 return beast::lexicalCast<std::size_t>(
                     std::string(smMatch[1]), maxResponseSize_);
             return maxResponseSize_;
@@ -445,22 +457,24 @@ public:
             JLOG(j_.trace()) << "Read error: " << mShutdown.message();
 
             invokeComplete(mShutdown);
+            return;
         }
-        else
-        {
-            if (mShutdown)
-            {
-                JLOG(j_.trace()) << "Complete.";
-            }
-            else
-            {
-                mResponse.commit(bytes_transferred);
-                std::string strBody{
-                    {std::istreambuf_iterator<char>(&mResponse)},
-                    std::istreambuf_iterator<char>()};
-                invokeComplete(ecResult, mStatus, mBody + strBody);
-            }
-        }
+
+        // Either the read completed normally or it ended at EOF. EOF is a
+        // successful completion for EOF-delimited responses, but it is an
+        // error when the server promised a Content-Length and closed early.
+        JLOG(j_.trace()) << "Complete.";
+
+        mResponse.commit(bytes_transferred);
+        std::string strBody{
+            {std::istreambuf_iterator<char>(&mResponse)},
+            std::istreambuf_iterator<char>()};
+
+        auto completeEc = ecResult;
+        if (completeEc == boost::asio::error::eof && !mReceivedContentLength)
+            completeEc.clear();
+
+        invokeComplete(completeEc, mStatus, mBody + strBody);
     }
 
     // Call cancel the deadline timer and invoke the completion routine.
@@ -516,6 +530,7 @@ private:
     boost::asio::streambuf mHeader;
     boost::asio::streambuf mResponse;
     std::string mBody;
+    bool mReceivedContentLength = false;
     const unsigned short mPort;
     std::size_t const maxResponseSize_;
     int mStatus;

src/xrpld/net/detail/RPCCall.cpp

@@ -1585,6 +1585,10 @@ struct RPCCallImp
             // callbackFuncP.
 
             // Receive reply
+            if (ecResult)
+                Throw<std::runtime_error>(
+                    "RPC transport error: " + ecResult.message());
+
             if (strData.empty())
                 Throw<std::runtime_error>(
                     "no response from server. Please "
@@ -1748,6 +1752,7 @@ rpcClient(
             }
 
             {
+                //@@start blocking-request
                 boost::asio::io_service isService;
                 RPCCall::fromNetwork(
                     isService,
@@ -1771,6 +1776,7 @@ rpcClient(
                     headers);
                 isService.run();  // This blocks until there are no more
                                   // outstanding async calls.
+                //@@end blocking-request
             }
             if (jvOutput.isMember("result"))
             {
@@ -1881,15 +1887,21 @@ fromNetwork(
 
     // Send request
 
-    // Number of bytes to try to receive if no
-    // Content-Length header received
-    constexpr auto RPC_REPLY_MAX_BYTES = megabytes(256);
+    // Number of bytes to try to receive if no Content-Length header is
+    // received. Webhook event deliveries ("event") ignore the response
+    // body, so a missing Content-Length must not pre-allocate the full
+    // 256MB RPC reply budget per in-flight delivery (maxInFlight can be
+    // 32 -> 8GB). Cap those small; genuine RPC replies (CLI) keep the
+    // large budget.
+    auto const RPC_REPLY_MAX_BYTES =
+        (strMethod == "event") ? megabytes(1) : megabytes(256);
 
     using namespace std::chrono_literals;
     // auto constexpr RPC_NOTIFY = 10min; // Wietse: lolwut 10 minutes for one
     // HTTP call?
     auto constexpr RPC_NOTIFY = 30s;
 
+    //@@start async-request
     HTTPClient::request(
         bSSL,
         io_service,
@@ -1914,6 +1926,7 @@ fromNetwork(
             std::placeholders::_3,
             j),
         j);
+    //@@end async-request
 }
 
 }  // namespace RPCCall

src/xrpld/net/detail/RPCSub.cpp

@@ -24,29 +24,30 @@
 #include <xrpl/basics/contract.h>
 #include <xrpl/json/to_string.h>
 #include <deque>
+#include <memory>
 
 namespace ripple {
 
 // Subscription object for JSON-RPC
-class RPCSubImp : public RPCSub
+class RPCSubImp : public RPCSub, public std::enable_shared_from_this<RPCSubImp>
 {
 public:
     RPCSubImp(
         InfoSub::Source& source,
-        boost::asio::io_service& io_service,
         JobQueue& jobQueue,
         std::string const& strUrl,
         std::string const& strUsername,
         std::string const& strPassword,
-        Logs& logs)
+        Logs& logs,
+        std::size_t maxQueueSize)
         : RPCSub(source)
-        , m_io_service(io_service)
         , m_jobQueue(jobQueue)
         , mUrl(strUrl)
         , mSSL(false)
         , mUsername(strUsername)
         , mPassword(strPassword)
         , mSending(false)
+        , maxQueueSize_(maxQueueSize)
         , j_(logs.journal("RPCSub"))
         , logs_(logs)
     {
@@ -78,14 +79,26 @@ public:
     {
         std::lock_guard sl(mLock);
 
-        // Wietse: we're not going to limit this, this is admin-port only, scale
-        // accordingly Dropping events just like this results in inconsistent
-        // data on the receiving end if (mDeque.size() >= eventQueueMax)
-        // {
-        //     // Drop the previous event.
-        //     JLOG(j_.warn()) << "RPCCall::fromNetwork drop";
-        //     mDeque.pop_back();
-        // }
+        if (mDeque.size() >= maxQueueSize_)
+        {
+            // Always advance mSeq so consumers can detect the gap, but
+            // rate-limit the log: a hopelessly behind endpoint drops on
+            // every send() and would otherwise flood the log. Warn on
+            // the first drop of a run and then once per dropLogInterval.
+            if (mDropped++ % dropLogInterval == 0)
+            {
+                JLOG(j_.warn())
+                    << "RPCCall::fromNetwork drop: queue full ("
+                    << mDeque.size() << "), seq=" << mSeq
+                    << ", endpoint=" << mIp << ", dropped=" << mDropped;
+            }
+            ++mSeq;
+            return;
+        }
+
+        // Endpoint caught up enough to accept again; reset so the next
+        // overflow burst logs its first drop immediately.
+        mDropped = 0;
 
         auto jm = broadcast ? j_.debug() : j_.info();
         JLOG(jm) << "RPCCall::fromNetwork push: " << jvObj;
@@ -97,10 +110,7 @@ public:
             // Start a sending thread.
             JLOG(j_.info()) << "RPCCall::fromNetwork start";
 
-            mSending = m_jobQueue.addJob(
-                jtCLIENT_SUBSCRIBE, "RPCSub::sendThread", [this]() {
-                    sendThread();
-                });
+            startSendingJob();
         }
     }
 
@@ -121,48 +131,66 @@ public:
     }
 
 private:
-    // XXX Could probably create a bunch of send jobs in a single get of the
-    // lock.
+    // Maximum concurrent HTTP deliveries per batch. Bounds file
+    // descriptor usage while still allowing parallel delivery to
+    // capable endpoints. With a 1024 FD process limit shared across
+    // peers, clients, and the node store, 32 per subscriber is a
+    // meaningful but survivable chunk even with multiple subscribers.
+    static constexpr int maxInFlight = 32;
+
+    // Log one drop warning per this many drops while the queue stays
+    // full, to avoid flooding the log on a persistently behind endpoint.
+    static constexpr std::size_t dropLogInterval = 1000;
+
+    // Schedule a sending job. Must be called under mLock. The job holds a
+    // weak_ptr and re-locks it on entry, so the RPCSub is kept alive for
+    // the duration of the batch even if it is unsubscribed (and would
+    // otherwise be destroyed) concurrently — sendThread dereferences this
+    // only via that strong ref. mDeque events are delivered until the sub
+    // is gone, after which weak.lock() fails and the job is a no-op.
+    void
+    startSendingJob()
+    {
+        std::weak_ptr<RPCSubImp> weak = weak_from_this();
+        mSending = m_jobQueue.addJob(
+            jtCLIENT_SUBSCRIBE, "RPCSub::sendThread", [weak]() {
+                if (auto self = weak.lock())
+                    self->sendThread();
+            });
+    }
+
     void
     sendThread()
     {
-        Json::Value jvEvent;
-        bool bSend;
+        // Process exactly ONE batch per job, then re-queue if more events
+        // remain, rather than draining the whole backlog in a single job.
+        // A local io_service's .run() blocks this worker thread for the
+        // batch (up to the per-request timeout), so re-queueing between
+        // batches keeps one slow/hung subscriber from monopolising a
+        // job-queue worker and starving consensus/ledger/RPC work.
+        //
+        // mSending must be cleared under the lock on every non-requeue
+        // exit path; if it ever stays set without a job in flight, send()
+        // sees mSending == true and never restarts us, stalling the queue
+        // forever — the original bug (xrpld issue #6341).
+        boost::asio::io_service io_service;
+        int dispatched = 0;
 
-        do
+        try
         {
             {
-                // Obtain the lock to manipulate the queue and change sending.
                 std::lock_guard sl(mLock);
 
-                if (mDeque.empty())
-                {
-                    mSending = false;
-                    bSend = false;
-                }
-                else
+                while (!mDeque.empty() && dispatched < maxInFlight)
                 {
                     auto const [seq, env] = mDeque.front();
-
                     mDeque.pop_front();
 
-                    jvEvent = env;
+                    Json::Value jvEvent = env;
                     jvEvent["seq"] = seq;
 
-                    bSend = true;
-                }
-            }
-
-            // Send outside of the lock.
-            if (bSend)
-            {
-                // XXX Might not need this in a try.
-                try
-                {
-                    JLOG(j_.info()) << "RPCCall::fromNetwork: " << mIp;
-
                     RPCCall::fromNetwork(
-                        m_io_service,
+                        io_service,
                         mIp,
                         mPort,
                         mUsername,
@@ -173,21 +201,51 @@ private:
                         mSSL,
                         true,
                         logs_);
-                }
-                catch (const std::exception& e)
-                {
-                    JLOG(j_.info())
-                        << "RPCCall::fromNetwork exception: " << e.what();
+                    ++dispatched;
                 }
             }
-        } while (bSend);
+
+            // dispatched is always > 0 here (send() only starts a job
+            // after enqueuing, and the re-queue below only fires with a
+            // non-empty deque), but guard anyway so an empty batch can't
+            // log/spin — it falls straight through to clear mSending.
+            if (dispatched > 0)
+            {
+                JLOG(j_.info()) << "RPCCall::fromNetwork: " << mIp
+                                << " dispatching " << dispatched << " events";
+
+                io_service.run();
+            }
+        }
+        catch (std::exception const& e)
+        {
+            // Bail rather than re-queue: a persistently failing endpoint
+            // would otherwise spin the job queue. mSending is reset so the
+            // next send() restarts delivery.
+            JLOG(j_.warn()) << "RPCSub::sendThread exception: " << e.what();
+            std::lock_guard sl(mLock);
+            mSending = false;
+            return;
+        }
+        catch (...)
+        {
+            JLOG(j_.warn()) << "RPCSub::sendThread unknown exception";
+            std::lock_guard sl(mLock);
+            mSending = false;
+            return;
+        }
+
+        // Batch complete: re-queue for the next one (mSending stays set)
+        // or clear mSending if the queue drained — both under the lock to
+        // avoid a lost-wakeup race with send().
+        std::lock_guard sl(mLock);
+        if (mDeque.empty())
+            mSending = false;
+        else
+            startSendingJob();
     }
 
 private:
-    // Wietse: we're not going to limit this, this is admin-port only, scale
-    // accordingly enum { eventQueueMax = 32 };
-
-    boost::asio::io_service& m_io_service;
     JobQueue& m_jobQueue;
 
     std::string mUrl;
@@ -200,8 +258,15 @@ private:
 
     int mSeq;  // Next id to allocate.
 
+    std::size_t mDropped = 0;  // Consecutive drops while queue is full.
+
     bool mSending;  // Sending threead is active.
 
+    // Maximum queued events before dropping. The default (16384) is a
+    // ~10-minute buffer at 100+ events/ledger; a hopelessly behind
+    // endpoint trips it and consumers detect the gap via the seq field.
+    std::size_t const maxQueueSize_;
+
     std::deque<std::pair<int, Json::Value>> mDeque;
 
     beast::Journal const j_;
@@ -217,21 +282,21 @@ RPCSub::RPCSub(InfoSub::Source& source) : InfoSub(source, Consumer())
 std::shared_ptr<RPCSub>
 make_RPCSub(
     InfoSub::Source& source,
-    boost::asio::io_service& io_service,
     JobQueue& jobQueue,
     std::string const& strUrl,
     std::string const& strUsername,
     std::string const& strPassword,
-    Logs& logs)
+    Logs& logs,
+    std::size_t maxQueueSize)
 {
     return std::make_shared<RPCSubImp>(
         std::ref(source),
-        std::ref(io_service),
         std::ref(jobQueue),
         strUrl,
         strUsername,
         strPassword,
-        logs);
+        logs,
+        maxQueueSize);
 }
 
 }  // namespace ripple

src/xrpld/overlay/Overlay.h

@@ -145,9 +145,12 @@ public:
     virtual void
     broadcast(protocol::TMProposeSet& m) = 0;
 
-    /** Broadcast a validation. */
+    /** Broadcast a validation.
+     * @param m the serialized validation
+     * @param validator The pubkey that signed the validation
+     */
     virtual void
-    broadcast(protocol::TMValidation& m) = 0;
+    broadcast(protocol::TMValidation& m, PublicKey const& validator) = 0;
 
     /** Relay a proposal.
      * @param m the serialized proposal
@@ -171,7 +174,8 @@ public:
     relay(
         protocol::TMValidation& m,
         uint256 const& uid,
-        PublicKey const& validator) = 0;
+        PublicKey const& validator,
+        std::shared_ptr<protocol::TMManifests const> const& prerequisite) = 0;
 
     /** Relay a transaction. If the tx reduce-relay feature is enabled then
      * randomly select peers to relay to and queue transaction's hash

src/xrpld/overlay/detail/OverlayImpl.cpp

@@ -631,12 +631,14 @@ OverlayImpl::onPeerDeactivate(Peer::id_t id)
 void
 OverlayImpl::onManifests(
     std::shared_ptr<protocol::TMManifests> const& m,
-    std::shared_ptr<PeerImp> const& from)
+    std::shared_ptr<PeerImp> const& from,
+    ManifestAdmission admission)
 {
     auto const n = m->list_size();
     auto const& journal = from->pjournal();
 
     protocol::TMManifests relay;
+    std::vector<std::pair<PublicKey, std::uint32_t>> relayAssertions;
 
     for (std::size_t i = 0; i < n; ++i)
     {
@@ -645,13 +647,72 @@ OverlayImpl::onManifests(
         if (auto mo = deserializeManifest(s))
         {
             auto const serialized = mo->serialized;
+            auto const masterKey = mo->masterKey;
+            auto const sequence = mo->sequence;
+            auto const revoked = mo->revoked();
+
+            // Cache membership is observation, not authority: a legacy row
+            // cannot perpetuate itself by presenting a newer normal
+            // signature. Current local policy admits ordinary updates; the
+            // one response-gated exception can only terminate an existing
+            // retained master.
+            auto const listed = app_.validators().listed(masterKey);
+            auto const retainedRevocationResponse = !listed && revoked &&
+                admission == ManifestAdmission::retainedRevocationResponse &&
+                [&]() {
+                    auto const current =
+                        app_.validatorManifests().getManifestSnapshot(
+                            masterKey);
+                    return current && current->sequence < sequence;
+                }();
+            if (!listed && !retainedRevocationResponse)
+            {
+                if (n == 1)
+                {
+                    JLOG(journal.debug())
+                        << "manifest_validation single_manifest_ignored master="
+                        << toBase58(TokenType::NodePublic, masterKey)
+                        << " sequence=" << sequence
+                        << " reason=unlisted_new_identity";
+                }
+                continue;
+            }
 
             auto const result =
                 app_.validatorManifests().applyManifest(std::move(*mo));
 
+            if (result == ManifestDisposition::invalid)
+                from->charge(
+                    Resource::feeInvalidSignature,
+                    "invalid validator manifest signature");
+            else if (
+                result == ManifestDisposition::badMasterKey ||
+                result == ManifestDisposition::badEphemeralKey)
+                from->charge(
+                    Resource::feeInvalidData,
+                    "invalid validator manifest key role");
+
+            if (n == 1)
+            {
+                JLOG(journal.debug())
+                    << "manifest_validation single_manifest_processed master="
+                    << toBase58(TokenType::NodePublic, masterKey)
+                    << " sequence=" << sequence
+                    << " disposition=" << to_string(result);
+            }
+
             if (result == ManifestDisposition::accepted)
             {
-                relay.add_list()->set_stobject(s);
+                if (revoked)
+                {
+                    // A revocation has no associated validation to carry it
+                    // onward, so it retains immediate network-wide relay.
+                    relay.add_list()->set_stobject(s);
+                    relayAssertions.emplace_back(masterKey, sequence);
+                    JLOG(journal.debug())
+                        << "manifest_revocation accepted_for_relay master="
+                        << toBase58(TokenType::NodePublic, masterKey);
+                }
 
                 // N.B.: this is important; the applyManifest call above moves
                 //       the loaded Manifest out of the optional so we need to
@@ -680,8 +741,11 @@ OverlayImpl::onManifests(
     }
 
     if (!relay.list().empty())
-        for_each([m2 = std::make_shared<Message>(relay, protocol::mtMANIFESTS)](
-                     std::shared_ptr<PeerImp>&& p) { p->send(m2); });
+        for_each([m2 = std::make_shared<Message>(relay, protocol::mtMANIFESTS),
+                  assertions = std::move(relayAssertions)](
+                     std::shared_ptr<PeerImp>&& p) {
+            p->sendManifestAssertions(m2, assertions);
+        });
 }
 
 void
@@ -1155,17 +1219,21 @@ OverlayImpl::relay(
 }
 
 void
-OverlayImpl::broadcast(protocol::TMValidation& m)
+OverlayImpl::broadcast(protocol::TMValidation& m, PublicKey const& validator)
 {
-    auto const sm = std::make_shared<Message>(m, protocol::mtVALIDATION);
-    for_each([sm](std::shared_ptr<PeerImp>&& p) { p->send(sm); });
+    auto const sm =
+        std::make_shared<Message>(m, protocol::mtVALIDATION, validator);
+    for_each([sm, validator](std::shared_ptr<PeerImp>&& p) {
+        p->sendValidation(sm, validator);
+    });
 }
 
 std::set<Peer::id_t>
 OverlayImpl::relay(
     protocol::TMValidation& m,
     uint256 const& uid,
-    PublicKey const& validator)
+    PublicKey const& validator,
+    std::shared_ptr<protocol::TMManifests const> const& prerequisite)
 {
     if (auto const toSkip = app_.getHashRouter().shouldRelay(uid))
     {
@@ -1173,43 +1241,13 @@ OverlayImpl::relay(
             std::make_shared<Message>(m, protocol::mtVALIDATION, validator);
         for_each([&](std::shared_ptr<PeerImp>&& p) {
             if (toSkip->find(p->id()) == toSkip->end())
-                p->send(sm);
+                p->sendValidation(sm, validator, prerequisite);
         });
         return *toSkip;
     }
     return {};
 }
 
-std::shared_ptr<Message>
-OverlayImpl::getManifestsMessage()
-{
-    std::lock_guard g(manifestLock_);
-
-    if (auto seq = app_.validatorManifests().sequence();
-        seq != manifestListSeq_)
-    {
-        protocol::TMManifests tm;
-
-        app_.validatorManifests().for_each_manifest(
-            [&tm](std::size_t s) { tm.mutable_list()->Reserve(s); },
-            [&tm, &hr = app_.getHashRouter()](Manifest const& manifest) {
-                tm.add_list()->set_stobject(
-                    manifest.serialized.data(), manifest.serialized.size());
-                hr.addSuppression(manifest.hash());
-            });
-
-        manifestMessage_.reset();
-
-        if (tm.list_size() != 0)
-            manifestMessage_ =
-                std::make_shared<Message>(tm, protocol::mtMANIFESTS);
-
-        manifestListSeq_ = seq;
-    }
-
-    return manifestMessage_;
-}
-
 void
 OverlayImpl::relay(
     uint256 const& hash,

src/xrpld/overlay/detail/OverlayImpl.h

@@ -124,13 +124,6 @@ private:
     // Transaction reduce-relay metrics
     metrics::TxMetrics txMetrics_;
 
-    // A message with the list of manifests we send to peers
-    std::shared_ptr<Message> manifestMessage_;
-    // Used to track whether we need to update the cached list of manifests
-    std::optional<std::uint32_t> manifestListSeq_;
-    // Protects the message and the sequence list of manifests
-    std::mutex manifestLock_;
-
     //--------------------------------------------------------------------------
 
 public:
@@ -221,7 +214,7 @@ public:
     broadcast(protocol::TMProposeSet& m) override;
 
     void
-    broadcast(protocol::TMValidation& m) override;
+    broadcast(protocol::TMValidation& m, PublicKey const& validator) override;
 
     std::set<Peer::id_t>
     relay(
@@ -233,7 +226,9 @@ public:
     relay(
         protocol::TMValidation& m,
         uint256 const& uid,
-        PublicKey const& validator) override;
+        PublicKey const& validator,
+        std::shared_ptr<protocol::TMManifests const> const& prerequisite)
+        override;
 
     void
     relay(
@@ -241,9 +236,6 @@ public:
         std::optional<std::reference_wrapper<protocol::TMTransaction>> m,
         std::set<Peer::id_t> const& skip) override;
 
-    std::shared_ptr<Message>
-    getManifestsMessage();
-
     //--------------------------------------------------------------------------
     //
     // OverlayImpl
@@ -293,11 +285,14 @@ public:
         }
     }
 
+    enum class ManifestAdmission { localPolicy, retainedRevocationResponse };
+
     // Called when TMManifests is received from a peer
     void
     onManifests(
         std::shared_ptr<protocol::TMManifests> const& m,
-        std::shared_ptr<PeerImp> const& from);
+        std::shared_ptr<PeerImp> const& from,
+        ManifestAdmission admission = ManifestAdmission::localPolicy);
 
     static bool
     isPeerUpgrade(http_request_type const& request);

src/xrpld/overlay/detail/PeerImp.cpp

@@ -36,6 +36,7 @@
 #include <xrpl/basics/base64.h>
 #include <xrpl/basics/random.h>
 #include <xrpl/basics/safe_cast.h>
+#include <xrpl/basics/scope.h>
 #include <xrpl/beast/core/LexicalCast.h>
 #include <xrpl/protocol/digest.h>
 
@@ -47,6 +48,7 @@
 #include <mutex>
 #include <numeric>
 #include <sstream>
+#include <utility>
 
 using namespace std::chrono_literals;
 
@@ -58,6 +60,7 @@ std::chrono::milliseconds constexpr peerHighLatency{300};
 
 /** How often we PING the peer to check for latency and sendq probe */
 std::chrono::seconds constexpr peerTimerInterval{60};
+std::size_t constexpr maxManifestBytes = 4096;
 }  // namespace
 
 // TODO: Remove this exclusion once unit tests are added after the hotfix
@@ -291,6 +294,92 @@ PeerImp::send(std::shared_ptr<Message> const& m)
                 std::placeholders::_2)));
 }
 
+void
+PeerImp::sendValidation(
+    std::shared_ptr<Message> const& validation,
+    PublicKey const& signingKey,
+    std::shared_ptr<protocol::TMManifests const> const& prerequisite)
+{
+    if (!strand_.running_in_this_thread())
+        return post(
+            strand_,
+            std::bind(
+                &PeerImp::sendValidation,
+                shared_from_this(),
+                validation,
+                signingKey,
+                prerequisite));
+
+    if (gracefulClose_ || detaching_)
+        return;
+
+    // Admit the pair as one operation. Otherwise send(manifest) could succeed,
+    // send(validation) could be squelched, and the connection would observe a
+    // credential with no associated validation.
+    if (!squelch_.expireSquelch(signingKey))
+        return;
+
+    std::optional<PublicKey> prerequisiteMaster;
+    std::uint32_t prerequisiteSequence = 0;
+    std::shared_ptr<protocol::TMManifests const> manifest = prerequisite;
+    if (!manifest)
+    {
+        if (auto const snapshot =
+                app_.validatorManifests().getManifestSnapshot(signingKey);
+            snapshot && !snapshot->revoked() && snapshot->signingKey &&
+            *snapshot->signingKey == signingKey)
+        {
+            auto value = std::make_shared<protocol::TMManifests>();
+            value->add_list()->set_stobject(snapshot->serialized);
+            manifest = value;
+            prerequisiteMaster = snapshot->masterKey;
+            prerequisiteSequence = snapshot->sequence;
+        }
+    }
+    else if (manifest->list_size() == 1)
+    {
+        if (auto parsed = deserializeManifest(manifest->list(0).stobject()))
+        {
+            prerequisiteMaster = parsed->masterKey;
+            prerequisiteSequence = parsed->sequence;
+        }
+    }
+
+    // UNLs are local. This sender cannot infer that the receiver retains the
+    // same identities, and there is deliberately no association ACK or
+    // per-connection receiver table in this cut. Therefore every validation
+    // with an available prerequisite carries it immediately beforehand.
+    bool const sendPrerequisite = manifest && prerequisiteMaster;
+
+    if (sendPrerequisite)
+    {
+        JLOG(p_journal_.debug())
+            << "manifest_validation send_prerequisite peer=" << id_
+            << " master="
+            << (prerequisiteMaster
+                    ? toBase58(TokenType::NodePublic, *prerequisiteMaster)
+                    : "unknown")
+            << " sequence=" << prerequisiteSequence;
+        if (auto const retained = app_.validatorManifests().getManifestSnapshot(
+                *prerequisiteMaster);
+            retained && retained->sequence >= prerequisiteSequence)
+            recordManifestAssertion(*prerequisiteMaster, prerequisiteSequence);
+        send(std::make_shared<Message>(*manifest, protocol::mtMANIFESTS));
+    }
+
+    send(validation);
+    if (sendPrerequisite)
+    {
+        JLOG(p_journal_.debug())
+            << "manifest_validation pair_enqueued peer=" << id_
+            << " order=manifest,validation master="
+            << (prerequisiteMaster
+                    ? toBase58(TokenType::NodePublic, *prerequisiteMaster)
+                    : "unknown")
+            << " sequence=" << prerequisiteSequence;
+    }
+}
+
 void
 PeerImp::sendTxQueue()
 {
@@ -870,9 +959,6 @@ PeerImp::doProtocolStart()
             });
     }
 
-    if (auto m = overlay_.getManifestsMessage())
-        send(m);
-
     setTimer();
 }
 
@@ -1058,10 +1144,162 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMManifests> const& m)
     }
 
     if (s > 100)
+    {
         fee_.update(Resource::feeModerateBurdenPeer, "oversize");
+        return;
+    }
 
+    for (auto const& item : m->list())
+    {
+        if (item.stobject().size() > maxManifestBytes)
+        {
+            fee_.update(
+                Resource::feeModerateBurdenPeer, "oversized manifest object");
+            return;
+        }
+    }
+
+    auto const that = shared_from_this();
+
+    // A single normal manifest is a bounded, connection-scoped candidate for
+    // a later matching validation. Structural decoding is cheap enough
+    // to identify its signing key; signature verification waits until that
+    // validation claims the same key. Revocations have no following validation
+    // and therefore retain immediate verification/application semantics.
+    if (s == 1)
+    {
+        auto const& serialized = m->list(0).stobject();
+        auto manifest = deserializeManifest(serialized, p_journal_);
+        if (!manifest)
+        {
+            fee_.update(Resource::feeMalformedRequest, "malformed manifest");
+            return;
+        }
+
+        if (manifest->revoked())
+        {
+            // A fresh self-signed revocation has no more claim on permanent
+            // state than a fresh ordinary manifest. Current local policy may
+            // admit it directly; an unlisted response may only terminate a
+            // master already retained here after this connection asserted an
+            // older sequence.
+            auto const listed = app_.validators().listed(manifest->masterKey);
+            auto const current = app_.validatorManifests().getManifestSnapshot(
+                manifest->masterKey);
+            auto const retainedResponse = !listed && current &&
+                current->sequence < manifest->sequence &&
+                assertedOlderManifest(manifest->masterKey, manifest->sequence);
+            if (!listed && !retainedResponse)
+            {
+                JLOG(p_journal_.debug())
+                    << "manifest_revocation ignored_unlisted master="
+                    << toBase58(TokenType::NodePublic, manifest->masterKey);
+                return;
+            }
+
+            app_.getJobQueue().addJob(
+                jtMANIFEST,
+                "receiveManifestRevocation",
+                [this, that, m, retainedResponse]() {
+                    overlay_.onManifests(
+                        m,
+                        that,
+                        retainedResponse
+                            ? OverlayImpl::ManifestAdmission::
+                                  retainedRevocationResponse
+                            : OverlayImpl::ManifestAdmission::localPolicy);
+                });
+            return;
+        }
+
+        XRPL_ASSERT(
+            manifest->signingKey,
+            "ripple::PeerImp::onMessage(TMManifests) : normal manifest has "
+            "signing key");
+
+        if (publicKeyType(*manifest->signingKey) != KeyType::secp256k1)
+        {
+            fee_.update(
+                Resource::feeInvalidData,
+                "validator manifest signing key is not secp256k1");
+            return;
+        }
+
+        if (auto const current = app_.validatorManifests().getManifestSnapshot(
+                manifest->masterKey);
+            current && current->sequence >= manifest->sequence)
+        {
+            JLOG(p_journal_.debug())
+                << "manifest_validation candidate_ignored peer=" << id_
+                << " reason=global_sequence master="
+                << toBase58(TokenType::NodePublic, manifest->masterKey);
+            // A strictly lower sequence claims that the sender is behind for
+            // a master this node retains. Answer with the retained manifest —
+            // or the revocation, the freshest possible answer — without
+            // treating this unverified trigger as ingress authority. The
+            // shared repair ledger suppresses repeats while its row survives;
+            // equal-sequence always-send traffic draws nothing.
+            if (current->sequence > manifest->sequence)
+            {
+                JLOG(p_journal_.debug())
+                    << "manifest_validation stale_correction peer=" << id_
+                    << " master="
+                    << toBase58(TokenType::NodePublic, manifest->masterKey)
+                    << " theirs=" << manifest->sequence
+                    << " ours=" << current->sequence;
+                sendManifestRepair(
+                    current->masterKey, current->sequence, current->serialized);
+            }
+            return;
+        }
+
+        if (pendingManifest_)
+        {
+            auto const current = app_.validatorManifests().getManifestSnapshot(
+                pendingManifest_->masterKey);
+            if (current && current->sequence >= pendingManifest_->sequence)
+                pendingManifest_.reset();
+        }
+
+        if (pendingManifest_)
+        {
+            auto const pendingRelevant =
+                app_.validators().listed(pendingManifest_->masterKey);
+            auto const incomingRelevant =
+                app_.validators().listed(manifest->masterKey);
+            auto const sameMasterNewer =
+                pendingManifest_->masterKey == manifest->masterKey &&
+                manifest->sequence > pendingManifest_->sequence;
+            if (!sameMasterNewer && (!incomingRelevant || pendingRelevant))
+            {
+                JLOG(p_journal_.debug())
+                    << "manifest_validation candidate_ignored peer=" << id_
+                    << " reason=pending_candidate";
+                return;
+            }
+
+            JLOG(p_journal_.debug())
+                << "manifest_validation candidate_replaced peer=" << id_
+                << " reason="
+                << (sameMasterNewer ? "same_master_newer" : "listed_priority")
+                << " master="
+                << toBase58(TokenType::NodePublic, manifest->masterKey);
+        }
+        pendingManifest_.emplace(PendingManifest{
+            m, manifest->masterKey, *manifest->signingKey, manifest->sequence});
+        JLOG(p_journal_.debug())
+            << "manifest_validation candidate_staged peer=" << id_ << " master="
+            << toBase58(TokenType::NodePublic, manifest->masterKey)
+            << " signing="
+            << toBase58(TokenType::NodePublic, *manifest->signingKey)
+            << " sequence=" << manifest->sequence;
+        return;
+    }
+
+    // Compatibility lane for legacy handshake-era batches. New propagation
+    // sends exactly one manifest immediately before its validation.
     app_.getJobQueue().addJob(
-        jtMANIFEST, "receiveManifests", [this, that = shared_from_this(), m]() {
+        jtMANIFEST, "receiveManifests", [this, that, m]() {
             overlay_.onManifests(m, that);
         });
 }
@@ -2289,12 +2527,59 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMValidation> const& m)
     {
         auto const closeTime = app_.timeKeeper().closeTime();
 
+        if (pendingManifest_)
+        {
+            auto const current = app_.validatorManifests().getManifestSnapshot(
+                pendingManifest_->masterKey);
+            if (current && current->sequence >= pendingManifest_->sequence)
+            {
+                JLOG(p_journal_.debug())
+                    << "manifest_validation candidate_dropped peer=" << id_
+                    << " reason=retained_sequence master="
+                    << toBase58(
+                           TokenType::NodePublic, pendingManifest_->masterKey);
+                pendingManifest_.reset();
+            }
+        }
+
+        // Claim phase: copy the strand-owned waiting row so admission can be
+        // decided. Ownership has not moved and no job right exists yet.
+        std::optional<PendingManifest> manifestContext;
+        SerialIter claimIter(makeSlice(m->validation()));
+        STObject claim(claimIter, sfValidation);
+        auto const claimedKeyBytes = claim.getFieldVL(sfSigningPubKey);
+        if (publicKeyType(makeSlice(claimedKeyBytes)) == KeyType::secp256k1)
+        {
+            PublicKey const claimedKey(makeSlice(claimedKeyBytes));
+            if (pendingManifest_ &&
+                claimedKey == pendingManifest_->signingKey &&
+                !manifestVerificationInFlight_)
+            {
+                manifestContext = pendingManifest_;
+            }
+            else if (
+                pendingManifest_ && claimedKey == pendingManifest_->signingKey)
+            {
+                JLOG(p_journal_.debug())
+                    << "manifest_validation candidate_retained peer=" << id_
+                    << " reason=verification_in_flight";
+            }
+            else if (pendingManifest_)
+            {
+                JLOG(p_journal_.debug())
+                    << "manifest_validation candidate_retained peer=" << id_
+                    << " reason=signing_key_mismatch";
+            }
+        }
+
         std::shared_ptr<STValidation> val;
         {
             SerialIter sit(makeSlice(m->validation()));
             val = std::make_shared<STValidation>(
                 std::ref(sit),
-                [this](PublicKey const& pk) {
+                [this, &manifestContext](PublicKey const& pk) {
+                    if (manifestContext && pk == manifestContext->signingKey)
+                        return calcNodeID(manifestContext->masterKey);
                     return calcNodeID(
                         app_.validatorManifests().getMasterKey(pk));
                 },
@@ -2302,6 +2587,15 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMValidation> const& m)
             val->setSeen(closeTime);
         }
 
+        auto const& signingKey = val->getSignerPublic();
+        auto const masterKey = manifestContext
+            ? manifestContext->masterKey
+            : app_.validatorManifests().getMasterKey(signingKey);
+        JLOG(p_journal_.debug())
+            << "manifest_validation validation_parsed peer=" << id_
+            << " signing=" << toBase58(TokenType::NodePublic, signingKey)
+            << " master=" << toBase58(TokenType::NodePublic, masterKey);
+
         if (!isCurrent(
                 app_.getValidations().parms(),
                 app_.timeKeeper().closeTime(),
@@ -2316,26 +2610,85 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMValidation> const& m)
         // RH TODO: when isTrusted = false we should probably also cache a key
         // suppression for 30 seconds to avoid doing a relatively expensive
         // lookup every time a spam packet is received
+        // A pending manifest is still only a claim here. Do not let its
+        // claimed master key promote this work to the trusted queue. The job
+        // verifies both signatures before applying the manifest and making
+        // the final trust decision.
         auto const isTrusted =
             app_.validators().trusted(val->getSignerPublic());
+        auto const candidateListed = manifestContext &&
+            app_.validators().listed(manifestContext->masterKey);
+
+        // A naked validation with neither a trusted signer nor a retained
+        // signing-to-master mapping cannot contribute to consensus and cannot
+        // be repaired locally. Drop it before it claims the validation hash;
+        // a later manifest+validation pair can then be verified normally.
+        if (!manifestContext && !isTrusted && masterKey == signingKey &&
+            !app_.validators().listed(signingKey))
+        {
+            JLOG(p_journal_.debug())
+                << "manifest_validation naked_unknown_dropped peer=" << id_
+                << " signing=" << toBase58(TokenType::NodePublic, signingKey);
+            return;
+        }
 
         // 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 && app_.config().RELAY_UNTRUSTED_VALIDATIONS == -1)
+        if (!isTrusted && !candidateListed &&
+            app_.config().RELAY_UNTRUSTED_VALIDATIONS == -1)
             return;
 
-        auto key = sha512Half(makeSlice(m->validation()));
+        if (!isTrusted && !candidateListed &&
+            tracking_.load() == Tracking::diverged)
+        {
+            JLOG(p_journal_.debug())
+                << "Dropping untrusted validation from diverged peer";
+            return;
+        }
+
+        if (!isTrusted && !candidateListed &&
+            app_.getFeeTrack().isLoadedLocal())
+        {
+            JLOG(p_journal_.debug())
+                << "Dropping untrusted validation for load";
+            return;
+        }
+
+        if (manifestContext)
+        {
+            JLOG(p_journal_.debug())
+                << "manifest_validation candidate_matched peer=" << id_
+                << " master="
+                << toBase58(TokenType::NodePublic, manifestContext->masterKey)
+                << " signing=" << toBase58(TokenType::NodePublic, signingKey)
+                << " sequence=" << manifestContext->sequence;
+        }
+
+        auto const key = sha512Half(makeSlice(m->validation()));
+        auto const suppressionKey = manifestContext
+            ? sha512Half(
+                  std::uint32_t{0x4d565031},  // "MVP1"
+                  makeSlice(m->validation()),
+                  makeSlice(manifestContext->message->list(0).stobject()))
+            : key;
 
         if (auto [added, relayed] =
-                app_.getHashRouter().addSuppressionPeerWithStatus(key, id_);
+                app_.getHashRouter().addSuppressionPeerWithStatus(
+                    suppressionKey, id_);
             !added)
         {
+            // A previously relayed copy of these exact bytes has already
+            // passed validation. Repair this particular naked sender even
+            // though another peer won global admission for the hash.
+            if (!manifestContext && relayed)
+                sendManifestRepairForSigningKey(val->getSignerPublic());
+
             // Count unique messages (Slots has it's own 'HashRouter'), which a
             // peer receives within IDLED seconds since the message has been
             // relayed. Wait WAIT_ON_BOOTUP time to let the server establish
             // connections to peers.
-            if (reduceRelayReady() && relayed &&
+            if (!manifestContext && reduceRelayReady() && relayed &&
                 (stopwatch().now() - *relayed) < reduce_relay::IDLED)
                 overlay_.updateSlotAndSquelch(
                     key, val->getSignerPublic(), id_, protocol::mtVALIDATION);
@@ -2343,39 +2696,67 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMValidation> const& m)
             return;
         }
 
-        if (!isTrusted && (tracking_.load() == Tracking::diverged))
-        {
-            JLOG(p_journal_.debug())
-                << "Dropping untrusted validation from diverged peer";
-        }
-        else if (isTrusted || !app_.getFeeTrack().isLoadedLocal())
-        {
-            std::string const name = [isTrusted, val]() {
-                std::string ret =
-                    isTrusted ? "Trusted validation" : "Untrusted validation";
+        std::string const name = [isTrusted, val]() {
+            std::string ret =
+                isTrusted ? "Trusted validation" : "Untrusted validation";
 
 #ifdef DEBUG
-                ret += " " +
-                    std::to_string(val->getFieldU32(sfLedgerSequence)) + ": " +
-                    to_string(val->getNodeID());
+            ret += " " + std::to_string(val->getFieldU32(sfLedgerSequence)) +
+                ": " + to_string(val->getNodeID());
 #endif
 
-                return ret;
-            }();
+            return ret;
+        }();
 
-            std::weak_ptr<PeerImp> weak = shared_from_this();
-            app_.getJobQueue().addJob(
+        std::weak_ptr<PeerImp> weak = shared_from_this();
+        auto pairedPeer = manifestContext ? shared_from_this() : nullptr;
+        bool const pairedJob = manifestContext.has_value();
+        if (pairedJob)
+        {
+            XRPL_ASSERT(
+                pendingManifest_ &&
+                    pendingManifest_->message == manifestContext->message,
+                "ripple::PeerImp::onMessage(TMValidation) : pending manifest "
+                "claim is current");
+            // Admission commit: move authority out of the strand-owned
+            // waiting row and mint this connection's sole active job token.
+            pendingManifest_.reset();
+            manifestVerificationInFlight_ = true;
+            JLOG(p_journal_.debug())
+                << "manifest_validation candidate_claimed peer=" << id_
+                << " state=verification_in_flight";
+        }
+
+        bool queued = false;
+        try
+        {
+            queued = app_.getJobQueue().addJob(
                 isTrusted ? jtVALIDATION_t : jtVALIDATION_ut,
                 name,
-                [weak, val, m, key]() {
-                    if (auto peer = weak.lock())
-                        peer->checkValidation(val, key, m);
+                [weak,
+                 pairedPeer = std::move(pairedPeer),
+                 val,
+                 m,
+                 key,
+                 manifestContext = std::move(manifestContext)]() mutable {
+                    if (auto peer = pairedPeer ? pairedPeer : weak.lock())
+                        peer->checkValidation(
+                            val, key, m, std::move(manifestContext));
                 });
         }
-        else
+        catch (...)
         {
+            if (pairedJob)
+                finishManifestVerification();
+            throw;
+        }
+
+        if (!queued && pairedJob)
+        {
+            finishManifestVerification();
             JLOG(p_journal_.debug())
-                << "Dropping untrusted validation for load";
+                << "manifest_validation candidate_rejected peer=" << id_
+                << " reason=job_queue_refused";
         }
     }
     catch (std::exception const& e)
@@ -2967,12 +3348,138 @@ PeerImp::checkPropose(
     }
 }
 
+void
+PeerImp::sendManifestRepairForSigningKey(PublicKey const& signingKey)
+{
+    if (auto const snapshot =
+            app_.validatorManifests().getManifestSnapshot(signingKey);
+        snapshot && !snapshot->revoked() && snapshot->signingKey &&
+        *snapshot->signingKey == signingKey)
+        sendManifestRepair(
+            snapshot->masterKey, snapshot->sequence, snapshot->serialized);
+}
+
+void
+PeerImp::sendManifestRepair(
+    PublicKey const& masterKey,
+    std::uint32_t sequence,
+    std::string serialized)
+{
+    if (!strand_.running_in_this_thread())
+        return post(
+            strand_,
+            std::bind(
+                &PeerImp::sendManifestRepair,
+                shared_from_this(),
+                masterKey,
+                sequence,
+                std::move(serialized)));
+
+    if (gracefulClose_ || detaching_)
+        return;
+
+    auto const it = manifestAssertionSequences_.find(masterKey);
+    if (it != manifestAssertionSequences_.end() && it->second >= sequence)
+        return;
+
+    recordManifestAssertion(masterKey, sequence);
+
+    protocol::TMManifests tm;
+    tm.add_list()->set_stobject(serialized);
+    send(std::make_shared<Message>(tm, protocol::mtMANIFESTS));
+    JLOG(p_journal_.debug())
+        << "manifest_validation repair_sent peer=" << id_
+        << " master=" << toBase58(TokenType::NodePublic, masterKey)
+        << " sequence=" << sequence;
+}
+
+void
+PeerImp::recordManifestAssertion(
+    PublicKey const& masterKey,
+    std::uint32_t sequence)
+{
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(),
+        "ripple::PeerImp::recordManifestAssertion : on strand");
+
+    auto const it = manifestAssertionSequences_.find(masterKey);
+    if (it != manifestAssertionSequences_.end())
+    {
+        if (it->second < sequence)
+            it->second = sequence;
+        return;
+    }
+
+    if (manifestAssertionSequences_.size() >= maxManifestAssertionEntries)
+        manifestAssertionSequences_.clear();
+    manifestAssertionSequences_.emplace(masterKey, sequence);
+}
+
+bool
+PeerImp::assertedOlderManifest(
+    PublicKey const& masterKey,
+    std::uint32_t sequence) const
+{
+    XRPL_ASSERT(
+        strand_.running_in_this_thread(),
+        "ripple::PeerImp::assertedOlderManifest : on strand");
+
+    auto const it = manifestAssertionSequences_.find(masterKey);
+    return it != manifestAssertionSequences_.end() && it->second < sequence;
+}
+
+void
+PeerImp::sendManifestAssertions(
+    std::shared_ptr<Message> const& message,
+    std::vector<std::pair<PublicKey, std::uint32_t>> assertions)
+{
+    if (!strand_.running_in_this_thread())
+        return post(
+            strand_,
+            std::bind(
+                &PeerImp::sendManifestAssertions,
+                shared_from_this(),
+                message,
+                std::move(assertions)));
+
+    if (gracefulClose_ || detaching_)
+        return;
+
+    for (auto const& [masterKey, sequence] : assertions)
+        recordManifestAssertion(masterKey, sequence);
+    send(message);
+}
+
+void
+PeerImp::finishManifestVerification()
+{
+    // Instrumentation is compiled out in NDEBUG builds; the state transition
+    // must not live inside XRPL_ASSERT.
+    auto const wasInFlight = manifestVerificationInFlight_.exchange(false);
+    XRPL_ASSERT(
+        wasInFlight,
+        "ripple::PeerImp::finishManifestVerification : verification in "
+        "flight");
+    JLOG(p_journal_.debug())
+        << "manifest_validation verification_finished peer=" << id_;
+}
+
 void
 PeerImp::checkValidation(
     std::shared_ptr<STValidation> const& val,
     uint256 const& key,
-    std::shared_ptr<protocol::TMValidation> const& packet)
+    std::shared_ptr<protocol::TMValidation> const& packet,
+    std::optional<PendingManifest> manifestContext)
 {
+    bool const pairedJob = manifestContext.has_value();
+    // The queued job owns the moved association. This scope guard is its
+    // all-exits terminal: success, refusal, and exception consume the active
+    // connection token exactly once.
+    scope_exit finishPairVerification([this, pairedJob]() {
+        if (pairedJob)
+            finishManifestVerification();
+    });
+
     if (!val->isValid())
     {
         std::string desc{"Validation forwarded by peer is invalid"};
@@ -2981,6 +3488,100 @@ PeerImp::checkValidation(
         return;
     }
 
+    std::shared_ptr<protocol::TMManifests const> prerequisite;
+    if (manifestContext)
+    {
+        auto manifest = deserializeManifest(
+            manifestContext->message->list(0).stobject(), p_journal_);
+        if (!manifest || manifest->revoked() || !manifest->signingKey ||
+            manifest->masterKey != manifestContext->masterKey ||
+            *manifest->signingKey != manifestContext->signingKey ||
+            val->getSignerPublic() != manifestContext->signingKey ||
+            !manifest->verify())
+        {
+            std::string const desc{
+                "Validation prerequisite manifest is invalid"};
+            JLOG(p_journal_.debug()) << desc;
+            charge(Resource::feeInvalidSignature, desc);
+            return;
+        }
+
+        // A valid signature does not make a key association admissible. In
+        // particular, an unlisted pair must not traverse ephemerally when its
+        // signing key is already retained for another master. Use the same
+        // key-role rules as durable cache admission before either path can
+        // relay the pair.
+        if (app_.validatorManifests().checkKeyRoles(*manifest))
+        {
+            std::string const desc{
+                "Validation prerequisite manifest has conflicting key roles"};
+            JLOG(p_journal_.debug()) << desc;
+            charge(Resource::feeInvalidData, desc);
+            return;
+        }
+
+        auto const cacheEligible =
+            app_.validators().listed(manifestContext->masterKey);
+
+        if (cacheEligible)
+        {
+            // Only current local policy may affect the durable cache in this
+            // slice. Cache membership alone is not provenance.
+            overlay_.onManifests(manifestContext->message, shared_from_this());
+
+            auto const current = app_.validatorManifests().getManifestSnapshot(
+                manifestContext->signingKey);
+            if (!current || current->revoked() || !current->signingKey ||
+                current->masterKey != manifestContext->masterKey ||
+                *current->signingKey != manifestContext->signingKey ||
+                current->sequence < manifest->sequence)
+            {
+                JLOG(p_journal_.debug())
+                    << "manifest_validation candidate_rejected peer=" << id_
+                    << " reason=not_current_after_application";
+                // Listed admission failed or raced with another update. Do
+                // not reinterpret that failure as permission to relay the
+                // same association ephemerally.
+                return;
+            }
+            else
+            {
+                auto message = std::make_shared<protocol::TMManifests>();
+                message->add_list()->set_stobject(current->serialized);
+                prerequisite = std::move(message);
+            }
+        }
+        else
+        {
+            // Unlisted validators may still traverse a node configured to
+            // relay untrusted validations, but the pair remains ephemeral.
+            prerequisite = manifestContext->message;
+            JLOG(p_journal_.debug())
+                << "manifest_validation candidate_ephemeral peer=" << id_
+                << " master="
+                << toBase58(TokenType::NodePublic, manifestContext->masterKey);
+        }
+
+        if (!app_.validators().trusted(manifestContext->masterKey) &&
+            app_.config().RELAY_UNTRUSTED_VALIDATIONS == -1)
+            return;
+
+        // The pre-verification suppression identity includes both packets.
+        // Only after both signatures pass may this pair claim the actual
+        // validation hash. An invalid prerequisite therefore cannot poison a
+        // later valid validation for the hash-router hold interval.
+        if (auto [added, relayed] =
+                app_.getHashRouter().addSuppressionPeerWithStatus(key, id_);
+            !added)
+        {
+            if (reduceRelayReady() && relayed &&
+                (stopwatch().now() - *relayed) < reduce_relay::IDLED)
+                overlay_.updateSlotAndSquelch(
+                    key, val->getSignerPublic(), id_, protocol::mtVALIDATION);
+            return;
+        }
+    }
+
     // FIXME it should be safe to remove this try/catch. Investigate codepaths.
     try
     {
@@ -2991,8 +3592,8 @@ PeerImp::checkValidation(
             // are the source of the message, consequently the message should
             // not be relayed to these peers. But the message must be counted
             // as part of the squelch logic.
-            auto haveMessage =
-                overlay_.relay(*packet, key, val->getSignerPublic());
+            auto haveMessage = overlay_.relay(
+                *packet, key, val->getSignerPublic(), prerequisite);
             if (reduceRelayReady() && !haveMessage.empty())
             {
                 overlay_.updateSlotAndSquelch(
@@ -3001,6 +3602,16 @@ PeerImp::checkValidation(
                     std::move(haveMessage),
                     protocol::mtVALIDATION);
             }
+
+            // A naked validation that just authenticated against the current
+            // cached manifest is an implicit request for that manifest.
+            // Repair the sender on this connection only; the strand-owned
+            // repair ledger bounds it to one singleton per master/sequence.
+            // Forged, malformed, and unknown-signer traffic never reaches
+            // this point, and paired traffic proves the sender already holds
+            // the prerequisite.
+            if (!pairedJob)
+                sendManifestRepairForSigningKey(val->getSignerPublic());
         }
     }
     catch (std::exception const& ex)

src/xrpld/overlay/detail/PeerImp.h

@@ -175,6 +175,42 @@ private:
     http_response_type response_;
     boost::beast::http::fields const& headers_;
     std::queue<std::shared_ptr<Message>> send_queue_;
+    /** One unverified manifest awaiting a later matching validation.
+
+        Owner/executor: this peer's strand. Normal manifests are structurally
+        decoded on receipt but are not
+        signature-verified or admitted to the global cache until a validation
+        on this connection claims the same signing key. An admitted claim
+        moves the association into the sole in-flight verification job, which
+        frees this slot for one later candidate. Access is serialized by
+        strand_.
+    */
+    struct PendingManifest
+    {
+        std::shared_ptr<protocol::TMManifests> message;
+        PublicKey masterKey;
+        PublicKey signingKey;
+        std::uint32_t sequence;
+    };
+    std::optional<PendingManifest> pendingManifest_;
+
+    // The one active verification-obligation token for this connection. The
+    // strand mints it when pending ownership moves into a job; admission
+    // rollback or that job's terminal clears it exactly once. Atomic because
+    // the job terminal executes on a JobQueue worker, not the peer strand.
+    std::atomic_bool manifestVerificationInFlight_{false};
+
+    // The bounded assertion ledger for this connection. Every retained
+    // manifest sent here records its master/sequence. Backward repair hints
+    // therefore deduplicate against prior prerequisites and each other, and a
+    // terminal response can prove it answers state this connection actually
+    // asserted. Ephemeral pairs do not allocate rows.
+    // Owner/executor: this peer's strand. The ledger is cleared wholesale only
+    // when a new master would exceed the cap; forgetting is safe and may cost
+    // one duplicate repair or one missed best-effort correction.
+    static constexpr std::size_t maxManifestAssertionEntries = 256;
+    hash_map<PublicKey, std::uint32_t> manifestAssertionSequences_;
+
     bool gracefulClose_ = false;
     int large_sendq_ = 0;
     std::unique_ptr<LoadEvent> load_event_;
@@ -442,6 +478,14 @@ public:
         return txReduceRelayEnabled_;
     }
 
+protected:
+    /** Dispatch derived-class work through this peer's serialized executor. */
+    void
+    dispatchOnStrand(std::function<void()> work)
+    {
+        boost::asio::dispatch(strand_, std::move(work));
+    }
+
 private:
     void
     close();
@@ -486,6 +530,18 @@ private:
     void
     doProtocolStart();
 
+    /** Send a validation after its current manifest on this connection.
+
+        Both existing protocol envelopes are enqueued on strand_ in wire
+        order. No receiver-side association or acknowledgement is assumed, so
+        an available prerequisite is sent with every validation.
+    */
+    void
+    sendValidation(
+        std::shared_ptr<Message> const& validation,
+        PublicKey const& signingKey,
+        std::shared_ptr<protocol::TMManifests const> const& prerequisite = {});
+
     // Called when protocol message bytes are received
     void
     onReadMessage(error_code ec, std::size_t bytes_transferred);
@@ -635,7 +691,45 @@ private:
     checkValidation(
         std::shared_ptr<STValidation> const& val,
         uint256 const& key,
-        std::shared_ptr<protocol::TMValidation> const& packet);
+        std::shared_ptr<protocol::TMValidation> const& packet,
+        std::optional<PendingManifest> manifestContext);
+
+    /** Consume the connection's sole active verification-obligation token.
+
+        This is the only terminal operation for both failed job admission and
+        every exit from an admitted verification job.
+    */
+    void
+    finishManifestVerification();
+
+    /** Return a cached manifest to the peer that sent its validation naked.
+
+        A naked validation that has authenticated against the current cached
+        manifest for its signing key is an implicit request for that
+        manifest. The strand consults the bounded repair ledger and sends at
+        most one singleton per master/sequence on this connection. Callable
+        from a verification job; the send hops to the strand.
+    */
+    void
+    sendManifestRepairForSigningKey(PublicKey const& signingKey);
+
+    void
+    sendManifestRepair(
+        PublicKey const& masterKey,
+        std::uint32_t sequence,
+        std::string serialized);
+
+    void
+    recordManifestAssertion(PublicKey const& masterKey, std::uint32_t sequence);
+
+    bool
+    assertedOlderManifest(PublicKey const& masterKey, std::uint32_t sequence)
+        const;
+
+    void
+    sendManifestAssertions(
+        std::shared_ptr<Message> const& message,
+        std::vector<std::pair<PublicKey, std::uint32_t>> assertions);
 
     void
     sendLedgerBase(

src/xrpld/rpc/handlers/Subscribe.cpp

@@ -76,7 +76,6 @@ doSubscribe(RPC::JsonContext& context)
             {
                 auto rspSub = make_RPCSub(
                     context.app.getOPs(),
-                    context.app.getIOService(),
                     context.app.getJobQueue(),
                     strUrl,
                     strUsername,