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"
+    )

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/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/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/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(