Integration with offer server

Author: scx1332

Cargo.lock

@@ -34,7 +34,6 @@ actix-web = "4"
 alloy = { version = "1.0", features = ["full"] }
 anyhow = "1.0"
 async-trait = { version = "0.1.33" }
-backtrace = "0.3.50"
 bincode = "1.3.3"
 bigdecimal = "0.4"
 chrono = { version = "0.4", features = ["serde"] }
@@ -45,7 +44,6 @@ diesel_migrations = "1.4"
 digest = "0.8.1"
 env_logger = { version = "0.7" }
 futures = "0.3"
-glob = "0.3.3"
 hex = "0.4"
 humantime = "2"
 lazy_static = "1.4"

core/market/Cargo.toml

@@ -1,13 +1,14 @@
 use chrono::{NaiveDateTime, TimeZone, Utc};
-
+use serde::{Deserialize, Serialize};
 use ya_client::model::{market::Demand as ClientDemand, ErrorMessage, NodeId};
 use ya_service_api_web::middleware::Identity;
 
 use super::SubscriptionId;
 use crate::db::schema::market_demand;
 use ya_client::model::market::NewDemand;
 
-#[derive(Clone, Debug, Identifiable, Insertable, Queryable)]
+#[derive(Clone, Debug, Identifiable, Insertable, Queryable, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
 #[table_name = "market_demand"]
 pub struct Demand {
     pub id: SubscriptionId,

core/market/src/db/model/demand.rs

@@ -9,7 +9,6 @@ use ya_client::model::market::{
     Agreement, AgreementListEntry, AgreementOperationEvent as ClientAgreementEvent, Demand,
     NewDemand, NewOffer, Offer, Reason, Role,
 };
-
 use ya_core_model::bus::GsbBindPoints;
 use ya_core_model::{self as model};
 use ya_service_api_interfaces::{Provider, Service};

core/market/src/market.rs

@@ -347,6 +347,34 @@ impl Matcher {
     ) -> Result<Demand, MatcherError> {
         let demand = self.store.create_demand(id, demand).await?;
 
+        let matcher = env::var("YAGNA_MARKET_MATCHER_URL");
+        if let Ok(matcher) = matcher {
+            let client = reqwest::Client::new();
+            let res = client
+                .post(&(matcher + "/requestor/demand/new"))
+                .header("Content-Type", "application/json")
+                .body(serde_json::to_string(&demand).unwrap())
+                .send()
+                .await;
+            match res {
+                Ok(response) => {
+                    if response.status().is_success() {
+                        log::info!("Successfully notified matcher about new demand");
+                    } else {
+                        log::error!(
+                            "Failed to notify matcher about new demand: HTTP {}",
+                            response.status()
+                        );
+                    }
+                }
+                Err(e) => {
+                    log::error!("Failed to notify matcher about new demand: {}", e);
+                }
+            }
+        } else {
+            log::warn!("YAGNA_MARKET_MATCHER_URL not set, skipping demand notification");
+        }
+
         // Track demand expiration
         self.expiration_tracker
             .send(TrackDeadline {
@@ -394,6 +422,34 @@ impl Matcher {
             .await
             .ok();
 
+        let matcher = env::var("YAGNA_MARKET_MATCHER_URL");
+        if let Ok(matcher) = matcher {
+            let client = reqwest::Client::new();
+            let res = client
+                .post(&(matcher + "/requestor/demand/cancel"))
+                .header("Content-Type", "application/json")
+                .body("{\"demandId\":\"".to_string() + &demand_id.to_string() + "\"}")
+                .send()
+                .await;
+            match res {
+                Ok(response) => {
+                    if response.status().is_success() {
+                        log::info!("Successfully notified matcher about demand cancellation");
+                    } else {
+                        log::error!(
+                            "Failed to notify matcher about demand cancellation: HTTP {}",
+                            response.status()
+                        );
+                    }
+                }
+                Err(e) => {
+                    log::error!("Failed to notify matcher about demand cancellation: {}", e);
+                }
+            }
+        } else {
+            log::warn!("YAGNA_MARKET_MATCHER_URL not set, skipping demand notification");
+        }
+
         // Check if this was the last demand - if so, stop listening for offers
         self.maybe_stop_listening().await;
 

core/market/src/matcher.rs

@@ -2,9 +2,7 @@ use anyhow::Result;
 use chrono::{DateTime, Utc};
 use offer::GolemBaseOffer;
 use std::collections::HashSet;
-use std::fs;
-use std::path::Path;
-use std::str::FromStr;
+use std::env;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::task::JoinHandle;
@@ -26,16 +24,13 @@ use crate::db::model::{Offer as ModelOffer, SubscriptionId};
 use crate::identity::{IdentityApi, IdentityError, YagnaIdSigner};
 use crate::protocol::discovery::error::*;
 use crate::protocol::discovery::message::*;
-use crate::testing::MatcherError;
 use arkiv_sdk::client::ArkivClient;
 use arkiv_sdk::entity::Create;
 use arkiv_sdk::events::Event;
 use arkiv_sdk::rpc::{QueryOptions, SearchResult};
 use arkiv_sdk::signers::TransactionSigner;
 use arkiv_sdk::{Address, Hash};
-use glob::glob;
 use rand::{thread_rng, Rng};
-use serde_json::Value;
 
 const ARKIV_CALLER: &str = "Arkiv";
 
@@ -311,74 +306,87 @@ impl Discovery {
         Ok(())
     }
 
-    async fn read_offer_from_file(path: &Path) -> anyhow::Result<ModelOffer> {
-        // Read the file
-        let data = fs::read_to_string(path)?;
+    async fn offers_events_loop(&self, _starting_block: u64) -> anyhow::Result<()> {
+        let default_identity = self
+            .inner
+            .identity
+            .default_identity()
+            .await
+            .expect("Failed to get default identity");
 
-        let val: Value = serde_json::from_str(&data)?;
+        loop {
+            // Wait for either Ctrl+C or timeout to continue loop
+            tokio::select! {
+                _ = tokio::signal::ctrl_c() => {
+                    log::info!("Received Ctrl+C, shutting down offers events loop");
+                    break;
+                }
+                _ = tokio::time::sleep(Duration::from_secs(10)) => {}
+            }
 
-        let hash = Hash::from_str(
-            val.as_object()
-                .ok_or_else(|| {
-                    MatcherError::GolemBaseOfferError(
-                        "Offer serialized to non-object JSON".to_string(),
+            let matcher = env::var("YAGNA_MARKET_MATCHER_URL");
+            if let Ok(matcher) = matcher {
+                let client = reqwest::Client::new();
+                let res = client
+                    .post(&(matcher + "/requestor/demand/take-from-queue"))
+                    .header("Content-Type", "application/json")
+                    .body(
+                        "{
+                            \"demandId\""
+                            .to_string()
+                            + ": \""
+                            + &default_identity.to_string()
+                            + "\"}",
                     )
-                })?
-                .get("id")
-                .ok_or_else(|| {
-                    MatcherError::GolemBaseOfferError("Offer JSON missing 'id' field".to_string())
-                })?
-                .as_str()
-                .ok_or_else(|| {
-                    MatcherError::GolemBaseOfferError(
-                        "Offer 'id' field is not a string".to_string(),
-                    )
-                })?,
-        )
-        .map_err(|e| {
-            MatcherError::GolemBaseOfferError(format!("Failed to hash from id field: {}", e))
-        })?;
-
-        let offer: GolemBaseOffer = serde_json::from_str(&data)?;
-
-        let offer = offer.into_model_offer(hash)?;
-
-        Ok(offer)
-    }
-
-    async fn offers_events_loop(&self, _starting_block: u64) -> anyhow::Result<()> {
-        loop {
-            //read file from offer.json
-            if let Ok(paths) = glob("incoming-offer-*.json") {
-                #[allow(clippy::manual_flatten)]
-                for entry in paths {
-                    if let Ok(path) = entry {
-                        match Self::read_offer_from_file(&path).await {
-                            Ok(offer) => {
-                                log::info!("Registering incoming offer from file: {:?}", path);
-                                self.register_incoming_offers(vec![offer]).await?;
+                    .send()
+                    .await;
+                match res {
+                    Ok(response) => {
+                        if response.status().is_success() {
+                            let data = response.text().await.unwrap_or_default();
+
+                            let offer = serde_json::from_str::<ModelOffer>(&data);
+                            match offer {
+                                Ok(offer) => {
+                                    log::info!(
+                                        "Registering incoming offer from matcher: {:?}",
+                                        offer.id
+                                    );
+                                    self.register_incoming_offers(vec![offer]).await?;
+                                }
+                                Err(e) => {
+                                    log::error!(
+                                        "Failed to parse offer from matcher response: {}",
+                                        e
+                                    );
+                                }
                             }
-                            Err(e) => {
+                        } else if response.status() == reqwest::StatusCode::NOT_FOUND {
+                            let response_text = response.text().await.unwrap_or_default();
+                            if response_text.is_empty() {
                                 log::error!(
-                                    "Failed to read and register offer from file {:?}: {}",
-                                    path,
-                                    e
+                                    "Failed to take offer due to other error (no body): {}",
+                                    default_identity
+                                );
+                            } else {
+                                log::info!(
+                                    "No offers available in matcher queue for demand {}",
+                                    response_text
                                 );
                             }
+                        } else {
+                            log::error!(
+                                "Failed to take offer, other status: {}",
+                                response.status()
+                            );
                         }
-                        fs::remove_file(path).map_err(|e| {
-                            anyhow::anyhow!("Failed to remove invalid offer file: {}", e)
-                        })?;
+                    }
+                    Err(e) => {
+                        log::error!("Failed to take offer due to other error: {}", e);
                     }
                 }
-            }
-            // Wait for either Ctrl+C or timeout to continue loop
-            tokio::select! {
-                _ = tokio::signal::ctrl_c() => {
-                    log::info!("Received Ctrl+C, shutting down offers events loop");
-                    break;
-                }
-                _ = tokio::time::sleep(Duration::from_secs(1)) => {}
+            } else {
+                log::warn!("YAGNA_MARKET_MATCHER_URL not set, skipping demand notification");
             }
         }
         Ok(())
@@ -402,7 +410,7 @@ impl Discovery {
             DiscoveryInitError::GolemBaseInitFailed(format!("Failed to register offers: {}", e))
         })?;
 
-        let handle = tokio::spawn(async move {
+        let handle = tokio::task::spawn_local(async move {
             discovery
                 .offers_events_loop(starting_block)
                 .await

core/market/src/protocol/discovery.rs

@@ -4,10 +4,9 @@ use std::collections::HashSet;
 use std::ops::Div;
 use std::sync::Arc;
 
-use arkiv_sdk::client::{ArkivClient, TransactionConfig};
-
 use crate::protocol::callback::{CallbackFuture, OutputFuture};
 use crate::protocol::callback::{CallbackHandler, CallbackMessage, HandlerSlot};
+use arkiv_sdk::client::{ArkivClient, TransactionConfig};
 
 use super::error::DiscoveryInitError;
 use super::{Discovery, DiscoveryImpl};