1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
//! `FeeRouter.SharesUpdated` watcher (ADR 041 / ADR 016 § Tunable Economics).
//!
//! The node seeds its live operator fee-share cell ([`OperatorShares`]) from an
//! authoritative `getShares()` read at startup (see `runtime::mod`). This
//! watcher keeps that cell current: it follows `SharesUpdated` events off the
//! same shared-head `eth_getLogs` poller every other chain watcher uses, and —
//! as a safety net against a missed log — periodically re-reads `getShares()`
//! authoritatively at `fee_shares_poll_interval`. Both paths `store` into the
//! shared [`OperatorShares`], so a governance retune of the fee split reaches
//! the running serve-economics margin calculation without a restart.
//!
//! The sink is read-only and holds no durable cursor, and it scans no history:
//! the tail starts AT head. There is nothing for a lookback to recover — the
//! startup `getShares()` read is authoritative and already reflects every event
//! ever emitted, so re-scanning blocks below it can only re-derive a value the
//! node already holds. The periodic re-read covers anything the tail drops.
use std::sync::Arc;
use std::time::{Duration, Instant};
use alloy::primitives::{Address, U256};
use alloy::providers::Provider;
use alloy::rpc::types::eth::Log;
use alloy::sol_types::SolEvent;
use anyhow::Result;
use decdn_common::redact::sanitize_err_chain;
use decdn_incentive::payment_pool::FeeRouter;
use crate::chain_events::multiplexed_poller::{Route, SinkSource};
use crate::chain_events::resumable_watcher::{CursorStart, LogSink, clear_cadence_on_recovery};
use crate::chain_events::timed;
use crate::fee_shares::{OperatorShares, operator_bps_from_shares};
/// Projection sink for `SharesUpdated`: decodes each event into the shared
/// operator-fee-share cell, and re-reads `getShares()` authoritatively once
/// per `poll_interval` at the end of a clean tick.
struct FeeSharesSink<P: Provider + Clone> {
contract: FeeRouter::FeeRouterInstance<P>,
shares: OperatorShares,
/// Authoritative-re-read cadence (`fee_shares_poll_interval`).
poll_interval: Duration,
/// When the last authoritative re-read ran; `None` until the first.
last_poll: Option<Instant>,
/// Counts failed re-reads into `decdn_fee_shares_watcher_poll_failures_total`.
metrics: Arc<crate::metrics::Metrics>,
}
impl<P: Provider + Clone> FeeSharesSink<P> {
/// Narrow the on-chain `[operator, buyback, treasury]` split to the
/// operator's bps and store it. Fail-closed at the narrowing step
/// (`operator_bps_from_shares` rejects an out-of-range split), so a bad
/// split here simply never overwrites the current cell: it is logged and the
/// previous share is kept, because signing quotes against a value we cannot
/// validate is worse than signing against the last-known-good one.
fn store_shares(&self, raw_shares: [U256; 3], source: &str) {
match operator_bps_from_shares(raw_shares, source) {
Ok(bps) => {
let previous = self.shares.store(bps);
if previous == bps {
tracing::debug!(bps, source, "operator fee share unchanged on chain");
} else {
tracing::info!(
bps,
previous,
source,
"operator fee share updated from chain"
);
}
}
Err(err) => {
tracing::error!(
error = %err,
source,
"fee-shares watcher: on-chain FeeRouter split could not be narrowed to \
operator bps; keeping the current share — governance must fix the split"
);
}
}
}
}
impl<P: Provider + Clone + 'static> LogSink for FeeSharesSink<P> {
async fn apply(&mut self, log: Log) -> Result<()> {
// Filter is scoped to the single SharesUpdated topic; match
// defensively so an unexpected log is skipped, not misdecoded.
if log.topic0() == Some(&FeeRouter::SharesUpdated::SIGNATURE_HASH) {
match FeeRouter::SharesUpdated::decode_log_data(&log.inner.data) {
Ok(ev) => {
self.store_shares(ev.newShares, "event");
}
Err(err) => {
// Undecodable log: log-and-skip (never Err — a deterministic
// re-scan would hot-loop the cursor).
tracing::warn!(error = %err, "fee-shares watcher: undecodable SharesUpdated log; skipping");
}
}
}
Ok(())
}
async fn on_tick_complete(&mut self) -> Result<()> {
let now = Instant::now();
let due = self
.last_poll
.is_none_or(|last| now.duration_since(last) >= self.poll_interval);
if !due {
return Ok(());
}
// Stamp BEFORE the call, not only on success: a persistently failing
// `getShares()` must retry on the slow cadence, not on every watcher
// tick, which would hammer the RPC.
self.last_poll = Some(now);
// Bounded by the per-call timeout: a provider that never answers would
// otherwise stall this tick, and with it every route on the poller.
match timed(
None,
"FeeRouter.getShares()",
self.contract.getShares().call(),
)
.await
{
Ok(shares) => {
self.store_shares(shares, "poll");
}
Err(err) => {
// Best-effort safety net: the event path is primary, so a failed
// re-read logs and keeps the current share rather than backing
// off the whole watcher (which would also stall event pickup).
// Returning Ok keeps the cursor advancing.
self.metrics.fee_shares_watcher_poll_failure();
tracing::warn!(
error = %sanitize_err_chain(&err),
"fee-shares watcher: authoritative getShares() poll failed; keeping current share"
);
}
}
Ok(())
}
/// Force the safety-net re-read on the tick the watcher recovers.
///
/// [`Self::on_tick_complete`] stamps its clock before the call, so an outage
/// spanning a due poll defers the re-read a further `poll_interval` — hours,
/// at the default. The node meanwhile prices its share of every settlement
/// off a split the chain may have moved.
fn on_recovered(&mut self) {
clear_cadence_on_recovery(&mut self.last_poll, self.poll_interval);
}
}
/// Build the fee-shares [`Route`] for the shared multiplexed poller.
/// `poll_interval` is the slower authoritative-re-read safety net
/// (`fee_shares_poll_interval`) that rides the sink's `on_tick_complete`; the
/// merged getLogs cadence is the poller's.
pub fn route<P>(
provider: P,
fee_router_addr: Address,
shares: OperatorShares,
poll_interval: Duration,
metrics: &Arc<crate::metrics::Metrics>,
) -> Route
where
P: Provider + Clone + 'static,
{
let contract = FeeRouter::new(fee_router_addr, provider);
let sink = FeeSharesSink {
contract,
shares,
poll_interval,
// Seeded to "just polled": the runtime performed the authoritative
// startup `getShares()` read moments ago, so leaving this `None`
// would fire a redundant re-read on the very first tick. The first
// safety-net poll is due one `poll_interval` from now.
last_poll: Some(Instant::now()),
metrics: Arc::clone(metrics),
};
Route {
addresses: vec![fee_router_addr],
topic0s: vec![FeeRouter::SharesUpdated::SIGNATURE_HASH],
// Start the tail at head — no historical scan at all. `window_blocks: 0`
// resolves to head exactly. The startup `getShares()` read is the
// authoritative baseline and already folds in every past event, so a
// lookback would re-derive a value the node holds; the periodic re-read
// is the backstop for anything the tail drops. No durable cursor needed.
start: CursorStart::HeadMinusWindow { window_blocks: 0 },
sink: SinkSource::Ready(Box::new(sink)),
label: "fee-shares",
on_established: Some(crate::metrics::metric_hook(
metrics,
crate::metrics::Metrics::fee_shares_watcher_cycle_established,
)),
on_backoff: Some(crate::metrics::metric_hook(
metrics,
crate::metrics::Metrics::fee_shares_watcher_backoff_started,
)),
// Liveness + panic signals. Load-bearing here more than for most
// watchers: this sink's poll-failure and undecodable-log paths both
// return `Ok` by design, so without these a route wedged in RPC backoff
// (or dead) looks identical to a healthy one while the node signs quotes
// against a stale operator fee share.
on_tick_success: Some(crate::metrics::metric_hook(
metrics,
crate::metrics::Metrics::fee_shares_watcher_tick,
)),
on_task_panic: Some(crate::metrics::metric_hook(
metrics,
crate::metrics::Metrics::fee_shares_watcher_task_panicked,
)),
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests;