diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index 101d9c9d..75011602 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -146,6 +146,20 @@ impl AgreementRegistry for RegistryProvider { .collect()) } + async fn get_active_indexing_agreements_by_indexer_ids( + &self, + indexer_ids: &[IndexerId], + ) -> RegistryResult> { + Ok(self + .inner + .get_active_indexing_agreements_by_indexer_ids(indexer_ids) + .await? + .into_iter() + .map(IndexingAgreement::try_from) + .filter_map(Result::ok) + .collect()) + } + async fn get_indexing_agreements_by_indexing_request_id( &self, request_id: &IndexingRequestId, diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index 47ba0ef7..72d6690c 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -44,6 +44,15 @@ pub trait AgreementRegistry { indexer_id: &IndexerId, ) -> RegistryResult>; + /// Get all active agreements for multiple indexers in a single query. + /// + /// Returns agreements that are in `Created` or `Accepted` status for any of the + /// provided indexer IDs. This is more efficient than querying each indexer separately. + async fn get_active_indexing_agreements_by_indexer_ids( + &self, + indexer_ids: &[IndexerId], + ) -> RegistryResult>; + /// Get all agreements by associated indexing request ID. async fn get_indexing_agreements_by_indexing_request_id( &self, diff --git a/bin/dipper-service/src/worker/handlers.rs b/bin/dipper-service/src/worker/handlers.rs index 40c855b4..7fcdf802 100644 --- a/bin/dipper-service/src/worker/handlers.rs +++ b/bin/dipper-service/src/worker/handlers.rs @@ -2,6 +2,7 @@ mod find_indexer_for_indexing_request; mod process_indexing_agreement_cancellation; mod process_indexing_request_cancellation; mod process_new_indexing_request; +mod selection_context; mod send_indexing_agreement_cancellation; mod send_indexing_agreement_proposal; diff --git a/bin/dipper-service/src/worker/handlers/find_indexer_for_indexing_request.rs b/bin/dipper-service/src/worker/handlers/find_indexer_for_indexing_request.rs index 4aea734d..779e5b52 100644 --- a/bin/dipper-service/src/worker/handlers/find_indexer_for_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/find_indexer_for_indexing_request.rs @@ -6,6 +6,7 @@ use jsonrpsee::core::Serialize; use serde::Deserialize; use thegraph_core::{DeploymentId, alloy::primitives::ChainId}; +use super::selection_context::gather_selection_context; use crate::{ config::{IndexingAgreementChainPrices, IndexingAgreementConfig}, network::NetworkProvider, @@ -97,9 +98,12 @@ where return Ok(()); } + // Gather load balancing context for IISA + let context = gather_selection_context(&ctx.registry, deployment_id, &indexers).await?; + let Some(candidate) = ctx .iisa - .select_one(*deployment_id, indexers) + .select_one(*deployment_id, indexers, &context) .await .map_err(|err| JobError::Fatal(err.into()))? else { diff --git a/bin/dipper-service/src/worker/handlers/process_new_indexing_request.rs b/bin/dipper-service/src/worker/handlers/process_new_indexing_request.rs index 0596050e..1a9f8440 100644 --- a/bin/dipper-service/src/worker/handlers/process_new_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/process_new_indexing_request.rs @@ -4,6 +4,7 @@ use dipper_core::ids::IndexingRequestId; use dipper_iisa::{CandidateSelection, Indexer as IndexerCandidate}; use thegraph_core::{DeploymentId, alloy::primitives::ChainId}; +use super::selection_context::gather_selection_context; use crate::{ config::{IndexingAgreementChainPrices, IndexingAgreementConfig}, network::NetworkProvider, @@ -57,7 +58,7 @@ where W: WorkerQueue, I: CandidateSelection, { - // Get the indexers that are not indexing the deployment amd treat it as the raw candidate list + // Get the indexers that are not indexing the deployment and treat it as the raw candidate list // and pass it to the IISA to get the final list of candidates let indexers = ctx .network @@ -76,9 +77,12 @@ where return Ok(()); } + // Gather load balancing context for IISA + let context = gather_selection_context(&ctx.registry, deployment_id, &indexers).await?; + let candidates = ctx .iisa - .select(*deployment_id, indexers, *num_candidates) + .select(*deployment_id, indexers, *num_candidates, &context) .await .map_err(|err| JobError::Fatal(err.into()))?; if candidates.is_empty() { diff --git a/bin/dipper-service/src/worker/handlers/selection_context.rs b/bin/dipper-service/src/worker/handlers/selection_context.rs new file mode 100644 index 00000000..a22e2ad8 --- /dev/null +++ b/bin/dipper-service/src/worker/handlers/selection_context.rs @@ -0,0 +1,69 @@ +//! Shared utilities for gathering IISA selection context. + +use std::collections::HashMap; + +use dipper_iisa::{Indexer as IndexerCandidate, SelectionContext}; +use thegraph_core::{DeploymentId, IndexerId}; + +use crate::{ + registry::{AgreementRegistry, IndexingAgreementStatus}, + worker::result::{JobError, JobResult}, +}; + +/// Gather load balancing context for IISA selection. +/// +/// This function queries the registry to build context about: +/// - Which indexers already have active agreements for this deployment +/// - What other deployments each candidate indexer is currently working on +pub async fn gather_selection_context( + registry: &R, + deployment_id: &DeploymentId, + candidates: &[IndexerCandidate], +) -> JobResult +where + R: AgreementRegistry, +{ + // Get indexers that already have active agreements for this deployment + let existing_indexers = registry + .get_indexing_agreements_by_deployment_id(deployment_id) + .await + .map_err(|err| JobError::Fatal(err.into()))? + .into_iter() + .filter(|a| is_active_agreement(&a.status)) + .map(|a| a.indexer.id) + .collect::>(); + + // Build pending agreements map for all candidates in a single batch query + // This tells IISA what other work each candidate is currently handling + let candidate_ids: Vec = candidates.iter().map(|c| c.id).collect(); + + let all_agreements = registry + .get_active_indexing_agreements_by_indexer_ids(&candidate_ids) + .await + .map_err(|err| JobError::Fatal(err.into()))?; + + // Group agreements by indexer ID + let mut pending_agreements: HashMap> = HashMap::new(); + for agreement in all_agreements { + pending_agreements + .entry(agreement.indexer.id) + .or_default() + .push(agreement.voucher.metadata.subgraph_deployment_id); + } + + Ok(SelectionContext { + existing_indexers, + pending_agreements, + }) +} + +/// Check if an agreement status represents an active agreement. +/// +/// Active agreements are those that are either pending acceptance (Created) +/// or currently in effect (Accepted). +fn is_active_agreement(status: &IndexingAgreementStatus) -> bool { + matches!( + status, + IndexingAgreementStatus::Created | IndexingAgreementStatus::Accepted { .. } + ) +} diff --git a/dipper-iisa/src/api.rs b/dipper-iisa/src/api.rs index 138eaf45..e38f9c86 100644 --- a/dipper-iisa/src/api.rs +++ b/dipper-iisa/src/api.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use async_trait::async_trait; use thegraph_core::{DeploymentId, IndexerId}; use url::Url; @@ -11,6 +13,24 @@ pub struct Indexer { pub url: Url, } +/// Context for load balancing during indexer selection. +/// +/// This context provides the IISA with information about current system state, +/// enabling intelligent load balancing decisions rather than naive selection. +#[derive(Debug, Clone, Default)] +pub struct SelectionContext { + /// Indexer IDs that already have active agreements for this deployment. + /// + /// Used to avoid selecting indexers that are already working on the same deployment. + pub existing_indexers: Vec, + + /// For each indexer, the deployments they have pending/active agreements for. + /// + /// Used to balance load across indexers by considering their current workload. + /// Key: Indexer ID, Value: List of deployment IDs they are working on. + pub pending_agreements: HashMap>, +} + /// The `SelectionError` enum represents the errors that can occur during the candidate selection /// process. #[derive(Debug, thiserror::Error)] @@ -31,17 +51,30 @@ pub enum SelectionError { #[async_trait] pub trait CandidateSelection { /// Select one indexer from the given list of candidates. + /// + /// # Arguments + /// * `deployment_id` - The deployment to select an indexer for + /// * `candidates` - List of candidate indexers to choose from + /// * `context` - Load balancing context with existing assignments and pending work async fn select_one( &self, deployment_id: DeploymentId, candidates: Vec, + context: &SelectionContext, ) -> Result, SelectionError>; /// Selects the best `num_candidates` indexers from the given list of candidates. + /// + /// # Arguments + /// * `deployment_id` - The deployment to select indexers for + /// * `candidates` - List of candidate indexers to choose from + /// * `num_candidates` - Maximum number of indexers to select + /// * `context` - Load balancing context with existing assignments and pending work async fn select( &self, deployment_id: DeploymentId, candidates: Vec, num_candidates: usize, + context: &SelectionContext, ) -> Result, SelectionError>; } diff --git a/dipper-iisa/src/http_client.rs b/dipper-iisa/src/http_client.rs index fc5505c0..cbe4a810 100644 --- a/dipper-iisa/src/http_client.rs +++ b/dipper-iisa/src/http_client.rs @@ -16,7 +16,7 @@ use reqwest::Client; use serde::{Deserialize, Serialize}; use thegraph_core::{DeploymentId, IndexerId}; -use crate::api::{CandidateSelection, Indexer, SelectionError}; +use crate::api::{CandidateSelection, Indexer, SelectionContext, SelectionError}; /// HTTP client for the IISA container service. #[derive(Clone)] @@ -108,6 +108,47 @@ impl HttpIisaClient { url: indexer.url.to_string(), } } + + /// Format existing indexers from context for the HTTP request. + /// + /// Returns `None` if the list is empty to skip serialization. + fn format_existing_indexers(context: &SelectionContext) -> Option> { + if context.existing_indexers.is_empty() { + None + } else { + Some( + context + .existing_indexers + .iter() + .map(|id| format!("{:#x}", id)) + .collect(), + ) + } + } + + /// Format pending agreements from context for the HTTP request. + /// + /// Returns `None` if the map is empty to skip serialization. + fn format_pending_agreements( + context: &SelectionContext, + ) -> Option>> { + if context.pending_agreements.is_empty() { + None + } else { + Some( + context + .pending_agreements + .iter() + .map(|(indexer_id, deployment_ids)| { + ( + format!("{:#x}", indexer_id), + deployment_ids.iter().map(|d| d.to_string()).collect(), + ) + }) + .collect(), + ) + } + } } #[async_trait] @@ -116,6 +157,7 @@ impl CandidateSelection for HttpIisaClient { &self, deployment_id: DeploymentId, candidates: Vec, + context: &SelectionContext, ) -> Result, SelectionError> { if candidates.is_empty() { return Ok(None); @@ -124,8 +166,8 @@ impl CandidateSelection for HttpIisaClient { let request = SelectionRequest { deployment_id: deployment_id.to_string(), candidates: Some(candidates.iter().map(Self::to_candidate).collect()), - existing_indexers: None, - pending_agreements: None, + existing_indexers: Self::format_existing_indexers(context), + pending_agreements: Self::format_pending_agreements(context), num_candidates: None, }; @@ -168,6 +210,7 @@ impl CandidateSelection for HttpIisaClient { deployment_id: DeploymentId, candidates: Vec, num_candidates: usize, + context: &SelectionContext, ) -> Result, SelectionError> { if candidates.is_empty() || num_candidates == 0 { return Ok(Vec::new()); @@ -176,8 +219,8 @@ impl CandidateSelection for HttpIisaClient { let request = SelectionRequest { deployment_id: deployment_id.to_string(), candidates: Some(candidates.iter().map(Self::to_candidate).collect()), - existing_indexers: None, - pending_agreements: None, + existing_indexers: Self::format_existing_indexers(context), + pending_agreements: Self::format_pending_agreements(context), num_candidates: Some(num_candidates), }; @@ -235,4 +278,64 @@ mod tests { let client = HttpIisaClient::new("http://localhost:8080/".to_string()); assert_eq!(client.endpoint, "http://localhost:8080/"); } + + #[test] + fn test_format_existing_indexers_empty() { + let context = SelectionContext::default(); + assert_eq!(HttpIisaClient::format_existing_indexers(&context), None); + } + + #[test] + fn test_format_existing_indexers_with_data() { + let indexer_id: IndexerId = "0x1234567890123456789012345678901234567890" + .parse() + .unwrap(); + let context = SelectionContext { + existing_indexers: vec![indexer_id], + pending_agreements: HashMap::new(), + }; + + let result = HttpIisaClient::format_existing_indexers(&context); + assert!(result.is_some()); + let indexers = result.unwrap(); + assert_eq!(indexers.len(), 1); + assert_eq!(indexers[0], "0x1234567890123456789012345678901234567890"); + } + + #[test] + fn test_format_pending_agreements_empty() { + let context = SelectionContext::default(); + assert_eq!(HttpIisaClient::format_pending_agreements(&context), None); + } + + #[test] + fn test_format_pending_agreements_with_data() { + let indexer_id: IndexerId = "0x1234567890123456789012345678901234567890" + .parse() + .unwrap(); + let deployment_id: DeploymentId = "QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG" + .parse() + .unwrap(); + + let mut pending = HashMap::new(); + pending.insert(indexer_id, vec![deployment_id]); + + let context = SelectionContext { + existing_indexers: vec![], + pending_agreements: pending, + }; + + let result = HttpIisaClient::format_pending_agreements(&context); + assert!(result.is_some()); + let agreements = result.unwrap(); + assert_eq!(agreements.len(), 1); + assert!(agreements.contains_key("0x1234567890123456789012345678901234567890")); + } + + #[test] + fn test_selection_context_default() { + let context = SelectionContext::default(); + assert!(context.existing_indexers.is_empty()); + assert!(context.pending_agreements.is_empty()); + } } diff --git a/dipper-iisa/src/lib.rs b/dipper-iisa/src/lib.rs index fdbe21ea..1a707a2e 100644 --- a/dipper-iisa/src/lib.rs +++ b/dipper-iisa/src/lib.rs @@ -1,5 +1,5 @@ mod api; pub mod http_client; -pub use api::{CandidateSelection, Indexer, SelectionError}; +pub use api::{CandidateSelection, Indexer, SelectionContext, SelectionError}; pub use http_client::HttpIisaClient; diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 7a1cd591..a68f600f 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -358,6 +358,47 @@ impl PgRegistry { .map_err(Into::into) } + /// Get all active agreements for multiple indexers in a single query. + /// + /// Returns agreements that are in `CREATED` or `ACCEPTED` status for any of the + /// provided indexer IDs. This is more efficient than querying each indexer separately. + pub async fn get_active_indexing_agreements_by_indexer_ids( + &self, + indexer_ids: &[IndexerId], + ) -> Result, Error> { + if indexer_ids.is_empty() { + return Ok(Vec::new()); + } + + let pg_indexer_ids: Vec = + indexer_ids.iter().map(|id| PgIndexerId(*id)).collect(); + + sqlx::query_as( + r#" + SELECT + id, + created_at, + updated_at, + status, + accepted_at_epoch, + indexing_request_id, + deployment_id, + accepted_at_epoch, + indexer_id, + indexer_url, + voucher + FROM dipper_reg_indexing_agreements + WHERE indexer_id = ANY($1) AND status IN ($2, $3) + "#, + ) + .bind(&pg_indexer_ids[..]) + .bind(IndexingAgreementStatus::Created) + .bind(IndexingAgreementStatus::Accepted) + .fetch_all(&self.pool) + .await + .map_err(Into::into) + } + pub async fn get_indexing_agreements_by_indexing_request_id( &self, request_id: &IndexingRequestId, diff --git a/dipper-pgregistry/src/postgres/common.rs b/dipper-pgregistry/src/postgres/common.rs index e4bcd751..76bf64df 100644 --- a/dipper-pgregistry/src/postgres/common.rs +++ b/dipper-pgregistry/src/postgres/common.rs @@ -2,7 +2,7 @@ use sqlx::{ Postgres, encode::IsNull, error::BoxDynError, - postgres::{PgArgumentBuffer, PgTypeInfo, PgValueRef}, + postgres::{PgArgumentBuffer, PgHasArrayType, PgTypeInfo, PgValueRef}, }; use thegraph_core::{ AllocationId, DeploymentId, IndexerId, ProofOfIndexing, @@ -171,6 +171,13 @@ impl<'r> sqlx::Decode<'r, Postgres> for PgIndexerId { } } +impl PgHasArrayType for PgIndexerId { + fn array_type_info() -> PgTypeInfo { + // bytea[] type for arrays of 20-byte addresses + PgTypeInfo::with_name("_bytea") + } +} + /// Wrapper type for `AllocationId` to implement `sqlx::Type` for `Postgres`. /// /// This _new-type_ pattern maps the `AllocationId` type to a `[u8; 20]` array diff --git a/dipper-pgregistry/tests/fixtures/0003_multi_indexer_agreements.sql b/dipper-pgregistry/tests/fixtures/0003_multi_indexer_agreements.sql new file mode 100644 index 00000000..49ec97aa --- /dev/null +++ b/dipper-pgregistry/tests/fixtures/0003_multi_indexer_agreements.sql @@ -0,0 +1,75 @@ +-- Fixture for testing batch queries across multiple indexers +-- Creates agreements for 3 different indexers with various statuses + +-- Insert an indexing request +INSERT INTO dipper_reg_indexing_requests (id, + created_at, + updated_at, + status, + requested_by, + deployment_id, + deployment_chain_id) +VALUES ('01930100-0000-7000-8000-000000000001'::uuid, + timezone('UTC', now()), + timezone('UTC', now()), + 0, -- IndexingRequestStatus::Open + '\x442a24985444cdc6a4db9503d354918d27b5ea97'::bytea, + 'QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv', + '\x000000000000a4b1'::bytea); + +-- Indexer A: 0x1111111111111111111111111111111111111111 +-- Has 2 active agreements (1 Created, 1 Accepted) +INSERT INTO dipper_reg_indexing_agreements (id, created_at, updated_at, status, indexing_request_id, deployment_id, accepted_at_epoch, indexer_id, indexer_url, voucher) +VALUES ('01930100-0001-7000-8000-000000000001'::uuid, timezone('UTC', now()), timezone('UTC', now()), + -1, -- Created + '01930100-0000-7000-8000-000000000001'::uuid, + 'QmAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA1a', + NULL, + '\x1111111111111111111111111111111111111111'::bytea, + 'https://indexer-a.com', + '{"payer": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "recipient": "0x1111111111111111111111111111111111111111", "service": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "duration_epochs": 100, "max_initial_amount": "1000", "max_ongoing_amount_per_epoch": "100", "min_epochs_per_collection": 1, "max_epochs_per_collection": 10, "deadline": 0, "metadata": {"base_price_per_epoch": "10", "price_per_entity": "1", "subgraph_deployment_id": "QmAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA1a", "protocol_network": 1, "chain_id": 1}}'::json); + +INSERT INTO dipper_reg_indexing_agreements (id, created_at, updated_at, status, indexing_request_id, deployment_id, accepted_at_epoch, indexer_id, indexer_url, voucher) +VALUES ('01930100-0001-7000-8000-000000000002'::uuid, timezone('UTC', now()), timezone('UTC', now()), + 0, -- Accepted + '01930100-0000-7000-8000-000000000001'::uuid, + 'QmBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB2b', + 12345, + '\x1111111111111111111111111111111111111111'::bytea, + 'https://indexer-a.com', + '{"payer": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "recipient": "0x1111111111111111111111111111111111111111", "service": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "duration_epochs": 100, "max_initial_amount": "1000", "max_ongoing_amount_per_epoch": "100", "min_epochs_per_collection": 1, "max_epochs_per_collection": 10, "deadline": 0, "metadata": {"base_price_per_epoch": "10", "price_per_entity": "1", "subgraph_deployment_id": "QmBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB2b", "protocol_network": 1, "chain_id": 1}}'::json); + +-- Indexer A also has 1 rejected agreement (should NOT be returned by batch query) +INSERT INTO dipper_reg_indexing_agreements (id, created_at, updated_at, status, indexing_request_id, deployment_id, accepted_at_epoch, indexer_id, indexer_url, voucher) +VALUES ('01930100-0001-7000-8000-000000000003'::uuid, timezone('UTC', now()), timezone('UTC', now()), + 2, -- Rejected + '01930100-0000-7000-8000-000000000001'::uuid, + 'QmCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC3c', + NULL, + '\x1111111111111111111111111111111111111111'::bytea, + 'https://indexer-a.com', + '{"payer": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "recipient": "0x1111111111111111111111111111111111111111", "service": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "duration_epochs": 100, "max_initial_amount": "1000", "max_ongoing_amount_per_epoch": "100", "min_epochs_per_collection": 1, "max_epochs_per_collection": 10, "deadline": 0, "metadata": {"base_price_per_epoch": "10", "price_per_entity": "1", "subgraph_deployment_id": "QmCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC3c", "protocol_network": 1, "chain_id": 1}}'::json); + +-- Indexer B: 0x2222222222222222222222222222222222222222 +-- Has 1 active agreement (Created) +INSERT INTO dipper_reg_indexing_agreements (id, created_at, updated_at, status, indexing_request_id, deployment_id, accepted_at_epoch, indexer_id, indexer_url, voucher) +VALUES ('01930100-0002-7000-8000-000000000001'::uuid, timezone('UTC', now()), timezone('UTC', now()), + -1, -- Created + '01930100-0000-7000-8000-000000000001'::uuid, + 'QmDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD4d', + NULL, + '\x2222222222222222222222222222222222222222'::bytea, + 'https://indexer-b.com', + '{"payer": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "recipient": "0x2222222222222222222222222222222222222222", "service": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "duration_epochs": 100, "max_initial_amount": "1000", "max_ongoing_amount_per_epoch": "100", "min_epochs_per_collection": 1, "max_epochs_per_collection": 10, "deadline": 0, "metadata": {"base_price_per_epoch": "10", "price_per_entity": "1", "subgraph_deployment_id": "QmDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD4d", "protocol_network": 1, "chain_id": 1}}'::json); + +-- Indexer C: 0x3333333333333333333333333333333333333333 +-- Has NO active agreements (only expired) +INSERT INTO dipper_reg_indexing_agreements (id, created_at, updated_at, status, indexing_request_id, deployment_id, accepted_at_epoch, indexer_id, indexer_url, voucher) +VALUES ('01930100-0003-7000-8000-000000000001'::uuid, timezone('UTC', now()), timezone('UTC', now()), + 5, -- Expired + '01930100-0000-7000-8000-000000000001'::uuid, + 'QmEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE5e', + 99999, + '\x3333333333333333333333333333333333333333'::bytea, + 'https://indexer-c.com', + '{"payer": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "recipient": "0x3333333333333333333333333333333333333333", "service": "0x442a24985444cdc6a4db9503d354918d27b5ea97", "duration_epochs": 100, "max_initial_amount": "1000", "max_ongoing_amount_per_epoch": "100", "min_epochs_per_collection": 1, "max_epochs_per_collection": 10, "deadline": 0, "metadata": {"base_price_per_epoch": "10", "price_per_entity": "1", "subgraph_deployment_id": "QmEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE5e", "protocol_network": 1, "chain_id": 1}}'::json); diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index ecd52705..12d20a6f 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -465,6 +465,100 @@ async fn get_rejected_indexing_agreements_by_indexing_request_id() { ); } +#[tokio::test] +async fn get_active_indexing_agreements_by_indexer_ids_batch_query() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + run_fixture( + &db, + include_str!("fixtures/0003_multi_indexer_agreements.sql"), + ) + .await + .expect("Failed to run fixture"); + let registry = PgRegistry::new(db); + + // Indexer IDs from the fixture + let indexer_a = indexer_id!("1111111111111111111111111111111111111111"); + let indexer_b = indexer_id!("2222222222222222222222222222222222222222"); + let indexer_c = indexer_id!("3333333333333333333333333333333333333333"); + let indexer_unknown = indexer_id!("9999999999999999999999999999999999999999"); + + //* When + // Query for all three indexers plus one that doesn't exist + let agreements = registry + .get_active_indexing_agreements_by_indexer_ids(&[ + indexer_a, + indexer_b, + indexer_c, + indexer_unknown, + ]) + .await + .expect("Failed to get agreements by indexer IDs"); + + //* Then + // Should return 3 active agreements: + // - Indexer A: 2 (Created + Accepted), NOT the Rejected one + // - Indexer B: 1 (Created) + // - Indexer C: 0 (only has Expired) + // - Unknown: 0 + assert_eq!(agreements.len(), 3); + + // Verify all returned agreements are active (Created or Accepted) + assert!( + agreements.iter().all(|a| matches!( + a.status, + IndexingAgreementStatus::Created | IndexingAgreementStatus::Accepted + )), + "All agreements should be in Created or Accepted status" + ); + + // Count agreements per indexer + let indexer_a_count = agreements + .iter() + .filter(|a| a.indexer.id == indexer_a) + .count(); + let indexer_b_count = agreements + .iter() + .filter(|a| a.indexer.id == indexer_b) + .count(); + let indexer_c_count = agreements + .iter() + .filter(|a| a.indexer.id == indexer_c) + .count(); + + assert_eq!( + indexer_a_count, 2, + "Indexer A should have 2 active agreements" + ); + assert_eq!( + indexer_b_count, 1, + "Indexer B should have 1 active agreement" + ); + assert_eq!( + indexer_c_count, 0, + "Indexer C should have 0 active agreements" + ); +} + +#[tokio::test] +async fn get_active_indexing_agreements_by_indexer_ids_empty_input() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + let registry = PgRegistry::new(db); + + //* When + let agreements = registry + .get_active_indexing_agreements_by_indexer_ids(&[]) + .await + .expect("Failed to get agreements for empty input"); + + //* Then + assert!( + agreements.is_empty(), + "Empty input should return empty result" + ); +} + #[tokio::test] async fn register_new_indexing_receipt_no_indexing_agreement() { //* Given