feat(core): implement core/fetcher

[emlautarom1]

The fetcher is stage 2 of the duty pipeline (Scheduler → Fetcher → Consensus → …): it takes a Duty plus a per-validator DutyDefinitionSet and fetches the corresponding unsigned duty data from the beacon node, producing an UnsignedDataSet that it hands to its subscribers.

Highlights

• New crates/core/src/fetcher/ module with the full Fetcher and a GraffitiBuilder, covering all four duty paths: attester, aggregator, block proposer, and sync-committee contribution.
• DutyDefinition reworked into a heterogeneous enum (Attester / Proposer / SyncCommittee), mirroring Charon's core.DutyDefinition interface — replacing the previous generic wrapper.
• UnsignedDutyData / UnsignedDataSet promoted into core::types so the fetcher, DutyDB, and consensus share one canonical, PubKey-keyed type (matching Go's map[core.PubKey]core.UnsignedData).
• Every Go test ported and passing.
• No new dependencies introduced.

Type-system changes (crates/core/src/types.rs)

The fetcher's input and output types did not previously exist in a usable form; both were reworked first.

DutyDefinition → enum

Charon models duty definitions as a DutyDefinition interface with three implementations (AttesterDefinition, ProposerDefinition, SyncCommitteeDefinition). Pluto previously had a generic DutyDefinition<T> / DutyDefinitionSet<T> which cannot represent a heterogeneous, type-dispatched set the way the fetcher needs.

This was replaced with:

• enum DutyDefinition { Attester(..), Proposer(..), SyncCommittee(..) } with as_attester / as_proposer / as_sync_committee accessors (the Rust equivalent of Go's type assertions).
• pub type DutyDefinitionSet = HashMap<PubKey, DutyDefinition> (matches Go's map[core.PubKey]core.DutyDefinition).
• New ProposerDuty and SyncCommitteeDuty structs (faithful to eth2 v1.ProposerDuty / v1.SyncCommitteeDuty) and the three *Definition newtype wrappers. AttesterDefinition wraps the pre-existing signeddata::AttesterDuty.

UnsignedDataSet promoted to core::types

UnsignedDutyData (enum over Proposal / Attestation / AggAttestation / SyncContribution) and UnsignedDataSet = HashMap<PubKey, UnsignedDutyData> were moved out of dutydb/memory.rs into core::types. The old generic type was removed.

Fetcher implementation (crates/core/src/fetcher/)

graffiti.rs

Port of charon/core/fetcher/graffiti.go: GraffitiBuilder, client_graffiti_mappings, and the build/default/token helpers. The default graffiti uses a pluto/<version>-<commit> prefix (vs Go's charon/...).

mod.rs

Port of charon/core/fetcher/fetcher.go:

• Fetcher::new, subscribe, register_agg_sig_db, register_await_att_data, and fetch (dispatch on DutyType).
• The four data paths: fetch_attester_data, fetch_aggregator_data, fetch_proposer_data, fetch_contribution_data, plus verify_fee_recipient and the pubkeys logging tracker.
• Beacon-node calls map onto the generated EthBeaconNodeApiClient: produce_attestation_data, get_aggregated_attestation_v2, produce_block_v3, produce_sync_committee_contribution. Loosely-typed responses are decoded into the strongly-typed signeddata types via JSON round-trip; the produce_block_v3 response is mapped into an unsigned VersionedProposal per fork/blinded variant.
• Aggregator selection reuses the existing pluto_eth2util::eth2exp (is_att_aggregator, is_sync_comm_aggregator).
• AggSigDB and DutyDB are injected as async callbacks (no hard dependency on those components); Box<dyn SignedData> values from AggSigDB are down-cast to concrete types via trait upcasting.
• Go error strings preserved, e.g. aggregate attestation not found by root (retryable), and the errors.Wrap(err, "fetch <type> data") wrapping is reproduced via a FetcherError::Fetch { context, source } variant.

Tests

All Go tests are ported and passing (test names kept where practical):

• Graffiti (5, internal): fetch_beacon_node_token, build_graffiti, default_graffiti, get_graffiti, new_graffiti_builder.
• Attester: fetch_attester.
• Aggregator (4): different_committee, same_committee, no_aggregator, nil_aggregate.
• Sync contribution (3): aggregator, not_aggregator, data_error.
• Blocks: fetch_blocks.
• Fee recipient (internal): verify_fee_recipient.

Per the chosen approach, beacon-node responses that depend on request parameters (aggregate attestation by data root, sync contribution by subcommittee/root, block proposal echoing randao + graffiti) are provided by test-local wiremock responders rather than by extending BeaconMock. The block-proposal test reuses the existing Electra VersionedProposal golden fixture as the produce_block_v3 payload. Aggregator/sync tests use a custom spec carrying the selection fields required by eth2exp.

verify_fee_recipient's comparison logic was factored into a pure fee_recipient_mismatch helper so it can be asserted directly without a tracing-log capture harness; the warning itself still lives in verify_fee_recipient.

Files changed

• crates/core/src/types.rs — DutyDefinition enum, new duty/definition types, promoted UnsignedDutyData/UnsignedDataSet.
• crates/core/src/dutydb/memory.rs, crates/core/src/dutydb/mod.rs — import and re-export the promoted types instead of defining them.
• crates/core/src/lib.rs — register the fetcher module.
• crates/core/src/fetcher/mod.rs, crates/core/src/fetcher/graffiti.rs — new.

Caveats and notes for review

• Pre-existing deviation: Pluto's AttesterDuty omits the PubKey field present in eth2's v1.AttesterDuty. The fetcher does not depend on it (the pubkey is always the map key), but it is worth keeping in mind when the scheduler begins populating these definitions.
• The fetcher's DutyDefinitionSet is produced by the scheduler, which is in-flight on emlautarom1/core-scheduler. The definition types added here may need reconciliation if that branch introduces overlapping types.

---

This PR was automatically generated by Claude.

[emlautarom1]

Walking through the parts of this change that warrant the most discussion. Overall the structure mirrors Charon faithfully and the test coverage looks solid. Leaving notes on a few design points rather than blocking changes.

[emlautarom1]

Async-callback shape note: Arc<dyn Fn(...) -> Pin<Box<dyn Future + Send>>> is the same idiom sigagg already uses, so we're consistent across components.

BoxError = Box<dyn std::error::Error + Send + Sync> is intentionally loose so callers (AggSigDB / DutyDB / subscribers) can return their own error types without dragging concrete types into the fetcher's public surface. The cost is that errors get stringified at the boundary.

[emlautarom1]

@claude Review this code

[claude]

Claude finished @emlautarom1's task in 3m 44s —— View job

---

Code review — feat(core): implement core/fetcher

• Read PR diff and changed files
• Cross-reference with Charon Go source
• Review types.rs changes (DutyDefinition enum, UnsignedDataSet promotion)
• Review fetcher/mod.rs (Fetcher core, dispatch, four data paths, fee-recipient)
• Review fetcher/graffiti.rs
• Review test coverage and parity
• Post review

Summary

Solid port. Structure, dispatch, sub-fetchers, graffiti, and tracker mirror charon/core/fetcher faithfully. Type-system reshuffle (DutyDefinition → enum, UnsignedDataSet → core::types) is the right call and lines up with how Charon's interface is used. Test parity is good. Findings below are mostly observability and edge-case concerns, not blockers.

---

Findings

1. PubkeysTracker::log() is skipped on early-return error paths — divergence from Go's defer

crates/core/src/fetcher/mod.rs:309 and :441

Go uses defer pt.log(ctx) in fetchAggregatorData / fetchContributionData, so the not-selected / resolved pubkey lists are logged even when the function bails out partway through (e.g., AggSigDB error, BN error). The Rust port calls tracker.log() only on the success path:

let agg_att = self.aggregate_attestation(slot, comm_idx, data_root).await?;  // ← early return
...
tracker.log();
Ok(resp)

So in the nil_aggregate / data_error scenarios you lose the partial observability Go provides. Easy fix: wrap the tracker in a small RAII guard (Drop impl that logs), or split the loop body into a closure and call tracker.log() after the closure returns regardless of result. Fix this →

2. fee_recipient_mismatch silently swallows missing/malformed payload fields

crates/core/src/fetcher/mod.rs:730-756

The cleanup to a pure helper is nice. But the JSON-path traversal uses ? on each step, so any of these conditions silently return None (= "no mismatch, no warning"):

• proposal_body(...) returns Null (every fork arm uses .unwrap_or(Value::Null))
• execution_payload / execution_payload_header key absent
• fee_recipient key absent
• as_str() fails

Go's typed accessors would panic / mis-compile if a future fork renames the key — the Rust port simply stops warning. Combined with the test gap below this means a hypothetical Fulu+1 rename could ship as a no-op warning. Consider one of:

• debug_assert!(actual_addr.is_some(), ...) so dev/test catches it,
• log a warn! if we have a Bellatrix+ proposal but couldn't extract the address.

You explicitly call this out in the PR description; flagging here as it directly intersects with finding 3.

3. verify_fee_recipient test coverage gap vs Go

crates/core/src/fetcher/mod.rs:929-952

TestVerifyFeeRecipient in Go exercises 10 fork/blinded combinations (Bellatrix..Fulu × {blinded, unblinded}). The Rust test only covers Electra-unblinded via the golden. With the silent-fail behavior in 2., the other 9 paths are effectively untested for both fee_recipient_mismatch and proposal_body matching.

You don't need a BeaconMock log harness for this — fee_recipient_mismatch is already pure. A table-driven test that builds a minimal proposal per fork (deterministic addresses, no golden needed for the smaller bodies) would lock the JSON shape per variant. The blinded variants in particular only need execution_payload_header.fee_recipient. Fix this →

4. Non-Ok BN response collapses 404 and 5xx into the same retryable error

crates/core/src/fetcher/mod.rs:534-539 (and similarly for produce_attestation_data / produce_sync_committee_contribution)

GetAggregatedAttestationV2Response::Ok(ok) => ok,
_ => return Err(FetcherError::AggregateAttestationNotFound),

Go inspects eth2Resp.Data == nil separately from err != nil, so a 5xx surfaces as a non-retryable API error while a nil-data 200 surfaces as the retryable "not found by root". The Rust collapse treats any non-Ok variant — including server errors — as retryable. Whether that's a problem depends on what the upstream pipeline does with AggregateAttestationNotFound: if it triggers a tight retry loop, you'll hammer a broken BN. Worth confirming.

5. Fetcher::fetch clones unsigned_set per subscriber — including the last

crates/core/src/fetcher/mod.rs:245-251

for sub in &self.subs {
    let clone = unsigned_set.clone();
    sub(duty.clone(), clone).await.map_err(FetcherError::Callback)?;
}

This clones once per subscriber even though the last subscriber could take the value by move. Charon does the same (it always clones), so this is parity, not a bug. Mentioning only because it's a per-fetch hot path and the set can be large. A small optimization for later: iterate with iter().enumerate() and mem::take (or into_iter over a vec) for the final call. Optional.

6. Fetcher::new returns Self while Go's New returns (*Fetcher, error)

crates/core/src/fetcher/mod.rs:163-182

Go's New never actually returns an error (always nil), so the Rust simplification is fine. Just flagging so anyone reading both files knows they're equivalent.

7. Phase0/Altair are accepted by versioned_proposal_from_response but BN should never produce them

crates/core/src/fetcher/mod.rs:632-633

(ConsensusVersion::Phase0, _) => ProposalBlock::Phase0(json_from(&data)?),
(ConsensusVersion::Altair, _) => ProposalBlock::Altair(json_from(&data)?),

Go has no such branches — produce_block_v3 only emits Bellatrix+. Decoding them isn't wrong, but it's quietly accepting input the upstream contract says is impossible, and downstream code (verify_fee_recipient) silently no-ops on them. Either reject explicitly (return Err(FetcherError::UnsupportedProposalVersion(...))) or document the looseness. Minor.

8. Minor: stale use std::any::Any is needed only inside downcast

crates/core/src/fetcher/mod.rs:9 imports Any at module scope. It's referenced only in downcast at :603. Keep or scope-local; clippy is fine either way.

---

Nits / observations

• crates/core/src/fetcher/mod.rs:42 — Subscriber = Arc<dyn Fn(...) -> ...> matches the sigagg idiom as noted; consistency is the right call.
• crates/core/src/fetcher/mod.rs:625 (versioned_proposal_from_response) — agreed with the inline note: once eth2api lands a versioned decode, swap the match out. The mapping is correct per-fork as far as I can verify; the Fulu arm is symmetrical to Deneb/Electra.
• crates/core/src/fetcher/graffiti.rs:135-138 — pluto/<version>-<commit> prefix (vs. charon/...) is the right call and matches the rest of the workspace.
• crates/core/src/types.rs:507-541 — DutyDefinition enum + as_* accessors is the natural Rust shape for the Go interface, and the deliberate inability to add variants outside the crate is correct (matches Charon's closed set of impls in core/).
• crates/core/src/types.rs:560-563 — promoting UnsignedDutyData / UnsignedDataSet to core::types and re-exporting from dutydb is the right de-duplication; aligns with how core.UnsignedData lives in Go's core package.
• The register_* / subscribe &mut self + "call before fetch" contract carries over cleanly; just make sure the integration site honours it (no mid-flight registration).

---

Conclusion

LGTM with the caveats above. The two I'd actually act on before merging:

1. Move tracker.log() to fire on every exit path (1-line change with Drop).
2. Expand verify_fee_recipient to cover every fork/blinded combo — closes the silent-fail risk in 2. at near-zero cost.

The rest are observations and judgment calls.
· Branch: worktree-graceful-spinning-wombat

[emlautarom1]

@claude Review this PR and take into consideration your previous comments.

[claude]

Claude finished @emlautarom1's task in 2m 3s —— View job

---

Re-review — feat(core): implement core/fetcher

Re-reviewed against my previous review. The branch has had 14+ follow-up commits, including a meaningful round of refactors. Here's where each prior finding stands.

✅ Addressed since last pass

#	Finding	Resolution
1	PubkeysTracker::log() skipped on early-return paths (Go's defer divergence)	impl Drop for PubkeysTracker (mod.rs:703) now logs on every scope exit — the RAII fix I suggested. The explicit .log() calls were removed, so no double-logging. Matches Go's defer pt.log(ctx) semantics.
8	use std::any::Any at module scope	Now a full path std::any::Any inside downcast (mod.rs:560).
—	JSON round-trips for BN responses (called out in PR description)	Replaced with direct TryFrom conversions: phase0::AttestationData::try_from (mod.rs:467), altair::SyncCommitteeContribution::try_from (:528), and a versioned attestation_payload (:574). The new ConversionError path is wired into FetcherError::Conversion. Cleaner and strongly typed — nice improvement.
—	produce_block_v3 per-fork mapping	Moved into VersionedProposal::try_from (signeddata.rs:1377), and ConsensusVersion → DataVersion into version.rs with full per-fork test coverage (version.rs:231-240). Good de-duplication.
7 (partial)	aggregate-attestation version/shape mismatch silently accepted	attestation_payload now explicitly returns FetcherError::UnexpectedResponse on any version/shape mismatch (mod.rs:593) instead of decoding loosely.

⚠️ Still open

2. fee_recipient_mismatch still silently swallows missing/malformed payload fields — mod.rs:612-638

Unchanged. The traversal still ?-bails to None (= "no mismatch, no warning") if execution_payload/execution_payload_header or fee_recipient is absent, or proposal_body returns Null. A future fork key rename would ship as a silent no-op warning. A debug_assert!/warn! when a Bellatrix+ proposal yields no extractable address would catch this in dev/test. Fix this →

3. verify_fee_recipient test still only covers Electra-unblinded — mod.rs:833-858

Unchanged. Go's TestVerifyFeeRecipient exercises 10 fork/blinded combos; the Rust test (verify_fee_recipient) covers Electra-unblinded only. Combined with 2., the other 9 paths — especially the blinded variants reading execution_payload_header.fee_recipient — are untested. Since fee_recipient_mismatch is pure, a table-driven test per variant locks the JSON shape at near-zero cost. This is the one I'd most encourage before merge. Fix this →

4. Non-Ok BN response still collapses 404 and 5xx into the same retryable error — mod.rs:466, :496, :530

Unchanged. _ => return Err(FetcherError::AggregateAttestationNotFound) (and the attestation-data / sync-contribution equivalents) treat any non-Ok variant — including server errors — as the retryable "not found by root", whereas Go separates nil-data from transport/API errors. Whether this matters depends on how the upstream pipeline reacts to the retryable variant; worth confirming it won't tight-loop against a 5xx-ing BN. Judgment call, not a blocker.

7 (proposal). Phase0/Altair arms in VersionedProposal::try_from — signeddata.rs:1387-1388

produce_block_v3 never emits pre-Bellatrix blocks, but the match still has Phase0/Altair arms (and verify_fee_recipient correctly no-ops on them). Harmless looseness; leaving as-is is defensible. Minor.

5. unsigned_set.clone() per subscriber — mod.rs:209-215 — parity with Charon, optional micro-opt. Unchanged.

Verdict

The two findings I actually flagged for action last time are half-done: 1. is fixed, 3. (test coverage) is the remaining gap and is the one I'd still close before merge — it directly de-risks the silent-fail in 2. at near-zero cost. The TryFrom/versioned-decode refactors are a clear quality improvement over the original JSON round-trips. CI is green so far (4 passed, 4 in-progress, 0 failed). LGTM with the fee-recipient test expansion encouraged.
· Branch: worktree-graceful-spinning-wombat

[emlautarom1]

The following issue is still present:

Non-Ok BN response still collapses 404 and 5xx into the same retryable error** — mod.rs:466, :496, :530

We'll keep it simple for now and revisit this once we start doing the wiring. I don't think we should be spending much time on getting retrying working immediately without sufficient context.

[emlautarom1]

@claude Review this PR and take into consideration your previous comments and my own.

[claude]

Claude finished @emlautarom1's task in 1m 27s —— View job

---

Re-review — feat(core): implement core/fetcher

Re-reviewed against my previous review and your note deferring finding 4. The two findings I'd flagged for action before merge are now both closed.

✅ Resolved

#	Finding	Resolution
1	PubkeysTracker::log() skipped on early-return (Go defer divergence)	impl Drop for PubkeysTracker (mod.rs:718) logs on every scope exit; explicit .log() calls removed. Matches defer pt.log(ctx).
2	fee_recipient_mismatch silently swallowed missing/malformed payload fields	Now guarded (mod.rs:636-646): a Bellatrix+ proposal that yields no extractable fee_recipient hits a debug_assert!(false, …) (fails in dev/test) and a tracing::warn! (surfaces in prod) instead of returning a false "no mismatch". Exactly the fix I suggested.
3	verify_fee_recipient test only covered Electra-unblinded	Now table-driven over all 10 combos — Bellatrix..Fulu × {unblinded, blinded} (mod.rs:910-981). Each case asserts a mismatch vs 0xdead, then None against its own address and its uppercased form (case-insensitive path). The blinded cases exercise the execution_payload_header fallback, and Electra+ blinded gets the required execution_requests. Closes the silent-fail risk from 2. at the structural level.

The earlier round of TryFrom/versioned-decode refactors (replacing the JSON round-trips, moving per-fork proposal decoding into VersionedProposal::try_from, ConsensusVersion → DataVersion in version.rs) remain a clear quality improvement over the original.

Acknowledged / deferred

4. Non-Ok BN response collapses 404 and 5xx into the retryable …NotFound — mod.rs:466, :496, :530

Understood — keeping it simple until the pipeline wiring lands and you have the context to decide the right retry/error split. Agree this shouldn't block here; just flagging that when wiring does happen, the call site that consumes AggregateAttestationNotFound is where a 5xx-vs-nil-data distinction will matter (to avoid tight-looping a broken BN). No action now.

Still-open minors (non-blocking, unchanged)

• 5 — unsigned_set.clone() per subscriber (mod.rs:209-215): parity with Charon, optional micro-opt.
• 7 — Phase0/Altair arms in VersionedProposal::try_from: produce_block_v3 never emits them; fee_recipient_mismatch correctly early-returns None for those versions (mod.rs:615-620). Harmless looseness.

Verdict

LGTM. Both pre-merge findings (1. fixed earlier, 2. + 3. fixed in a668f7a) are resolved, 4 is a deliberate deferral, and the remainder are judgment-call minors. CI green so far (6 passed, "Build and test" in progress, 0 failed).
· Branch: worktree-graceful-spinning-wombat