[watcher] dedup WithdrawalSignedEvent applies via signatures.is_some()

Sui's SubscribeCheckpoints stream can redeliver a checkpoint, which made
every pod's watcher apply the same WithdrawalSignedEvent twice — the
second apply correctly hit InsufficientCapacity but tripped the sticky
guardian_limiter_drifted bit even though the local limiter was still in
lockstep with the guardian. Gate the apply on signatures.is_some(); the
same gate also skips historical events that the bootstrap snapshot
already counted via GetGuardianInfo's next_seq.

Author: 0xsiddharthks

crates/hashi/src/metrics.rs

@@ -47,6 +47,7 @@ pub struct Metrics {
     pub guardian_limiter_validate_total: IntCounterVec,
     pub guardian_limiter_apply_total: IntCounterVec,
     pub guardian_limiter_anchor_events_total: IntCounter,
+    pub guardian_limiter_anchor_events_skipped_total: IntCounter,
     pub guardian_limiter_batch_truncated_total: IntCounter,
     pub guardian_limiter_batch_stuck_head_total: IntCounter,
     pub guardian_rpc_total: IntCounterVec,
@@ -352,6 +353,12 @@ impl Metrics {
                 registry,
             )
             .unwrap(),
+            guardian_limiter_anchor_events_skipped_total: register_int_counter_with_registry!(
+                "hashi_guardian_limiter_anchor_events_skipped_total",
+                "WithdrawalSignedEvent observations skipped because signatures were already recorded — covers checkpoint-stream redelivery and bootstrap-replay events",
+                registry,
+            )
+            .unwrap(),
             guardian_limiter_batch_truncated_total: register_int_counter_with_registry!(
                 "hashi_guardian_limiter_batch_truncated_total",
                 "Times the leader's approved batch was truncated by local-limiter capacity (the head fits, the tail does not)",

crates/hashi/src/onchain/watcher.rs

@@ -451,6 +451,14 @@ async fn handle_events(
                 // Advance uses the event's checkpoint timestamp (~sign-time)
                 // rather than `txn.timestamp_ms` (creation time) to stay in
                 // lockstep with the guardian's `last_updated_at`.
+                //
+                // `signatures.is_some()` doubles as a "have we already
+                // accounted for this txn?" flag, set on the first apply
+                // here and by `scrape_hashi` at bootstrap. It keeps the
+                // limiter idempotent if `SubscribeCheckpoints` redelivers
+                // a checkpoint, and it also skips historical events that
+                // the bootstrap snapshot already counted via the
+                // guardian's `next_seq`.
                 let (limiter_inputs, pick_to_sign_ms) = {
                     let mut state = state.state_mut();
                     state
@@ -462,17 +470,26 @@ async fn handle_events(
                         .withdrawal_txns
                         .get_mut(&event.withdrawal_txn_id)
                     {
-                        Some(txn) => {
+                        Some(txn) if txn.signatures.is_none() => {
                             txn.signatures = Some(event.signatures.clone());
                             let amount_sats = withdrawal_limiter_consumption_amount(txn);
                             let timestamp_secs = checkpoint_timestamp_ms / 1000;
                             let pick_to_sign =
                                 checkpoint_timestamp_ms.saturating_sub(txn.timestamp_ms);
                             (Some((amount_sats, timestamp_secs)), Some(pick_to_sign))
                         }
-                        None => (None, None),
+                        _ => (None, None),
                     }
                 };
+                if limiter_inputs.is_none() {
+                    tracing::debug!(
+                        withdrawal_txn_id = %event.withdrawal_txn_id,
+                        "Skipping limiter apply: WithdrawalSignedEvent for already-signed txn"
+                    );
+                    if let Some(metrics) = state.metrics() {
+                        metrics.guardian_limiter_anchor_events_skipped_total.inc();
+                    }
+                }
                 if let Some(d) = pick_to_sign_ms
                     && let Some(metrics) = state.metrics()
                 {