feat: accept agreement proposals from senders with escrow

crates/config/maximal-config-example.toml

@@ -248,6 +248,10 @@ role_subgraph_max_lag_secs = 1800
 # rolling 24h; rejected and expired proposals don't count. Unset = no cap; 0 rejects all.
 # max_new_agreements_per_24h = 30
 
+# Escrow floor (GRT) admitting a sender that lacks the agreement-manager role: if its recovered
+# signer holds at least this much escrow, the proposal proceeds to pricing/manifest checks. Default 1.
+# min_escrow_grt = "1"
+
 [dips.min_grt_per_30_days]  # base rate component (per-network)
 # arbitrum-one = "450"
 # matic = "300"

crates/config/src/config.rs

@@ -712,6 +712,10 @@ pub struct DipsConfig {
     /// window; rejected and expired proposals don't count. None (unset) disables the
     /// cap; a best-effort safeguard, not a security control.
     pub max_new_agreements_per_24h: Option<u64>,
+    /// Minimum escrow balance an untrusted sender's recovered signer must hold to
+    /// pass the escrow fallback. TODO: derive from the configured price minimum
+    /// plus a buffer for unknown entity counts instead of using a flat floor.
+    pub min_escrow_grt: GRT,
 }
 
 impl Default for DipsConfig {
@@ -729,6 +733,7 @@ impl Default for DipsConfig {
             role_subgraph_max_lag_secs: 1_800,
             rpc_url: None,
             max_new_agreements_per_24h: None,
+            min_escrow_grt: GRT::ONE,
         }
     }
 }

crates/config/src/grt.rs

@@ -12,6 +12,8 @@ pub struct GRT(u128);
 
 impl GRT {
     pub const ZERO: GRT = GRT(0);
+    /// One GRT in wei (10^18).
+    pub const ONE: GRT = GRT(1_000_000_000_000_000_000);
 
     /// Convert GRT string to wei for test construction.
     /// Panics on invalid input - only use in tests.

crates/dips/src/escrow.rs

@@ -0,0 +1,87 @@
+// Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Escrow fallback for agreement-proposal senders that are not on-chain agreement
+//! managers: the role gate would reject them, but this lets one through when its
+//! recovered signer resolves to a payer with enough escrow (a limited permissionless path).
+
+use thegraph_core::alloy::primitives::{Address, U256};
+
+/// Outcome of looking a signer up in the escrow snapshot.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EscrowLookup {
+    /// The snapshot has no signers at all (cold start or total outage), so the
+    /// answer is "can't tell" rather than "unfunded".
+    Unavailable,
+    /// The signer resolved to a funded payer holding this balance (wei).
+    Balance(U256),
+    /// The snapshot is populated but the signer is not a funded payer.
+    NotFound,
+}
+
+/// Read-side view of the escrow snapshot, keyed on the cryptographically
+/// recovered signer (never a caller-written proposal field).
+pub trait EscrowSource: std::fmt::Debug + Send + Sync {
+    fn lookup(&self, signer: Address) -> EscrowLookup;
+}
+
+/// A fixed escrow snapshot. Used in tests and as a simple in-memory source;
+/// production uses the shared TAP escrow watcher.
+#[derive(Debug, Clone, Default)]
+pub struct StaticEscrow(pub std::collections::HashMap<Address, U256>);
+
+impl EscrowSource for StaticEscrow {
+    fn lookup(&self, signer: Address) -> EscrowLookup {
+        if self.0.is_empty() {
+            return EscrowLookup::Unavailable;
+        }
+        match self.0.get(&signer) {
+            Some(balance) => EscrowLookup::Balance(*balance),
+            None => EscrowLookup::NotFound,
+        }
+    }
+}
+
+#[cfg(feature = "db")]
+impl EscrowSource for indexer_monitor::EscrowAccountsWatcher {
+    fn lookup(&self, signer: Address) -> EscrowLookup {
+        let snapshot = self.borrow();
+        if snapshot.signer_count() == 0 {
+            return EscrowLookup::Unavailable;
+        }
+        match snapshot.get_balance_for_signer(&signer) {
+            Ok(balance) => EscrowLookup::Balance(balance),
+            Err(_) => EscrowLookup::NotFound,
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use thegraph_core::alloy::primitives::{address, U256};
+
+    use super::*;
+
+    const SIGNER: Address = address!("f39fd6e51aad88f6f4ce6ab8827279cfffb92266");
+
+    #[test]
+    fn empty_snapshot_is_unavailable() {
+        assert_eq!(
+            StaticEscrow::default().lookup(SIGNER),
+            EscrowLookup::Unavailable
+        );
+    }
+
+    #[test]
+    fn known_signer_returns_balance() {
+        let escrow = StaticEscrow([(SIGNER, U256::from(5))].into_iter().collect());
+        assert_eq!(escrow.lookup(SIGNER), EscrowLookup::Balance(U256::from(5)));
+    }
+
+    #[test]
+    fn populated_snapshot_without_signer_is_not_found() {
+        let other = Address::repeat_byte(0x11);
+        let escrow = StaticEscrow([(other, U256::from(5))].into_iter().collect());
+        assert_eq!(escrow.lookup(SIGNER), EscrowLookup::NotFound);
+    }
+}

crates/dips/src/lib.rs

@@ -50,6 +50,7 @@
 
 use std::sync::Arc;
 
+use escrow::{EscrowLookup, EscrowSource};
 use server::DipsServerContext;
 use thegraph_core::alloy::{
     core::primitives::Address,
@@ -62,6 +63,7 @@ use thegraph_core::alloy::{
 #[cfg(feature = "db")]
 pub mod database;
 pub mod eip5267;
+pub mod escrow;
 pub mod inflight;
 pub mod ipfs;
 #[cfg(any(feature = "rpc", feature = "db"))]
@@ -209,6 +211,10 @@ pub enum DipsError {
     TrustVerificationUnavailable(String),
     #[error("indexer is at its DIPs agreement capacity ({limit} per 24h)")]
     CapacityExceeded { limit: u64 },
+    #[error("sender {signer} has insufficient escrow to send agreement proposals")]
+    InsufficientEscrow { signer: Address },
+    #[error("could not verify sender escrow: {0}")]
+    EscrowVerificationUnavailable(String),
 }
 
 #[cfg(feature = "rpc")]
@@ -297,6 +303,25 @@ pub(crate) fn try_extract_deployment_id(rca_bytes: &[u8]) -> Option<String> {
 /// Window over which the cap counts live agreements (pending or accepted).
 const CAPACITY_WINDOW: std::time::Duration = std::time::Duration::from_secs(24 * 60 * 60);
 
+/// Authorize an untrusted sender by escrow: a resolved balance at or above
+/// `min_wei` passes; below or absent is InsufficientEscrow; a wholesale-empty
+/// snapshot is transient, so a funded sender isn't branded unfunded during an outage.
+fn verify_escrow(
+    escrow: &dyn EscrowSource,
+    signer: Address,
+    min_wei: U256,
+) -> Result<(), DipsError> {
+    match escrow.lookup(signer) {
+        EscrowLookup::Balance(balance) if balance >= min_wei => Ok(()),
+        EscrowLookup::Balance(_) | EscrowLookup::NotFound => {
+            Err(DipsError::InsufficientEscrow { signer })
+        }
+        EscrowLookup::Unavailable => Err(DipsError::EscrowVerificationUnavailable(
+            "escrow snapshot is empty".to_string(),
+        )),
+    }
+}
+
 /// Validate and create a RecurringCollectionAgreement.
 ///
 /// Performs validation:
@@ -323,17 +348,27 @@ pub async fn validate_and_create_rca(
         rca_domain,
         trusted_signers,
         max_new_agreements_per_24h,
+        escrow_source,
+        min_escrow_wei,
     } = ctx.as_ref();
 
     // Decode SignedRCA
     let signed_rca = SignedRecurringCollectionAgreement::abi_decode(rca_bytes.as_ref())
         .map_err(|e| DipsError::AbiDecoding(e.to_string()))?;
 
     // Authenticate then authorize the sender: recover the EIP-712 signer, then
-    // require it to hold the on-chain agreement-manager role before doing any work.
+    // require it to be an on-chain agreement manager, or else a payer with escrow.
     let signer = signed_rca.recover_signer(rca_domain)?;
     tracing::debug!(%signer, "recovered RCA signer");
-    trusted_signers.verify_trusted(signer).await?;
+    match trusted_signers.verify_trusted(signer).await {
+        Ok(()) => {}
+        // Not a manager: fall back to an escrow check on the recovered signer
+        // rather than rejecting; a transient role error stays transient.
+        Err(DipsError::SenderNotTrusted { .. }) => {
+            verify_escrow(escrow_source.as_ref(), signer, *min_escrow_wei)?;
+        }
+        Err(transient) => return Err(transient),
+    }
 
     // Validate service provider
     signed_rca.validate(expected_service_provider)?;
@@ -499,6 +534,7 @@ mod test {
 
     use crate::{
         derive_agreement_id,
+        escrow::StaticEscrow,
         ipfs::{EmptyNetworkIpfsFetcher, FailingIpfsFetcher, MockIpfsFetcher},
         price::PriceCalculator,
         rca_eip712_domain,
@@ -547,21 +583,16 @@ mod test {
             rca_domain: test_rca_domain(),
             trusted_signers: trusted_signers_for_test(),
             max_new_agreements_per_24h: None,
+            escrow_source: Arc::new(StaticEscrow::default()),
+            min_escrow_wei: U256::ZERO,
         })
     }
 
-    #[tokio::test]
-    async fn test_validate_and_create_rca_untrusted_signer_rejected() {
-        let service_provider = Address::repeat_byte(0x11);
-        let rca = create_test_rca(
-            Address::repeat_byte(0x42),
-            service_provider,
-            U256::from(200),
-            U256::from(100),
-        );
-
-        // Trusted set excludes the test signer, so a valid signature is rejected.
-        let ctx = Arc::new(DipsServerContext {
+    /// A context whose trusted set is empty (so the recovered signer is never a
+    /// role holder), with a fixed escrow snapshot and minimum, to exercise C's
+    /// escrow fallback path.
+    fn untrusted_escrow_ctx(escrow: StaticEscrow, min_escrow_wei: U256) -> Arc<DipsServerContext> {
+        Arc::new(DipsServerContext {
             rca_store: Arc::new(InMemoryRcaStore::default()),
             ipfs_fetcher: Arc::new(MockIpfsFetcher::default()),
             price_calculator: Arc::new(PriceCalculator::new(
@@ -574,12 +605,97 @@ mod test {
             rca_domain: test_rca_domain(),
             trusted_signers: Arc::new(StaticTrustedSigners::default()),
             max_new_agreements_per_24h: None,
-        });
-        let rca_bytes = rca_to_wire_bytes(rca);
+            escrow_source: Arc::new(escrow),
+            min_escrow_wei,
+        })
+    }
 
-        let result = super::validate_and_create_rca(ctx, &service_provider, rca_bytes).await;
+    #[tokio::test]
+    async fn untrusted_sender_with_sufficient_escrow_passes() {
+        let service_provider = Address::repeat_byte(0x11);
+        let rca = create_test_rca(
+            Address::repeat_byte(0x42),
+            service_provider,
+            U256::from(200),
+            U256::from(100),
+        );
+        // The recovered signer (test_signer) holds 2 wei; the minimum is 1.
+        let escrow = StaticEscrow(
+            [(test_signer().address(), U256::from(2))]
+                .into_iter()
+                .collect(),
+        );
+        let ctx = untrusted_escrow_ctx(escrow, U256::from(1));
+
+        let result =
+            super::validate_and_create_rca(ctx, &service_provider, rca_to_wire_bytes(rca)).await;
+
+        assert!(result.is_ok(), "got: {:?}", result);
+    }
+
+    #[tokio::test]
+    async fn untrusted_sender_below_minimum_is_insufficient() {
+        let service_provider = Address::repeat_byte(0x11);
+        let rca = create_test_rca(
+            Address::repeat_byte(0x42),
+            service_provider,
+            U256::from(200),
+            U256::from(100),
+        );
+        let escrow = StaticEscrow(
+            [(test_signer().address(), U256::from(1))]
+                .into_iter()
+                .collect(),
+        );
+        let ctx = untrusted_escrow_ctx(escrow, U256::from(5));
+
+        let result =
+            super::validate_and_create_rca(ctx, &service_provider, rca_to_wire_bytes(rca)).await;
+
+        assert!(matches!(result, Err(DipsError::InsufficientEscrow { .. })));
+    }
+
+    #[tokio::test]
+    async fn untrusted_sender_absent_from_escrow_is_insufficient() {
+        let service_provider = Address::repeat_byte(0x11);
+        let rca = create_test_rca(
+            Address::repeat_byte(0x42),
+            service_provider,
+            U256::from(200),
+            U256::from(100),
+        );
+        // Snapshot has a different address, so it is populated but not the signer.
+        let escrow = StaticEscrow(
+            [(Address::repeat_byte(0xEE), U256::from(99))]
+                .into_iter()
+                .collect(),
+        );
+        let ctx = untrusted_escrow_ctx(escrow, U256::from(1));
 
-        assert!(matches!(result, Err(DipsError::SenderNotTrusted { .. })));
+        let result =
+            super::validate_and_create_rca(ctx, &service_provider, rca_to_wire_bytes(rca)).await;
+
+        assert!(matches!(result, Err(DipsError::InsufficientEscrow { .. })));
+    }
+
+    #[tokio::test]
+    async fn untrusted_sender_with_empty_snapshot_is_transient() {
+        let service_provider = Address::repeat_byte(0x11);
+        let rca = create_test_rca(
+            Address::repeat_byte(0x42),
+            service_provider,
+            U256::from(200),
+            U256::from(100),
+        );
+        let ctx = untrusted_escrow_ctx(StaticEscrow::default(), U256::from(1));
+
+        let result =
+            super::validate_and_create_rca(ctx, &service_provider, rca_to_wire_bytes(rca)).await;
+
+        assert!(matches!(
+            result,
+            Err(DipsError::EscrowVerificationUnavailable(_))
+        ));
     }
 
     /// Sign an RCA with the fixed test key over the test domain and ABI-encode
@@ -927,6 +1043,8 @@ mod test {
             rca_domain: test_rca_domain(),
             trusted_signers: trusted_signers_for_test(),
             max_new_agreements_per_24h: None,
+            escrow_source: Arc::new(StaticEscrow::default()),
+            min_escrow_wei: U256::ZERO,
         });
 
         let rca_bytes = rca_to_wire_bytes(rca);
@@ -1103,6 +1221,8 @@ mod test {
             rca_domain: test_rca_domain(),
             trusted_signers: trusted_signers_for_test(),
             max_new_agreements_per_24h: None,
+            escrow_source: Arc::new(StaticEscrow::default()),
+            min_escrow_wei: U256::ZERO,
         });
 
         let rca_bytes = rca_to_wire_bytes(rca);
@@ -1140,6 +1260,8 @@ mod test {
             rca_domain: test_rca_domain(),
             trusted_signers: trusted_signers_for_test(),
             max_new_agreements_per_24h: None,
+            escrow_source: Arc::new(StaticEscrow::default()),
+            min_escrow_wei: U256::ZERO,
         });
 
         let rca_bytes = rca_to_wire_bytes(rca);
@@ -1177,6 +1299,8 @@ mod test {
             rca_domain: test_rca_domain(),
             trusted_signers: trusted_signers_for_test(),
             max_new_agreements_per_24h: None,
+            escrow_source: Arc::new(StaticEscrow::default()),
+            min_escrow_wei: U256::ZERO,
         });
 
         let rca_bytes = rca_to_wire_bytes(rca);
@@ -1214,6 +1338,8 @@ mod test {
             rca_domain: test_rca_domain(),
             trusted_signers: trusted_signers_for_test(),
             max_new_agreements_per_24h: None,
+            escrow_source: Arc::new(StaticEscrow::default()),
+            min_escrow_wei: U256::ZERO,
         });
 
         let rca_bytes = rca_to_wire_bytes(rca);