Do not broadcast offers if the agreements is open (approved)

scx1332 · scx1332 · commit 8bd2e2ebbd5c · 2025-12-18T19:18:00.000+01:00
diff --git a/core/market/src/db/model.rs b/core/market/src/db/model.rs
@@ -1,4 +1,4 @@
-mod agreement;
+pub(crate) mod agreement;
 mod agreement_events;
 mod demand;
 mod negotiation_events;
diff --git a/core/market/src/identity.rs b/core/market/src/identity.rs
@@ -1,9 +1,6 @@
-use anyhow::anyhow;
-use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 
-use arkiv_sdk::{keccak256, signers::TransactionSigner, Address, Signature};
 use ya_client::model::NodeId;
 use ya_core_model::{
     bus::GsbBindPoints,
@@ -133,43 +130,3 @@ impl IdentityGSB {
         })
     }
 }
-
-pub struct YagnaIdSigner {
-    pub identity_api: Arc<dyn IdentityApi>,
-    pub node_id: NodeId,
-}
-
-#[async_trait]
-impl TransactionSigner for YagnaIdSigner {
-    fn address(&self) -> Address {
-        Address::from(&self.node_id.into_array())
-    }
-
-    async fn sign(&self, data: &[u8]) -> anyhow::Result<Signature> {
-        let hash = keccak256(data);
-        let identity_api = self.identity_api.clone();
-        let node_id = self.node_id;
-
-        Ok(tokio::task::spawn_local(async move {
-            let mut sig_bytes = identity_api.sign(&node_id, hash.as_ref()).await?;
-            let v = sig_bytes[0];
-            sig_bytes.append(&mut vec![v]);
-
-            anyhow::Ok(
-                Signature::from_raw(&sig_bytes[1..])
-                    .map_err(|e| anyhow!("Failed to parse signature: {e}"))?,
-            )
-        })
-        .await
-        .map_err(|e| anyhow!("Failed to sign data: {e}"))??)
-    }
-}
-
-impl YagnaIdSigner {
-    pub fn new(identity_api: Arc<dyn IdentityApi>, node_id: NodeId) -> Self {
-        Self {
-            identity_api,
-            node_id,
-        }
-    }
-}
diff --git a/core/market/src/matcher.rs b/core/market/src/matcher.rs
@@ -318,7 +318,6 @@ impl Matcher {
             .await
             .ok();
 
-
         log::info!(
             "Unsubscribed Offer: [{}] using identity: {} [{}]",
             &offer_id,
diff --git a/core/market/src/matcher/error.rs b/core/market/src/matcher/error.rs
@@ -17,6 +17,8 @@ pub enum DemandError {
     NotFound(SubscriptionId),
     #[error(transparent)]
     JsonObjectExpected(#[from] serde_json::error::Error),
+    #[error("{0}")]
+    Other(String),
 }
 
 #[derive(thiserror::Error, Debug)]
diff --git a/core/market/src/matcher/resolver.rs b/core/market/src/matcher/resolver.rs
@@ -1,5 +1,5 @@
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
-
+use tokio::time::Instant;
 use ya_market_resolver::{match_demand_offer, Match};
 
 use super::{error::ResolverError, RawProposal, SubscriptionStore};
@@ -84,12 +84,57 @@ impl Resolver {
             }
             Subscription::Demand(id) => {
                 let demand = self.store.get_demand(id).await?;
-                self.store
+
+                let perf_time = Instant::now();
+
+                let current_offers = self
+                    .store
                     .get_offers_before(demand.insertion_ts.unwrap())
                     .await?
                     .into_iter()
-                    .filter(|offer| matches(offer, &demand))
-                    .for_each(|offer| self.emit_proposal(offer, demand.clone()));
+                    .filter(|offer| matches(offer, &demand));
+                let agreements = match self.store.get_approved_agreements().await {
+                    Ok(agreements) => agreements,
+                    Err(e) => {
+                        return Err(ResolverError::Demand(e));
+                    }
+                };
+
+                let mut proposals_to_emit = Vec::new();
+                for offer in current_offers {
+                    //@todo check if offer is not used in open agreement
+                    if agreements.iter().any(|a| a.provider_id == offer.node_id) {
+                        log::info!(
+                            "Skipping offer from provider [{}] as it is used in approved agreement",
+                            offer.node_id
+                        );
+                        continue;
+                    }
+                    proposals_to_emit.push(offer);
+                }
+                let elapsed = perf_time.elapsed().as_secs_f64();
+
+                if elapsed > 0.1 {
+                    log::warn!(
+                        "Emitting {} proposals, preparation took {:.2} ms",
+                        proposals_to_emit.len(),
+                        elapsed * 1000.0
+                    );
+                } else {
+                    log::info!(
+                        "Emitting {} proposals, preparation took {:.2} ms",
+                        proposals_to_emit.len(),
+                        elapsed * 1000.0
+                    );
+                }
+                for offer in proposals_to_emit {
+                    log::info!(
+                        "Emitting proposal for Demand [{}] and Offer [{}]",
+                        demand.id,
+                        offer.id
+                    );
+                    self.emit_proposal(offer, demand.clone());
+                }
             }
         }
         Ok(())
@@ -98,6 +143,7 @@ impl Resolver {
     pub fn emit_proposal(&self, offer: Offer, demand: Demand) {
         let offer_id = offer.id.clone();
         let demand_id = demand.id.clone();
+        log::info!("PROPOSAL: Offer [{}] <-> Demand [{}]", offer_id, demand_id);
         if let Err(e) = self.proposal_tx.send(RawProposal { offer, demand }) {
             log::warn!(
                 "Emitting proposal for Offer [{}] and Demand [{}] error: {}",
diff --git a/core/market/src/matcher/store.rs b/core/market/src/matcher/store.rs
@@ -3,7 +3,6 @@ use chrono::{NaiveDateTime, Utc};
 use std::collections::HashSet;
 use std::sync::Arc;
 use std::time::Duration;
-
 use ya_client::model::market::{Demand as ClientDemand, NewDemand, Offer as ClientOffer};
 use ya_client::model::NodeId;
 use ya_service_api_web::middleware::Identity;
@@ -19,6 +18,7 @@ use crate::matcher::error::{
 use crate::matcher::sync::{OfferPlaceholderGuard, OfferSync};
 use crate::negotiation::ScannerSet;
 use crate::protocol::discovery::message::{QueryOffers, QueryOffersResult};
+use crate::testing::AgreementState;
 
 #[derive(Clone)]
 pub struct SubscriptionStore {
@@ -318,6 +318,22 @@ impl SubscriptionStore {
         }
     }
 
+    pub async fn get_approved_agreements(
+        &self,
+    ) -> Result<Vec<crate::db::model::agreement::Agreement>, DemandError> {
+        match self
+            .db
+            .as_dao::<AgreementDao>()
+            .list(None, Some(AgreementState::Approved), None, None, None)
+            .await
+        {
+            Ok(agreements) => Ok(agreements),
+            Err(e) => Err(DemandError::Other(format!(
+                "Failed to get Agreements from database {e}"
+            ))),
+        }
+    }
+
     pub async fn get_client_demands(
         &self,
         node_id: Option<NodeId>,
diff --git a/core/market/src/protocol/discovery.rs b/core/market/src/protocol/discovery.rs
@@ -1,42 +1,23 @@
 use anyhow::Result;
 use chrono::{DateTime, Utc};
-use offer::GolemBaseOffer;
 use std::collections::HashSet;
 use std::env;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::task::JoinHandle;
-use ya_service_bus::timeout::IntoTimeoutFuture;
 
 use ya_client::model::NodeId;
 use ya_core_model::bus::GsbBindPoints;
-use ya_core_model::identity::event::IdentityEvent;
-use ya_core_model::identity::Error;
-use ya_core_model::market::{local, GetGolemBaseOffer, GetGolemBaseOfferResponse};
-use ya_core_model::market::{
-    FundGolemBase, GetGolemBaseBalance, GolemBaseCommand, RpcMessageError,
-};
-use ya_service_bus::typed as bus;
 
 use super::callback::HandlerSlot;
 use crate::config::DiscoveryConfig;
-use crate::db::model::{Offer as ModelOffer, SubscriptionId};
-use crate::identity::{IdentityApi, IdentityError, YagnaIdSigner};
+use crate::db::model::Offer as ModelOffer;
+use crate::identity::IdentityApi;
 use crate::protocol::discovery::error::*;
 use crate::protocol::discovery::message::*;
-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 rand::{thread_rng, Rng};
 
 const ARKIV_CALLER: &str = "Arkiv";
 
-// TODO: Get this value from node configuration
-const BLOCK_TIME_SECONDS: i64 = 2;
-
 pub mod builder;
 pub mod error;
 pub mod faucet;
@@ -61,24 +42,11 @@ pub struct DiscoveryImpl {
     //arkiv: ArkivClient,
     offer_handlers: OfferHandlers,
     config: DiscoveryConfig,
-    identities: Mutex<HashSet<NodeId>>,
+    _identities: Mutex<HashSet<NodeId>>,
     websocket_task: Mutex<Option<JoinHandle<()>>>,
 }
 
 impl Discovery {
-
-    /// Checks if an offer belongs to us based on metadata and entity_id
-    fn _is_own_offer(&self, metadata: &SearchResult) -> bool {
-        let Some(owner) = metadata.owner.as_ref() else {
-            log::warn!("[Programming error] Entity metadata should contain owner!");
-            return false;
-        };
-
-        let identities = self.inner.identities.lock().unwrap();
-        let owner_bytes = NodeId::from(owner.as_slice());
-        identities.contains(&owner_bytes)
-    }
-
     async fn offers_events_loop(&self, _starting_block: u64) -> anyhow::Result<()> {
         let default_identity = self
             .inner
@@ -246,10 +214,9 @@ impl Discovery {
         Ok(())
     }
 
-
     /// Function doesn't bind any GSB handlers.
     /// It's only used to sync with GolemBase node and initialize Discovery struct state.
-    pub async fn bind_gsb(&self, gsb: GsbBindPoints) -> Result<(), DiscoveryInitError> {
+    pub async fn bind_gsb(&self, _gsb: GsbBindPoints) -> Result<(), DiscoveryInitError> {
         log::info!("Arkiv Configuration:");
         log::info!("  Network: {:?}", self.inner.config.get_network_type());
         log::info!("  RPC URL: {}", self.inner.config.get_rpc_url());
@@ -295,10 +262,6 @@ impl Discovery {
         Ok(())
     }
 
-
-
-
-
     pub(crate) async fn get_last_bcast_ts(&self) -> DateTime<Utc> {
         Utc::now()
     }
diff --git a/core/market/src/protocol/discovery/builder.rs b/core/market/src/protocol/discovery/builder.rs
@@ -1,12 +1,10 @@
 use std::any::{Any, TypeId};
 use std::collections::HashMap;
 use std::collections::HashSet;
-use std::ops::Div;
 use std::sync::Arc;
 
 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};
@@ -76,14 +74,13 @@ impl DiscoveryBuilder {
         let config = self.config.clone().ok_or_else(|| {
             DiscoveryInitError::BuilderIncomplete("Configuration is required".to_string())
         })?;
-        
 
         let discovery = Discovery {
             inner: Arc::new(DiscoveryImpl {
                 identity: self.get_data(),
                 offer_handlers,
                 config: config.clone(),
-                identities: std::sync::Mutex::new(HashSet::new()),
+                _identities: std::sync::Mutex::new(HashSet::new()),
                 websocket_task: std::sync::Mutex::new(None),
             }),
         };

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-mod agreement;`
	`1`	`+pub(crate) mod agreement;`
`2`	`2`	`mod agreement_events;`
`3`	`3`	`mod demand;`
`4`	`4`	`mod negotiation_events;`
Original file line number	Diff line number	Diff line change
`@@ -17,6 +17,8 @@ pub enum DemandError {`
`17`	`17`	`NotFound(SubscriptionId),`
`18`	`18`	`#[error(transparent)]`
`19`	`19`	`JsonObjectExpected(#[from] serde_json::error::Error),`
	`20`	`+ #[error("{0}")]`
	`21`	`+ Other(String),`
`20`	`22`	`}`
`21`	`23`
`22`	`24`	`#[derive(thiserror::Error, Debug)]`