handle (panic) on missing payment chosen platform

agent/provider/src/market/negotiator/builtin.rs

@@ -1,3 +1,4 @@
+pub mod demand_validation;
 pub mod expiration;
 pub mod manifest;
 pub mod max_agreements;

agent/provider/src/market/negotiator/builtin/demand_validation.rs

@@ -0,0 +1,139 @@
+use ya_agreement_utils::OfferDefinition;
+
+use crate::market::negotiator::factory::DemandValidationNegotiatorConfig;
+use crate::market::negotiator::{
+    AgreementResult, NegotiationResult, NegotiatorComponent, ProposalView,
+};
+
+/// Negotiator that verifies that all required fields are present in proposal.
+pub struct DemandValidation {
+    required_fields: Vec<String>,
+}
+
+impl DemandValidation {
+    pub fn new(config: &DemandValidationNegotiatorConfig) -> DemandValidation {
+        let required_fields = config.required_fields.clone();
+        Self { required_fields }
+    }
+}
+
+impl NegotiatorComponent for DemandValidation {
+    fn negotiate_step(
+        &mut self,
+        demand: &ProposalView,
+        offer: ProposalView,
+    ) -> anyhow::Result<NegotiationResult> {
+        let missing_fields = self
+            .required_fields
+            .iter()
+            .filter(|x| demand.pointer(x).is_none())
+            .cloned()
+            .collect::<Vec<String>>();
+        if missing_fields.is_empty() {
+            Ok(NegotiationResult::Ready { offer })
+        } else {
+            log::info!(
+                "'DemandValidation' negotiator: Reject proposal [{}] due to missing fields: {}",
+                demand.id,
+                missing_fields.join(",")
+            );
+            Ok(NegotiationResult::Reject {
+                message: format!("Missing fields: {}", missing_fields.join(",")),
+                is_final: false,
+            })
+        }
+    }
+
+    fn fill_template(
+        &mut self,
+        offer_template: OfferDefinition,
+    ) -> anyhow::Result<OfferDefinition> {
+        Ok(offer_template)
+    }
+
+    fn on_agreement_terminated(
+        &mut self,
+        _agreement_id: &str,
+        _result: &AgreementResult,
+    ) -> anyhow::Result<()> {
+        Ok(())
+    }
+
+    fn on_agreement_approved(&mut self, _agreement_id: &str) -> anyhow::Result<()> {
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use serde_json::json;
+    use ya_agreement_utils::agreement::expand;
+    use ya_agreement_utils::OfferTemplate;
+    use ya_client_model::market::proposal::State;
+
+    fn config() -> DemandValidationNegotiatorConfig {
+        DemandValidationNegotiatorConfig {
+            required_fields: vec![
+                "/golem/com/freebies".to_string(),
+                "/golem/com/payment/address".to_string(),
+            ],
+        }
+    }
+
+    fn properties_to_proposal(properties: serde_json::Value) -> ProposalView {
+        ProposalView {
+            content: OfferTemplate {
+                properties: expand(properties),
+                constraints: "()".to_string(),
+            },
+            id: "proposalId".to_string(),
+            issuer: Default::default(),
+            state: State::Initial,
+            timestamp: Utc::now(),
+        }
+    }
+
+    /// Negotiator accepts demand if all of the required fields exist
+    #[test]
+    fn test_required_fields_exist() {
+        let config = config();
+        let mut negotiator = DemandValidation::new(&config);
+
+        let offer = properties_to_proposal(json!({}));
+        let demand = properties_to_proposal(json!({
+            "golem.com.freebies": "mug",
+            "golem.com.payment.address": "0x123",
+        }));
+
+        let expected_result = NegotiationResult::Ready {
+            offer: offer.clone(),
+        };
+        assert_eq!(
+            negotiator.negotiate_step(&demand, offer).unwrap(),
+            expected_result
+        );
+    }
+
+    /// Negotiator rejects demand if some of the required fields are missing
+    #[test]
+    fn test_required_fields_missing() {
+        let config = config();
+        let mut negotiator = DemandValidation::new(&config);
+
+        let offer = properties_to_proposal(json!({}));
+        let demand = properties_to_proposal(json!({
+            "golem.com.freebies": "mug",
+        }));
+
+        let expected_result = NegotiationResult::Reject {
+            message: "Missing fields: /golem/com/payment/address".to_string(),
+            is_final: false,
+        };
+        assert_eq!(
+            negotiator.negotiate_step(&demand, offer).unwrap(),
+            expected_result
+        );
+    }
+}

agent/provider/src/market/negotiator/composite.rs

@@ -13,6 +13,7 @@ use super::builtin::{
 };
 use super::common::{offer_definition_to_offer, AgreementResponse, Negotiator, ProposalResponse};
 use super::{NegotiationResult, NegotiatorsPack};
+use crate::market::negotiator::builtin::demand_validation::DemandValidation;
 use crate::market::negotiator::common::{
     reason_with_extra, AgreementFinalized, CreateOffer, ReactToAgreement, ReactToProposal,
 };
@@ -33,6 +34,10 @@ impl CompositeNegotiator {
         agent_negotiators_cfg: AgentNegotiatorsConfig,
     ) -> anyhow::Result<CompositeNegotiator> {
         let components = NegotiatorsPack::default()
+            .add_component(
+                "Validation",
+                Box::new(DemandValidation::new(&config.validation_config)),
+            )
             .add_component(
                 "LimitAgreements",
                 Box::new(MaxAgreements::new(&config.limit_agreements_config)),

agent/provider/src/market/negotiator/factory.rs

@@ -11,6 +11,13 @@ use crate::market::negotiator::{AcceptAllNegotiator, CompositeNegotiator};
 use crate::market::ProviderMarket;
 use crate::provider_agent::AgentNegotiatorsConfig;
 
+/// Configuration for Demand Validation Negotiator.
+#[derive(StructOpt, Clone, Debug)]
+pub struct DemandValidationNegotiatorConfig {
+    #[structopt(long, default_value = "/golem/com/payment/chosen-platform")]
+    pub required_fields: Vec<String>,
+}
+
 /// Configuration for LimitAgreements Negotiator.
 #[derive(StructOpt, Clone, Debug)]
 pub struct LimitAgreementsNegotiatorConfig {
@@ -58,6 +65,8 @@ pub struct PaymentTimeoutConfig {
 /// Configuration for LimitAgreements Negotiator.
 #[derive(StructOpt, Clone, Debug)]
 pub struct CompositeNegotiatorConfig {
+    #[structopt(flatten)]
+    pub validation_config: DemandValidationNegotiatorConfig,
     #[structopt(flatten)]
     pub limit_agreements_config: LimitAgreementsNegotiatorConfig,
     #[structopt(flatten)]

core/payment/src/api/allocations.rs

@@ -23,7 +23,6 @@ use crate::accounts::{init_account, Account};
 use crate::dao::*;
 use crate::error::{DbError, Error};
 use crate::utils::response;
-use crate::DEFAULT_PAYMENT_PLATFORM;
 
 pub fn register_endpoints(scope: Scope) -> Scope {
     scope
@@ -46,10 +45,11 @@ async fn create_allocation(
     // TODO: Handle deposits & timeouts
     let allocation = body.into_inner();
     let node_id = id.identity;
-    let payment_platform = allocation
-        .payment_platform
-        .clone()
-        .unwrap_or_else(|| DEFAULT_PAYMENT_PLATFORM.to_string());
+    let payment_platform = match &allocation.payment_platform {
+        Some(platform) => platform.clone(),
+        None => return response::bad_request(&"payment platform must be provided"),
+    };
+
     let address = allocation
         .address
         .clone()

core/payment/src/dao/agreement.rs

@@ -198,7 +198,7 @@ impl<'a> AgreementDao<'a> {
                 return Ok(());
             }
 
-            let agreement = WriteObj::new(agreement, role);
+            let agreement = WriteObj::try_new(agreement, role).map_err(DbError::Query)?;
             diesel::insert_into(dsl::pay_agreement)
                 .values(agreement)
                 .execute(conn)?;

core/payment/src/lib.rs

@@ -28,8 +28,6 @@ pub mod migrations {
     struct _Dummy;
 }
 
-pub const DEFAULT_PAYMENT_PLATFORM: &str = "erc20-rinkeby-tglm";
-
 pub use ya_core_model::payment::local::DEFAULT_PAYMENT_DRIVER;
 
 lazy_static::lazy_static! {

core/payment/src/models/agreement.rs

@@ -1,5 +1,4 @@
 use crate::schema::pay_agreement;
-use crate::DEFAULT_PAYMENT_PLATFORM;
 use serde_json::Value;
 use ya_agreement_utils::agreement::{expand, TypedPointer};
 use ya_client_model::market::Agreement;
@@ -25,7 +24,7 @@ pub struct WriteObj {
 }
 
 impl WriteObj {
-    pub fn new(agreement: Agreement, role: Role) -> Self {
+    pub fn try_new(agreement: Agreement, role: Role) -> Result<Self, String> {
         let provider_id = *agreement.provider_id();
         let requestor_id = *agreement.requestor_id();
         let (owner_id, peer_id) = match &role {
@@ -39,8 +38,8 @@ impl WriteObj {
         let payment_platform = demand_properties
             .pointer("/golem/com/payment/chosen-platform")
             .as_typed(Value::as_str)
-            .unwrap_or(DEFAULT_PAYMENT_PLATFORM)
-            .to_owned();
+            .map(ToOwned::to_owned)
+            .map_err(|_| "Missing golem.com.payment.chosen-platform".to_string())?;
         let payee_addr = offer_properties
             .pointer(format!("/golem/com/payment/platform/{}/address", payment_platform).as_str())
             .as_typed(Value::as_str)
@@ -52,7 +51,7 @@ impl WriteObj {
             .map(ToOwned::to_owned)
             .unwrap_or_else(|_| requestor_id.to_string().to_lowercase());
 
-        Self {
+        Ok(Self {
             id: agreement.agreement_id,
             owner_id,
             role,
@@ -65,8 +64,66 @@ impl WriteObj {
             total_amount_scheduled: Default::default(),
             total_amount_paid: Default::default(),
             app_session_id: agreement.app_session_id,
-        }
+        })
     }
 }
 
 pub type ReadObj = WriteObj;
+
+#[cfg(test)]
+mod tests {
+    use crate::models::agreement::WriteObj;
+    use chrono::{Duration, Utc};
+    use serde_json::json;
+    use std::ops::Add;
+    use ya_client_model::market::agreement::State;
+    use ya_client_model::market::{Agreement, Demand, Offer};
+    use ya_persistence::types::Role;
+
+    fn mock_agreement_with_demand_properties(properties: serde_json::Value) -> Agreement {
+        let demand = Demand::new(
+            properties,
+            "()".to_string(),
+            "demand_id".to_string(),
+            Default::default(),
+            Default::default(),
+        );
+
+        let offer = Offer::new(
+            json!({}),
+            "()".to_string(),
+            "offer_id".to_string(),
+            Default::default(),
+            Default::default(),
+        );
+
+        Agreement::new(
+            "agreement_id".to_string(),
+            demand,
+            offer,
+            Utc::now().add(Duration::days(1)),
+            State::Proposal,
+            Utc::now(),
+        )
+    }
+
+    #[test]
+    fn cannot_create_agreement_without_chosen_platform() {
+        let agreement = mock_agreement_with_demand_properties(json!({}));
+        let role: Role = Role::Provider;
+
+        let result = WriteObj::try_new(agreement, role);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn create_agreement_with_valid_chosen_platform() {
+        let agreement = mock_agreement_with_demand_properties(json!({
+            "golem.com.payment.chosen-platform": "test-network"
+        }));
+        let role: Role = Role::Provider;
+
+        let result = WriteObj::try_new(agreement, role);
+        assert_eq!(result.unwrap().payment_platform, "test-network");
+    }
+}

goth_tests/helpers/negotiation.py

@@ -36,6 +36,7 @@ def props_from_template(self, task_package: Optional[str]) -> "DemandBuilder":
             "golem.srv.comp.expiration": int(
                 (datetime.now() + timedelta(minutes=10)).timestamp() * 1000
             ),
+            "golem.com.payment.chosen-platform": self._requestor.payment_config.platform_string,
         }
 
         if task_package is not None: