feat: add allocation reconciliation (#882)

* feat(tap-agent): add periodic allocation reconciliation

   Add a background task that periodically triggers allocation
   reconciliation to ensure recovery after subgraph connectivity issues.

   Previously, the allocation watcher was purely event-driven
   and only fired when the allocation list changed. If allocations were
   closed during a connectivity outage and the list became static afterward,
   no messages would fire and stale SenderAllocation actors would keep
   running, never triggering mark_rav_last().

   This change adds:
   - New config option
    (default: 5 min)
   - ReconcileAllocations message type for SenderAccount
   - Periodic task that sends ReconcileAllocations every
   interval
   - Handler that forces re-check of all allocations against
   the watcher

   This ensures closed allocations are detected and RAVs are
   properly marked
   as 'last' for redemption, even after connectivity issues.

* test(tap-agent): add tests and validation for allocation reconciliation

   - Add unit test for ReconcileAllocations message handling that verifies
     it triggers UpdateAllocationIds with current allocations from watcher
   - Add config validation for allocation_reconciliation_interval_secs:
     error if 0, warn if < 60s
   - Add test for periodic task lifecycle verifying task spawn/abort
   - Update create_sender_account() test helper to expose
     indexer_allocations_tx and configurable reconciliation interval
   - Add tokio test-util feature for time control in tests

* chore(tap-agent): add observability improvements for allocation reconciliation

   - Add ALLOCATION_RECONCILIATION_RUNS counter metric
   - Elevate reconciliation logs from debug to info
   - Enhance config documentation explaining the connectivity failure
   scenario

Author: suchapalaver

crates/config/src/config.rs

@@ -226,6 +226,20 @@ impl Config {
             );
         }
 
+        if self.tap.allocation_reconciliation_interval_secs == Duration::ZERO {
+            return Err(
+                "tap.allocation_reconciliation_interval_secs must be greater than 0".to_string(),
+            );
+        }
+
+        if self.tap.allocation_reconciliation_interval_secs < Duration::from_secs(60) {
+            tracing::warn!(
+                "Your `tap.allocation_reconciliation_interval_secs` value is too low. \
+                This may cause unnecessary load on the system. \
+                A recommended value is at least 60 seconds."
+            );
+        }
+
         // Horizon configuration validation
         // Explicit toggle via `horizon.enabled`. When enabled, require both
         // `blockchain.subgraph_service_address` and
@@ -441,6 +455,23 @@ pub struct TapConfig {
     /// over the escrow balance
     #[serde(default)]
     pub trusted_senders: HashSet<Address>,
+
+    /// Interval in seconds for periodic allocation reconciliation.
+    ///
+    /// Allocation state is normally updated via watcher events from the network subgraph.
+    /// However, if connectivity to the subgraph is lost, allocation closure events may be
+    /// missed. This periodic reconciliation forces a re-check of all allocations against
+    /// the current subgraph state, ensuring stale allocations are detected and processed
+    /// even after connectivity failures.
+    ///
+    /// Default: 300 (5 minutes)
+    #[serde(default = "default_allocation_reconciliation_interval_secs")]
+    #[serde_as(as = "DurationSecondsWithFrac<f64>")]
+    pub allocation_reconciliation_interval_secs: Duration,
+}
+
+fn default_allocation_reconciliation_interval_secs() -> Duration {
+    Duration::from_secs(300)
 }
 
 #[derive(Debug, Deserialize)]

crates/tap-agent/Cargo.toml

@@ -67,3 +67,4 @@ test-log = { workspace = true, features = ["trace"] }
 rstest.workspace = true
 stdext.workspace = true
 insta.workspace = true
+tokio = { workspace = true, features = ["test-util"] }