payment: finish deposit release

Cargo.toml

@@ -232,8 +232,8 @@ members = [
 # diesel 1.4.* supports up to 0.23.0, but sqlx 0.5.9 requires 0.22.0
 # sqlx 0.5.10 need 0.23.2, so 0.5.9 is last version possible
 derive_more = "0.99.11"
-erc20_payment_lib = { git = "https://github.com/golemfactory/erc20_payment_lib", rev = "d82a4e2031ec2f434e301190480b5522f371848b" }
-erc20_processor = { git = "https://github.com/golemfactory/erc20_payment_lib", rev = "d82a4e2031ec2f434e301190480b5522f371848b" }
+erc20_payment_lib = { git = "https://github.com/golemfactory/erc20_payment_lib", rev = "3404b727d79fbacd87a8c708f6b7dd5a2745630f" }
+erc20_processor = { git = "https://github.com/golemfactory/erc20_payment_lib", rev = "3404b727d79fbacd87a8c708f6b7dd5a2745630f" }
 #erc20_payment_lib = { path = "../../payments/erc20_payment_lib/crates/erc20_payment_lib" }
 #erc20_processor = { path = "../../payments/erc20_payment_lib" }
 #erc20_payment_lib = { version = "=0.4.1" }
@@ -280,9 +280,9 @@ ya-service-api-interfaces = { path = "core/serv-api/interfaces" }
 ya-service-api-web = { path = "core/serv-api/web" }
 
 ## CLIENT
-ya-client = { git = "https://github.com/golemfactory/ya-client.git", rev = "e58208bdd521afbb626021d94b0b91a7d17d4e9d" }
+ya-client = { git = "https://github.com/golemfactory/ya-client.git", rev = "2539fcefd02b12e0078dc1ddcc9bcc0e841e685f" }
 #ya-client = { path = "../ya-client" }
-ya-client-model = { git = "https://github.com/golemfactory/ya-client.git", rev = "e58208bdd521afbb626021d94b0b91a7d17d4e9d" }
+ya-client-model = { git = "https://github.com/golemfactory/ya-client.git", rev = "2539fcefd02b12e0078dc1ddcc9bcc0e841e685f" }
 #ya-client-model = "0.7"
 golem-certificate = { git = "https://github.com/golemfactory/golem-certificate.git", rev = "f2d7514c18fc066e9cfb796090b90f5b27cfe1c6" }
 

core/model/src/driver.rs

@@ -590,6 +590,22 @@ pub enum DriverStatusError {
     NetworkNotFound(String),
 }
 
+// ************************* DEPOSIT *************************
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct DriverReleaseDeposit {
+    pub platform: String,
+    pub from: String,
+    pub deposit_contract: String,
+    pub deposit_id: String,
+}
+
+impl RpcMessage for DriverReleaseDeposit {
+    const ID: &'static str = "DriverReleaseDeposit";
+    type Item = ();
+    type Error = GenericError;
+}
+
 // ************************* SHUT DOWN *************************
 
 #[derive(Clone, Debug, Serialize, Deserialize)]

core/model/src/payment.rs

@@ -453,22 +453,16 @@ pub mod local {
 
     #[derive(Clone, Debug, Serialize, Deserialize)]
     pub struct ReleaseDeposit {
-        pub signer: NodeId,
+        pub platform: String,
+        pub from: String,
+        pub deposit_contract: String,
         pub deposit_id: String,
     }
 
     impl RpcMessage for ReleaseDeposit {
         const ID: &'static str = "ReleaseDeposit";
-        type Item = bool;
-        type Error = ReleaseDepositError;
-    }
-
-    #[derive(Clone, Debug, Serialize, Deserialize, thiserror::Error)]
-    pub enum ReleaseDepositError {
-        #[error("Account not registered")]
-        AccountNotRegistered,
-        #[error("Error while releasing allocation: {0}")]
-        Other(String),
+        type Item = ();
+        type Error = GenericError;
     }
 
     #[derive(Clone, Debug, Serialize, Deserialize)]

core/payment-driver/base/src/bus.rs

@@ -83,6 +83,9 @@ pub async fn bind_service<Driver: PaymentDriver + 'static>(
         .bind_with_processor(
             move |_, dr, c, m| async move { dr.status( c, m).await }
         )
+        .bind_with_processor(
+            move |_, dr, c, m| async move { dr.release_deposit( c, m).await }
+        )
         .bind_with_processor(
             move |_, dr, c, m| async move { dr.shut_down( c, m).await }
         );

core/payment-driver/base/src/driver.rs

@@ -89,6 +89,13 @@ pub trait PaymentDriver {
         msg: ValidateAllocation,
     ) -> Result<bool, GenericError>;
 
+    async fn release_deposit(
+        &self,
+
+        caller: String,
+        msg: DriverReleaseDeposit,
+    ) -> Result<(), GenericError>;
+
     async fn sign_payment(
         &self,
 

core/payment-driver/erc20/src/driver.rs

@@ -271,12 +271,6 @@ impl Erc20Driver {
                         network,
                     })
                 }
-                LibStatusProperty::InvalidChainId { chain_id, .. } => {
-                    Some(DriverStatusProperty::InvalidChainId {
-                        driver: DRIVER_NAME.into(),
-                        chain_id,
-                    })
-                }
             })
             .collect())
     }
@@ -994,6 +988,46 @@ impl PaymentDriver for Erc20Driver {
         }
     }
 
+    async fn release_deposit(
+        &self,
+        _caller: String,
+        msg: DriverReleaseDeposit,
+    ) -> Result<(), GenericError> {
+        let network = &msg
+            .platform
+            .split('-')
+            .nth(1)
+            .ok_or(GenericError::new(format!(
+                "Malformed platform string: {}",
+                msg.platform
+            )))
+            .unwrap();
+
+        self.payment_runtime
+            .close_deposit(
+                network,
+                H160::from_str(&msg.from).map_err(|e| {
+                    GenericError::new(format!("`{}` address parsing error: {}", msg.from, e))
+                })?,
+                H160::from_str(&msg.deposit_contract).map_err(|e| {
+                    GenericError::new(format!(
+                        "`{}` address parsing error: {}",
+                        msg.deposit_contract, e
+                    ))
+                })?,
+                U256::from_str(&msg.deposit_id).map_err(|e| {
+                    GenericError::new(format!(
+                        "`{}` deposit id parsing error: {}",
+                        msg.deposit_id, e
+                    ))
+                })?,
+            )
+            .await
+            .map_err(|err| GenericError::new(format!("Error releasing deposit: {}", err)))?;
+
+        Ok(())
+    }
+
     async fn status(
         &self,
         _caller: String,

core/payment/src/api/allocations.rs

@@ -523,12 +523,14 @@ async fn release_allocation(
     let dao = db.as_dao::<AllocationDao>();
 
     match dao.release(allocation_id.clone(), node_id).await {
-        Ok(AllocationReleaseStatus::Released { deposit }) => {
+        Ok(AllocationReleaseStatus::Released { deposit, platform }) => {
             if let Some(deposit) = deposit {
                 let release_result = bus::service(LOCAL_SERVICE)
                     .send(ReleaseDeposit {
-                        signer: id.identity,
+                        from: id.identity.to_string(),
                         deposit_id: deposit.id,
+                        deposit_contract: deposit.contract,
+                        platform,
                     })
                     .await;
                 match release_result {
@@ -640,7 +642,7 @@ pub async fn forced_release_allocation(
         .release(allocation_id.clone(), node_id)
         .await
     {
-        Ok(AllocationReleaseStatus::Released { deposit: _ }) => {
+        Ok(AllocationReleaseStatus::Released { deposit, platform }) => {
             log::info!("Allocation {} released.", allocation_id);
         }
         Err(e) => {

core/payment/src/dao/allocation.rs

@@ -179,7 +179,7 @@ impl<'c> AllocationDao<'c> {
                 .first(conn)
                 .optional()?;
 
-            let deposit = match allocation {
+            let (deposit, platform) = match allocation {
                 Some(allocation) => {
                     if let Some(owner_id) = owner_id {
                         if owner_id != allocation.owner_id {
@@ -191,7 +191,9 @@ impl<'c> AllocationDao<'c> {
                         return Ok(AllocationReleaseStatus::Gone);
                     }
 
-                    Allocation::from(allocation).deposit
+                    let allocation = Allocation::from(allocation);
+
+                    (allocation.deposit, allocation.payment_platform)
                 }
                 None => return Ok(AllocationReleaseStatus::NotFound),
             };
@@ -203,7 +205,7 @@ impl<'c> AllocationDao<'c> {
                 .execute(conn)?;
 
             match num_released {
-                1 => Ok(AllocationReleaseStatus::Released { deposit }),
+                1 => Ok(AllocationReleaseStatus::Released { deposit, platform }),
                 _ => Err(DbError::Query(format!(
                     "Update error occurred when releasing allocation {}",
                     allocation_id
@@ -249,5 +251,8 @@ pub enum AllocationStatus {
 pub enum AllocationReleaseStatus {
     Gone,
     NotFound,
-    Released { deposit: Option<Deposit> },
+    Released {
+        deposit: Option<Deposit>,
+        platform: String,
+    },
 }

core/payment/src/processor.rs

@@ -25,13 +25,13 @@ use ya_client_model::payment::{
     Account, ActivityPayment, AgreementPayment, DriverDetails, Network, Payment,
 };
 use ya_core_model::driver::{
-    self, driver_bus_id, AccountMode, GasDetails, GetRpcEndpointsResult, PaymentConfirmation,
-    PaymentDetails, ShutDown, ValidateAllocation,
+    self, driver_bus_id, AccountMode, DriverReleaseDeposit, GasDetails, GetRpcEndpointsResult,
+    PaymentConfirmation, PaymentDetails, ShutDown, ValidateAllocation,
 };
 use ya_core_model::payment::local::{
-    GetAccountsError, GetDriversError, NotifyPayment, RegisterAccount, RegisterAccountError,
-    RegisterDriver, RegisterDriverError, SchedulePayment, UnregisterAccount,
-    UnregisterAccountError, UnregisterDriver, UnregisterDriverError,
+    GenericError, GetAccountsError, GetDriversError, NotifyPayment, RegisterAccount,
+    RegisterAccountError, RegisterDriver, RegisterDriverError, ReleaseDeposit, SchedulePayment,
+    UnregisterAccount, UnregisterAccountError, UnregisterDriver, UnregisterDriverError,
 };
 use ya_core_model::payment::public::{SendPayment, BUS_ID};
 use ya_core_model::NodeId;
@@ -884,6 +884,29 @@ impl PaymentProcessor {
         }
     }
 
+    pub async fn release_deposit(&self, msg: ReleaseDeposit) -> Result<(), GenericError> {
+        let driver = self
+            .registry
+            .timeout_read(REGISTRY_LOCK_TIMEOUT)
+            .await
+            .map_err(GenericError::new)?
+            .driver(&msg.platform, &msg.from, AccountMode::SEND)
+            .map_err(GenericError::new)?;
+
+        driver_endpoint(&driver)
+            .send(DriverReleaseDeposit {
+                platform: msg.platform,
+                from: msg.from,
+                deposit_contract: msg.deposit_contract,
+                deposit_id: msg.deposit_id,
+            })
+            .await
+            .map_err(GenericError::new)?
+            .map_err(GenericError::new)?;
+
+        Ok(())
+    }
+
     pub async fn shut_down(
         &self,
         timeout: Duration,

core/payment/src/service.rs

@@ -85,6 +85,7 @@ mod local {
             .bind_with_processor(get_drivers)
             .bind_with_processor(payment_driver_status)
             .bind_with_processor(handle_status_change)
+            .bind_with_processor(release_deposit)
             .bind_with_processor(shut_down);
 
         // Initialize counters to 0 value. Otherwise they won't appear on metrics endpoint
@@ -706,6 +707,18 @@ mod local {
         Ok(Ack {})
     }
 
+    async fn release_deposit(
+        db: DbExecutor,
+        processor: Arc<PaymentProcessor>,
+        sender: String,
+        msg: ReleaseDeposit,
+    ) -> Result<(), GenericError> {
+        log::debug!("Schedule payment processor started");
+        let res = processor.release_deposit(msg).await;
+        log::debug!("Schedule payment processor finished");
+        res
+    }
+
     async fn shut_down(
         db: DbExecutor,
         processor: Arc<PaymentProcessor>,