fix(signature_collector): skip re-broadcasting an already-broadcast proposer-preferences partial (SIP-94 §5)

[shane-moore]

Blocked on #1142. The guard removes the sender-side re-send, which is currently the only recovery path for a peer that wrongly dropped the first copy; it must not land before receivers validate against complete, unfiltered proposer views (#1142). go-ssv receivers are already tolerant (ssvlabs/ssv#2901). Stacks on #1125.

Goal

Publish a ProposerPreferences partial signature at most once per signing root, while keeping every later signing call for that root a pure poll against the live collector: the share is still contributed idempotently and the caller still awaits reconstruction. Analog of go-ssv's broadcastPreferences guard (proposerPreferencesSlotRunner, ssvlabs/ssv#2901, motivated by ssvlabs/ssv#2934).

Context / motivation

sign_and_collect broadcasts the partial-signature message unconditionally on every call (anchor/signature_collector/src/lib.rs:198-220 at the #1125 head ce37a43b). For role 8 the caller can invoke signing repeatedly for the same root: at the current LH pin the ProposerPreferencesService retries an unpublished epoch every slot, and under any pin a VC restart re-schedules unpublished tuples (the guard must hold regardless of the LH caller's retry discipline, which is changing upstream).

A re-broadcast is byte-identical (deterministic BLS share signature; deterministic PKCS#1 v1.5 operator signature, anchor/message_sender/src/network.rs:195-205; content-hash message id, anchor/network/src/behaviour.rs:45-58), so:

• Within the gossipsub duplicate-cache window (one epoch, anchor/network/src/behaviour.rs:108) it is filtered locally as PublishError::Duplicate and never leaves the node (anchor/network/src/network.rs:399-402). Harmless, but useless.
• Beyond that window it escapes, and peers that accepted the first copy classify the repeated root from the same signer as a duplicate and REJECT it, applying the gossipsub P4 invalid-message penalty against this operator. fix(message_validator): classify proposer-preferences duplicate roots per peer (SIP-94 §7) #1131 narrows this to same-peer repeats (relays become Ignore), but a direct repeat of our own message stays REJECT by design; SIP-94 §7 pins that classification.

§5 partials are one-shot: peers hold the first copy in live state (Anchor: collectors are created on receipt via get_or_spawn, keyed (signing_root, validator_index); go-ssv: per-slot sub-runner plus a pending stash), so once #1142 removes the wrong-drop failure mode a re-send helps nobody and can only cost gossip score.

Suggested approach

Guard the network broadcast, not the share contribution or the await. References are to the #1125 head ce37a43b.

1. Caller-declared policy, derived at one seam

Do not branch on PartialSignatureKind inside the collector crate. Add a policy to the requester:

#[derive(Clone, Copy, Debug, Default, PartialEq)]
pub enum BroadcastPolicy {
    /// Publish the partial signature on every call (today's behavior, all existing duties).
    #[default]
    Always,
    /// Publish at most once per signing root; later calls for the same root only poll the
    /// collector. For one-shot convergence duties (SIP-94 §5) where a same-root re-broadcast
    /// is a same-peer duplicate at peers.
    OncePerRoot,
}

Add broadcast: BroadcastPolicy to SignatureRequester::SingleValidator (signature_collector/src/lib.rs:458-463). collect_signature (validator_store/src/lib.rs:464-537) is the only production constructor of SignatureRequester (workspace-verified); derive the policy in its match arm at :490 from signature_kind == PartialSignatureKind::ProposerPreferences. The ~16 collect_signature call sites need no changes. Known compile fallout: two exhaustive test match patterns gain the field (validator_store/src/testing/payload_attestation.rs:73, validator_store/src/testing/proposer_preferences.rs:121); asserting the new field's value there is better than ...

2. Guard state on the manager

Add a third pruned map next to signature_collectors and committee_partial_signature_batches (signature_collector/src/lib.rs:89-94), following the same keyed-with-slot pattern:

/// Signing roots already published under BroadcastPolicy::OncePerRoot, with the wire slot
/// for pruning. Only consulted for OncePerRoot requesters.
broadcast_roots: DashMap<(Hash256, ValidatorIndex), Slot>,

Keyed on (signing_root, validator_index), never (slot, validator): a re-emission under a changed dependent_root hashes to a new root and broadcasts normally for free, while a same-root retry is suppressed. Root equality is preference equality for a given proposal slot (the domain is fixed by the slot's epoch), matching go-ssv's whole-preference comparison.

3. Guard placement in sign_and_collect

In the SingleValidator arm of the signing closure (:198-220), before create_message:

• If the policy is OncePerRoot and broadcast_roots contains the key, skip create_message and message_sender.sign_and_send entirely and fall through.
• Otherwise send as today, and on the OncePerRoot path insert the key only after sign_and_send returns Ok. Note Ok means enqueued on the sender's processor queue, not on-wire (message_sender/src/network.rs:67-98); insert-after-Ok still correctly avoids latching on the realistic synchronous failures (NotSynced at startup, queue full/closed), and a post-enqueue drop latching the guard without a wire publish is an accepted residual.
• The own-share contribution (:283-287, idempotent; the conflicting-share check at :660-674 stays) and the notifier await (:294) run unconditionally, so with the feat(validator_store): implement sign_proposer_preferences (SingleValidator partial-sig) #1063 bounded collection timeout each caller retry acts as a pure poll that picks up a late-forming quorum. The skip path has no early return; the only return in the arm sits inside the skipped create_message error block (:207-210).

4. Pruning: epoch-scale cutoff, not the shared collector cutoff

The cleaner (:411-429) prunes collectors at cutoff = now - SIGNATURE_COLLECTOR_RETAIN_SLOTS(1). Reusing that cutoff for the guard would free entries at proposal_slot + 2 while the peer REJECT window runs one epoch from first publish, reopening a re-publish window for a late same-epoch retry. Prune broadcast_roots with its own retain, for_slot >= now.saturating_sub(slots_per_epoch), using the manager's existing slots_per_epoch field (:85). Memory stays trivial: one entry per preferences signing, held for about an epoch past its proposal slot.

Intentional behaviors (do not "fix" later)

• The guard latches in impostor mode too: the impostor sender's "Would send" is unconditional today (message_sender/src/impostor.rs:31-45), so recording keeps the observable behavior faithful.
• A retry arriving after the collector was reaped suppresses the publish and fails with QueueClosedError. Correct: re-publishing our share could not regather the peers' shares either; they send once, at their own emission tick.
• Check-then-insert is not atomic across truly concurrent same-root calls; the worst case is today's behavior (one extra publish, filtered by the local duplicate cache). Calls for one validator come from the sequential LH loop, so the race is not reachable in practice.

go-ssv parity

go-ssv (ssvlabs/ssv#2901)	This issue
broadcastPreferences: skip re-sign/re-broadcast of an unchanged preference	broadcast_roots: root-keyed publish skip (root equality = preference equality per slot)
Guard carried across sub-runner replacements	Manager-level map survives caller retries by construction
Pruned by evictPastSlots	Epoch-scale cleaner retain
Pending stash re-seeds the replacement's quorum	Collectors are created on receipt (get_or_spawn); a guarded retry awaits as a pure poll
In-memory, lost on restart	Same; accepted on both sides

Acceptance criteria

• A repeated collection for the same (signing_root, validator_index) under OncePerRoot publishes no second gossip message; the await still resolves from the live collector when quorum forms between calls.
• A collection for the same proposal_slot under a different signing root (dependent_root re-emission) publishes normally.
• Broadcast behavior for every Always requester (all other duties) is unchanged.
• Our own share is still contributed on a guarded retry; a conflicting duplicate share still errors as today.
• A synchronous sign_and_send failure does not record the root; the next call attempts the publish again.
• Guard entries are pruned roughly one epoch after their wire slot; collector reap timing is unchanged.
• ProposerPreferences is the only kind deriving OncePerRoot; the derivation lives in collect_signature, not in the collector crate.

Tests

The existing signature_collector/src/tests.rs only unit-tests SignatureCollectorState; guard behavior needs a small manager-level harness: processor Senders + ManualSlotClock + ForkSchedule + message_sender::testing::MockMessageSender (message_sender/src/testing.rs:9-51, captures sent messages on a channel; construction pattern to copy: qbft_manager/src/tests.rs:311).

• N same-root OncePerRoot calls publish exactly once; collection resolves when peer shares arrive between calls.
• Changed root, same slot: a second publish is observed.
• An Always requester called twice publishes twice.
• A mock sender failing the first sign_and_send: the second call publishes (root not recorded on failure).
• Guard entries survive the collector cutoff and are gone after the epoch-scale cutoff.
• validator_store (existing mocked-collector harness, testing/common.rs:120-166): the captured requester carries OncePerRoot for sign_proposer_preferences and Always for sign_payload_attestation.

Notes

• Do not implement before fix(duties_tracker): validate proposer assignments against complete current/next duty views (SIP-94 §5/§7) #1142 merges; landing this first makes a wrongly-dropped partial permanently unrecoverable at Anchor peers.
• Alternatives considered and rejected: broadcasting from inside the collector task on first own-share receipt (threads message construction and I/O into the deliberately minimal collector task at :576-593 and delays first publish by a queue hop); a guard inside validator_store (duplicates lifecycle machinery in a crate that owns neither broadcast nor pruning); no guard (every caller retry is a duplicate publish, no go-ssv parity).
• File and line references are at the feat(validator_store): implement sign_proposer_preferences (SingleValidator partial-sig) #1125 head (ce37a43b); verify symbols at PR time.

Issues are directionally correct, not prescriptive; verify symbols at PR time.