diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e9d32685..e13269d9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -101,10 +101,30 @@ jobs: sudo apt-get install -y postgresql-17 echo "/usr/lib/postgresql/$(pg_lsclusters | awk 'NR==2 {print $1}')/bin" >> $GITHUB_PATH + # The Kafka roundtrip and consumer tests skip themselves when + # REDPANDA_BROKERS is unset, so the broker must be up before they run. + - name: Start Redpanda for Kafka integration tests + run: | + docker run -d --name redpanda -p 9092:9092 redpandadata/redpanda:v24.2.7 \ + redpanda start --mode dev-container --smp 1 --memory 1G \ + --kafka-addr PLAINTEXT://0.0.0.0:9092 \ + --advertise-kafka-addr PLAINTEXT://localhost:9092 + for _ in $(seq 1 30); do + if docker exec redpanda rpk cluster health 2>/dev/null | grep -q 'Healthy:.*true'; then + exit 0 + fi + sleep 1 + done + echo "Redpanda did not become healthy in time" >&2 + docker logs redpanda >&2 + exit 1 + - name: Integration tests (in-tree) uses: LNSD/sops-exec-action@6da1fbca63459d9796097496d5f5e6233555b31a # v1 env: SOPS_AGE_KEY: ${{ secrets.IT_TESTS_AGE_KEY }} + REDPANDA_BROKERS: localhost:9092 + REQUIRE_REDPANDA: '1' with: env_file: .env run: cargo nextest run --all-features -E 'test(~tests::it_)' @@ -113,6 +133,8 @@ jobs: uses: LNSD/sops-exec-action@6da1fbca63459d9796097496d5f5e6233555b31a # v1 env: SOPS_AGE_KEY: ${{ secrets.IT_TESTS_AGE_KEY }} + REDPANDA_BROKERS: localhost:9092 + REQUIRE_REDPANDA: '1' with: env_file: .env run: cargo nextest run --all-features -E 'kind(test)' diff --git a/Cargo.lock b/Cargo.lock index a1055819..0e0663d1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2460,6 +2460,7 @@ dependencies = [ "rustls", "rustls-pemfile 2.2.0", "serde", + "serde_json", "thegraph-core", "thiserror 2.0.18", "tokio", @@ -2500,8 +2501,10 @@ dependencies = [ "futures-lite", "graph-networks-registry 0.7.0", "jsonrpsee", + "pgtemp", "rand 0.9.2", "reqwest 0.12.28", + "rskafka", "serde", "serde_json", "serde_with", diff --git a/bin/dipper-service/Cargo.toml b/bin/dipper-service/Cargo.toml index 49a792ef..35100145 100644 --- a/bin/dipper-service/Cargo.toml +++ b/bin/dipper-service/Cargo.toml @@ -37,6 +37,8 @@ url.workspace = true [dev-dependencies] fake = { workspace = true, features = ["url"] } +pgtemp = "0.6.0" +rskafka = { version = "0.6.0", features = ["transport-tls"] } # `test-util` gives tests a paused clock, so timing tests assert on an elapsed # interval instead of really waiting for one. tokio = { workspace = true, features = ["test-util"] } diff --git a/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs b/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs index af36ee22..485baa33 100644 --- a/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs +++ b/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs @@ -2,7 +2,7 @@ use std::{collections::BTreeSet, sync::Arc}; use async_trait::async_trait; use dipper_core::{ids::IndexingRequestId, state::FromState}; -use dipper_producer::{events::SubgraphIndexingAgreementEventsProducer, proto}; +use dipper_producer::events::SubgraphIndexingAgreementEventsProducer; use dipper_rpc::admin::{ SignedMessage, indexing_requests::{ @@ -17,8 +17,9 @@ use super::error_handling::{handle_list_result, handle_optional_result}; use crate::{ registry::{ IndexingRequest as IndexingRequestRecord, IndexingRequestRegistry, - IndexingRequestStatus as IndexingRequestRecordStatus, SetTargetOutcome, + IndexingRequestStatus as IndexingRequestRecordStatus, }, + set_indexing_target::{ApplyError, SetIndexingTarget, apply_set_indexing_target}, signing::eip712::Eip712Signer, worker::service::{JobPriority, WorkerQueue}, }; @@ -86,12 +87,6 @@ where ) } - #[expect( - clippy::cast_possible_truncation, - clippy::cast_possible_wrap, - clippy::too_many_lines, - reason = "predates this lint; fix when next touched" - )] async fn set_indexing_target_candidates( &self, req: SignedMessage, @@ -115,117 +110,27 @@ where let num_candidates = num_candidates.unwrap_or(self.max_candidates); - let outcome = match self - .registry - .set_indexing_target_candidates(requested_by, deployment_id, chain_id, num_candidates) - .await - { - Ok(outcome) => outcome, - Err(err) => { - tracing::error!(error=?err, "Failed to set indexing target candidates"); - return Err(ErrorObject::borrowed(503, "Service unavailable", None)); - } - }; - - // Translate the outcome into the appropriate follow-up worker job and the - // wire-level return value. - let (id_opt, reassess_count): (Option, Option) = match outcome { - SetTargetOutcome::Inserted { id } => { - tracing::info!( - indexing_request_id = %id, - %requested_by, - %deployment_id, - %chain_id, - num_candidates, - "Inserted new indexing request" - ); - - // A new request was received: emit the lifecycle event. Only the - // `Inserted` outcome is a genuinely new request; `Updated`/`Canceled` - // are later transitions in the lifecycle, not "request received". - // `the_graph_network` is the protocol network (signer chain id), not - // the deployment's data-source `chain_id`. - self.subgraph_indexing_agreements_events_emitter - .produce_subgraph_indexing_agreement_request_received( - deployment_id, - self.signer.chain_id(), - proto::SubgraphIndexingAgreementRequestReceived { - agreements_requested: num_candidates as i32, - }, - ); - - (Some(id), Some(num_candidates)) - } - SetTargetOutcome::Updated { - id, - new_num_candidates, - } => { - tracing::info!( - indexing_request_id = %id, - %requested_by, - %deployment_id, - %chain_id, - num_candidates = new_num_candidates, - "Updated num_candidates on open indexing request" - ); - (Some(id), Some(new_num_candidates)) - } - SetTargetOutcome::NoOp { id } => { - tracing::debug!( - indexing_request_id = %id, - "Set target candidates is a no-op (count unchanged)" - ); - (Some(id), None) - } - SetTargetOutcome::Canceled { id } => { - tracing::info!( - indexing_request_id = %id, - %requested_by, - %deployment_id, - %chain_id, - "Canceled indexing request (target candidates set to zero)" - ); - (Some(id), Some(0)) - } - SetTargetOutcome::NoOpAlreadyEmpty => { - tracing::warn!( - %requested_by, - %deployment_id, - %chain_id, - "set_indexing_target_candidates with num_candidates=0 against a key with no open request \ - - nothing to cancel" - ); - (None, None) + apply_set_indexing_target( + &self.registry, + &self.worker, + &self.subgraph_indexing_agreements_events_emitter, + self.signer.chain_id(), + SetIndexingTarget { + requested_by, + deployment_id, + deployment_chain_id: chain_id, + num_candidates, + // Interactive: a caller is waiting on this set-target result. + priority: JobPriority::Interactive, + }, + ) + .await + .map_err(|err| match err { + ApplyError::Registry(_) => ErrorObject::borrowed(503, "Service unavailable", None), + ApplyError::QueueReassess { .. } => { + ErrorObject::borrowed(500, "Internal server error", None) } - }; - - // Queue reassessment if the row changed. Reassessment computes the - // diff between the IISA target group of size `num_candidates` and the - // current active agreements, then grows or shrinks accordingly. With - // num_candidates=0 it shrinks to zero, firing the on-chain cancel for - // every active agreement on the key. - if let (Some(id), Some(count)) = (id_opt, reassess_count) - && let Err(err) = self - .worker - .reassess_indexing_request( - id, - deployment_id, - chain_id, - count, - // Interactive: a caller is waiting on this set-target result. - JobPriority::Interactive, - ) - .await - { - tracing::error!( - indexing_request_id = %id, - error = ?err, - "Failed to queue task: 'reassess_indexing_request'" - ); - return Err(ErrorObject::borrowed(500, "Internal server error", None)); - } - - Ok(id_opt) + }) } } @@ -273,7 +178,7 @@ mod tests { use super::*; use crate::{ - registry::Result as RegistryResult, + registry::{Result as RegistryResult, SetTargetOutcome}, test_support::{CapturedEvent, CapturingEventsProducer}, worker::queue::JobId, }; @@ -386,8 +291,7 @@ mod tests { } } - /// Build a signed `set_indexing_target_candidates` request using the - /// canonical `thegraph_core::signed_message::sign` helper and the admin + /// Build a signed `set_indexing_target_candidates` request under the admin /// EIP-712 domain, returning both the signer (so its address can be /// allowlisted) and the wrapped `SignedMessage`. fn signed_request( @@ -406,10 +310,9 @@ mod tests { (signer, inner.into()) } - /// Assemble an `RpcServerImpl` whose signer's chain id is - /// [`SIGNER_CHAIN_ID`], whose allowlist contains `allowed`, and whose - /// registry returns `outcome`. Returns the server and the shared events - /// capture for assertions. + /// Assemble an `RpcServerImpl` with [`SIGNER_CHAIN_ID`], allowlist + /// `allowed`, and a registry returning `outcome`; also returns the shared + /// events capture for assertions. fn server( allowed: Address, outcome: SetTargetOutcome, diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index 810af7ac..7f6b1a35 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -9,7 +9,7 @@ use std::{ }; use dipper_core::config::{Hidden, HiddenSecretKeyAsHexStr}; -use dipper_producer::kafka::KafkaConfig; +use dipper_producer::kafka::{KafkaConfig, KafkaConsumerConfig}; use serde_with::serde_as; use thegraph_core::alloy::{ primitives::{Address, ChainId, U256}, @@ -86,6 +86,10 @@ pub struct Config { /// Events configuration for sending dipper events on the configured topic for streaming #[serde(default)] pub event_streaming_config: Option, + /// The Studio indexing request consumer configuration (reads subgraph + /// indexing requests from a Redpanda topic; absent means off) + #[serde(default)] + pub indexing_request_consumer: Option, /// Number of concurrent worker loops draining the job queue (default: 8). /// Each loop can hold up to three pooled DB connections at once and shares /// the pool with the registry and background services; size accordingly. @@ -1517,10 +1521,152 @@ pub fn default_event_queue_capacity() -> NonZeroUsize { NonZeroUsize::new(1024).expect("default event queue capacity is non-zero") } +/// Configuration for the Studio indexing request consumer, which reads +/// subgraph indexing request events from a Redpanda topic and applies them +/// through the same path as the admin RPC. +#[serde_as] +#[derive(Debug, Clone, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IndexingRequestConsumerConfig { + /// Whether the consumer is enabled (default: false). Leave off until + /// Studio's propose events carry `indexed_network_caip2id`: without the + /// field every consumed request is skipped and its offset committed, so + /// enabling early permanently discards real requests. + #[serde(default)] + pub enabled: bool, + + /// Kafka connection settings and the topic Studio produces on. The topic + /// is required with no default; a wrong or missing name fails at startup. + pub kafka: KafkaConsumerConfig, + + /// The identity recorded as the requester on consumed requests, since + /// Kafka messages carry no signature to recover one from. Use the address + /// Studio signs the admin RPC with, so both doors share request rows. + pub requested_by: Address, + + /// How long a fetch waits server-side for new records before returning + /// empty, in seconds (default: 5). + #[serde_as(as = "serde_with::DurationSeconds")] + #[serde(default = "default_indexing_request_consumer_max_wait")] + pub max_wait: Duration, + + /// Maximum bytes per fetch (default: 1,048,576). + #[serde(default = "default_indexing_request_consumer_fetch_max_bytes")] + pub fetch_max_bytes: i32, +} + +fn default_indexing_request_consumer_max_wait() -> Duration { + Duration::from_secs(5) +} + +fn default_indexing_request_consumer_fetch_max_bytes() -> i32 { + 1_048_576 +} + +impl IndexingRequestConsumerConfig { + /// Reject a configuration the consumer cannot run with. Checked even when + /// disabled, so a broken value cannot lie dormant until the flag flips. + pub fn validate(&self) -> Result<(), String> { + if self.kafka.brokers.is_empty() { + return Err( + "indexing_request_consumer.kafka.brokers must list at least 1 broker".to_string(), + ); + } + // 0 makes every connect time out instantly, so startup would fail with + // a message blaming the broker instead of the config typo. + if self.kafka.connect_timeout_secs == 0 { + return Err( + "indexing_request_consumer.kafka.connect_timeout_secs must be at least 1" + .to_string(), + ); + } + // A zero requester is almost certainly an unset value, and it would + // silently key every consumed request under the zero address. + if self.requested_by == Address::ZERO { + return Err( + "indexing_request_consumer.requested_by must be a non-zero address".to_string(), + ); + } + // Below ~1 KB a fetch cannot hold a whole record, so the consumer + // would poll forever without ever making progress. + if self.fetch_max_bytes < 1_024 { + return Err(format!( + "indexing_request_consumer.fetch_max_bytes ({}) must be at least 1024", + self.fetch_max_bytes + )); + } + if self.max_wait.is_zero() || self.max_wait.as_millis() > i32::MAX as u128 { + return Err(format!( + "indexing_request_consumer.max_wait ({}s) must be between 1 second and {} seconds", + self.max_wait.as_secs(), + i32::MAX / 1_000 + )); + } + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; + fn consumer_config(json: serde_json::Value) -> IndexingRequestConsumerConfig { + serde_json::from_value(json).expect("deserializes") + } + + #[test] + fn indexing_request_consumer_config_defaults_and_validation() { + let config = consumer_config(serde_json::json!({ + "kafka": { "brokers": ["localhost:9092"], "topic": "t" }, + "requested_by": "0x8f8c426f956876325b1e037c6eae9b189952994c", + })); + assert!( + !config.enabled, + "the consumer must be off unless opted into" + ); + assert_eq!(config.max_wait, Duration::from_secs(5)); + assert_eq!(config.fetch_max_bytes, 1_048_576); + assert!(config.validate().is_ok()); + + // Validation runs even for a disabled section, so broken values are + // caught at startup rather than the day the flag flips. + let broken = consumer_config(serde_json::json!({ + "enabled": false, + "kafka": { "brokers": ["localhost:9092"], "topic": "t" }, + "requested_by": "0x0000000000000000000000000000000000000000", + })); + assert!(broken.validate().unwrap_err().contains("requested_by")); + + let no_brokers = consumer_config(serde_json::json!({ + "kafka": { "brokers": [], "topic": "t" }, + "requested_by": "0x8f8c426f956876325b1e037c6eae9b189952994c", + })); + assert!(no_brokers.validate().unwrap_err().contains("brokers")); + + let zero_connect_timeout = consumer_config(serde_json::json!({ + "kafka": { "brokers": ["localhost:9092"], "topic": "t", "connect_timeout_secs": 0 }, + "requested_by": "0x8f8c426f956876325b1e037c6eae9b189952994c", + })); + assert!( + zero_connect_timeout + .validate() + .unwrap_err() + .contains("connect_timeout_secs") + ); + + let tiny_fetch = consumer_config(serde_json::json!({ + "kafka": { "brokers": ["localhost:9092"], "topic": "t" }, + "requested_by": "0x8f8c426f956876325b1e037c6eae9b189952994c", + "fetch_max_bytes": 512, + })); + assert!( + tiny_fetch + .validate() + .unwrap_err() + .contains("fetch_max_bytes") + ); + } + #[test] fn test_dips_agreement_config_deserialization() { //* Arrange - JSON config with all new field names diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 707ce765..4809e475 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -29,6 +29,7 @@ mod health; mod indexer_rpc_client; mod network; mod registry; +mod set_indexing_target; mod signing; mod supervisor; #[cfg(test)] @@ -103,6 +104,11 @@ pub async fn main() -> anyhow::Result<()> { { anyhow::bail!("invalid event streaming config: {err}"); } + if let Some(consumer_conf) = &conf.indexing_request_consumer + && let Err(err) = consumer_conf.validate() + { + anyhow::bail!("invalid indexing request consumer config: {err}"); + } if let Err(err) = conf.health.validate(conf.admin_rpc.listen_addr) { anyhow::bail!("invalid health config: {err}"); } @@ -605,6 +611,34 @@ pub async fn main() -> anyhow::Result<()> { None }; + //- The Studio indexing request consumer service (optional, enabled by config) + // Reads subgraph indexing requests from Redpanda and applies them through + // the same path as the admin RPC. + let indexing_request_consumer_handle = match conf.indexing_request_consumer { + Some(ref consumer_conf) if consumer_conf.enabled => { + let ctx = network::service::indexing_request_consumer::Ctx { + registry: registry.clone(), + worker_queue: worker_handle.queue().clone(), + events: subgraph_indexing_agreements_events_emitter.clone(), + protocol_chain_id: chain_id, + max_candidates: DEFAULT_MAX_CANDIDATES, + config: consumer_conf.clone(), + }; + let (handle, service) = network::service::indexing_request_consumer::new(ctx); + // A reassessment push lost at the shutdown edge is only repaired by + // the periodic reassignment sweep; without it the request would sit + // open with no job behind it. + if !conf.reassignment.as_ref().is_some_and(|r| r.enabled) { + tracing::warn!( + "the indexing request consumer is enabled without the reassignment service; \ + a reassessment job lost during a shutdown would never be retried" + ); + } + Some((handle, service)) + } + _ => None, + }; + //- The admin RPC service let (admin_rpc_handle, admin_rpc_service) = { let config = admin_rpc_server::service::Config { @@ -719,6 +753,16 @@ pub async fn main() -> anyhow::Result<()> { None }; + // Spawn the indexing request consumer service if enabled + let indexing_request_consumer_stop_handle = + if let Some((handle, service)) = indexing_request_consumer_handle { + let task_handle = task_tree.spawn(service); + tracing::debug!(task_id=%task_handle.id(), "Indexing request consumer service started"); + Some(handle) + } else { + None + }; + // Spawn the health endpoint if enabled let health_stop_handle = if let Some((handle, service)) = health_handle { let task_handle = task_tree.spawn(service); @@ -798,6 +842,11 @@ pub async fn main() -> anyhow::Result<()> { all_stopped &= stop_service("Escrow reconciler", handle.stop()).await; } + // Stop the indexing request consumer before worker (it queues worker jobs) + if let Some(handle) = indexing_request_consumer_stop_handle { + all_stopped &= stop_service("Indexing request consumer", handle.stop()).await; + } + // Stop entity count cache service if let Some(handle) = entity_count_handle { all_stopped &= stop_service("Entity count cache", handle.stop()).await; diff --git a/bin/dipper-service/src/network/service.rs b/bin/dipper-service/src/network/service.rs index 3fc9d0f0..35cbf849 100644 --- a/bin/dipper-service/src/network/service.rs +++ b/bin/dipper-service/src/network/service.rs @@ -6,5 +6,6 @@ pub mod entity_count_cache; pub mod escrow_reconciler; pub mod expiration; pub mod indexer_urls; +pub mod indexing_request_consumer; pub mod liveness_checker; pub mod reassignment; diff --git a/bin/dipper-service/src/network/service/indexing_request_consumer.rs b/bin/dipper-service/src/network/service/indexing_request_consumer.rs new file mode 100644 index 00000000..5ee9d07f --- /dev/null +++ b/bin/dipper-service/src/network/service/indexing_request_consumer.rs @@ -0,0 +1,1505 @@ +//! Consumes the subgraph indexing request events Studio produces on Redpanda +//! and applies each one as a set-indexing-target change, making the topic a +//! second front door to the same path the admin RPC serves. + +use std::{future::Future, sync::Arc, time::Duration}; + +use async_trait::async_trait; +use dipper_producer::{ + events::SubgraphIndexingAgreementEventsProducer, + kafka::{ConsumerError, KafkaConsumer, OffsetAt}, + prost::Message as _, + proto::studio, +}; +use thegraph_core::{ + DeploymentId, + alloy::primitives::{Address, ChainId}, +}; +use tokio::{ + sync::{mpsc, watch}, + task::JoinSet, +}; + +use crate::{ + config::IndexingRequestConsumerConfig, + registry::IndexingRequestRegistry, + set_indexing_target::{ApplyError, SetIndexingTarget, apply_set_indexing_target}, + worker::service::{JobPriority, WorkerQueue}, +}; + +/// The propose event type Studio sends. +const EVENT_TYPE_PROPOSE: &str = "subgraph.indexing.request.propose"; + +/// Studio's producer also defines a terminate event, but nothing sends it: +/// cancellation is a propose with a count of 0. Logged and skipped if seen. +const EVENT_TYPE_TERMINATE: &str = "subgraph.indexing.agreements.terminate"; + +/// Extra connection attempts after the first before startup fails visibly. +const CONNECT_MAX_RETRIES: u32 = 5; + +/// Delay before retrying after a fetch or apply failure. +const RETRY_BACKOFF: Duration = Duration::from_secs(5); + +/// Pause after a fetch below the high watermark that returned no usable +/// records, so a run of dropped batches cannot spin the loop hot. +const EMPTY_FETCH_PAUSE: Duration = Duration::from_secs(1); + +/// How often to re-read topic metadata to notice a partition count change. +const PARTITION_METADATA_CHECK_INTERVAL: Duration = Duration::from_secs(300); + +/// Bound on one metadata re-read. Deliberately shorter than the 5-second stop +/// cap, since the stop channel is not polled while the read is in flight. +const PARTITION_METADATA_CHECK_TIMEOUT: Duration = Duration::from_secs(3); + +/// Handle for controlling the indexing request consumer lifecycle +#[derive(Clone)] +pub struct Handle { + tx_stop: mpsc::Sender<()>, +} + +impl Handle { + /// Stop the consumer gracefully + pub async fn stop(&self) { + if self.tx_stop.is_closed() { + return; + } + + let _ = self.tx_stop.send(()).await; + self.tx_stop.closed().await; + } +} + +/// Registry for persisting per-partition consumer progress. +#[async_trait] +pub trait KafkaConsumerOffsetRegistry { + /// Get the next offset to fetch for a topic partition, `None` on first run. + async fn get_kafka_consumer_offset( + &self, + topic: &str, + partition_id: i32, + ) -> Result, crate::registry::Error>; + + /// Record the next offset to fetch for a topic partition. + async fn set_kafka_consumer_offset( + &self, + topic: &str, + partition_id: i32, + next_offset: i64, + ) -> Result<(), crate::registry::Error>; +} + +/// Context required by the indexing request consumer service +pub struct Ctx { + /// Registry for indexing requests and consumer offsets + pub registry: R, + /// Worker queue for the reassessment jobs that follow an applied request + pub worker_queue: W, + /// Lifecycle events emitter (request-received on newly inserted requests) + pub events: Arc, + /// The protocol network chain id (signer chain id), for validating the + /// envelope's network and stamping emitted lifecycle events + pub protocol_chain_id: ChainId, + /// Ceiling on the indexer count a consumed request may ask for; larger + /// counts are clamped with a warning. Kafka records carry no signature, + /// so this door gets a cap the signed admin RPC does not need. + pub max_candidates: usize, + /// Service configuration + pub config: IndexingRequestConsumerConfig, +} + +/// Create a new indexing request consumer service: a control handle plus a +/// future to spawn that reads Studio's propose events from Kafka, applies each +/// as a set-indexing-target change, and records its progress per partition. +pub fn new(ctx: Ctx) -> (Handle, impl Future>) +where + R: IndexingRequestRegistry + KafkaConsumerOffsetRegistry + Clone + Send + Sync + 'static, + W: WorkerQueue + Clone + Send + Sync + 'static, +{ + let (tx_stop, mut rx_stop) = mpsc::channel(1); + + let service = async move { + let consumer = match connect_with_retries(&ctx.config, &mut rx_stop).await { + Ok(Some(consumer)) => Arc::new(consumer), + // Stop was requested while still connecting; a clean exit. + Ok(None) => return Ok(()), + Err(err) => return Err(err), + }; + + let partitions = consumer.partitions(); + tracing::info!( + topic = consumer.topic(), + partitions = partitions.len(), + requested_by = %ctx.config.requested_by, + "indexing request consumer started" + ); + + let (shutdown_tx, shutdown_rx) = watch::channel(false); + let mut tasks: JoinSet<()> = JoinSet::new(); + for partition in partitions { + tasks.spawn(partition_loop( + Arc::clone(&consumer), + partition, + ctx.registry.clone(), + ctx.worker_queue.clone(), + Arc::clone(&ctx.events), + ctx.protocol_chain_id, + ctx.max_candidates, + ctx.config.clone(), + shutdown_rx.clone(), + )); + } + + // Partition loops only return on shutdown, so one finishing early means + // it panicked or hit a bug; tear the service down so the process + // restarts instead of consuming a partial set of partitions. The + // metadata timer notices a topic growing partitions: new partitions + // would otherwise be consumed by nobody, silently losing requests, so + // that also restarts the service to pick up the full layout. + let serving = tasks.len(); + let mut metadata_check = tokio::time::interval(PARTITION_METADATA_CHECK_INTERVAL); + metadata_check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let result = loop { + tokio::select! { + _ = rx_stop.recv() => break Ok(()), + joined = tasks.join_next() => break match joined { + Some(Ok(())) => Err(anyhow::anyhow!( + "an indexing request consumer partition loop exited unexpectedly" + )), + Some(Err(err)) => Err(anyhow::anyhow!( + "an indexing request consumer partition loop panicked: {err}" + )), + None => Err(anyhow::anyhow!( + "the indexing request consumer had no partition loops to run" + )), + }, + _ = metadata_check.tick() => { + let count = tokio::time::timeout( + PARTITION_METADATA_CHECK_TIMEOUT, + consumer.current_partition_count(), + ) + .await; + match count { + Ok(Ok(count)) if count > serving => break Err(anyhow::anyhow!( + "topic '{}' now has {count} partitions but this consumer serves \ + {serving}; restarting to consume the full set", + consumer.topic() + )), + // Kafka partitions only ever grow, so a lower count is a + // stale or partial metadata read, never a real change; + // restarting the process over it would be a false alarm. + Ok(Ok(count)) if count < serving => tracing::warn!( + count, + serving, + "metadata reported fewer partitions than this consumer serves; \ + ignoring the stale read" + ), + Ok(Ok(_)) => {} + Ok(Err(err)) => { + tracing::warn!(error = %err, "failed to re-check topic partition metadata"); + } + Err(_) => { + tracing::warn!("topic partition metadata re-check timed out"); + } + } + }, + } + }; + + let _ = shutdown_tx.send(true); + while tasks.join_next().await.is_some() {} + + tracing::info!("indexing request consumer stopped"); + result + }; + + (Handle { tx_stop }, service) +} + +/// Connects to the brokers, retrying transient failures a bounded number of +/// times. A missing topic fails immediately: it is configuration, and reading +/// from a wrong or absent topic must be loud, not an idle consumer. +async fn connect_with_retries( + config: &IndexingRequestConsumerConfig, + rx_stop: &mut mpsc::Receiver<()>, +) -> anyhow::Result> { + let mut attempt: u32 = 0; + loop { + match KafkaConsumer::connect(&config.kafka).await { + Ok(consumer) => return Ok(Some(consumer)), + Err(err @ ConsumerError::TopicNotFound { .. }) => { + return Err(anyhow::anyhow!( + "indexing request consumer startup failed: {err}; check the configured topic \ + name against the topic Studio produces on" + )); + } + Err(err) if attempt < CONNECT_MAX_RETRIES => { + attempt += 1; + let delay = Duration::from_secs(2u64.pow(attempt.min(5))); + tracing::warn!( + attempt, + delay_secs = delay.as_secs(), + error = %err, + "indexing request consumer connect failed, retrying" + ); + tokio::select! { + _ = rx_stop.recv() => return Ok(None), + _ = tokio::time::sleep(delay) => {} + } + } + Err(err) => { + return Err(anyhow::anyhow!( + "indexing request consumer failed to connect after {} attempts: {err}", + CONNECT_MAX_RETRIES + 1 + )); + } + } + } +} + +/// Consumes one partition sequentially: fetch from the persisted offset, apply +/// each record, then persist the offset past it. Delivery is at-least-once; +/// redelivering an open request's count is a registry no-op, though a replay +/// reaching back past a cancellation briefly re-opens it until the cancel +/// replays too, which is why re-anchoring below never rewinds further than +/// the broker forces it to. +#[expect( + clippy::cognitive_complexity, + clippy::too_many_arguments, + clippy::too_many_lines, + reason = "predates this lint; fix when next touched" +)] +async fn partition_loop( + consumer: Arc, + partition: i32, + registry: R, + worker_queue: W, + events: Arc, + protocol_chain_id: ChainId, + max_candidates: usize, + config: IndexingRequestConsumerConfig, + mut shutdown_rx: watch::Receiver, +) where + R: IndexingRequestRegistry + KafkaConsumerOffsetRegistry + Send + Sync, + W: WorkerQueue + Send + Sync, +{ + let topic = consumer.topic().to_string(); + let expected_network = format!("eip155:{protocol_chain_id}"); + let max_wait_ms = i32::try_from(config.max_wait.as_millis()).unwrap_or(i32::MAX); + + // Resume from the persisted offset; a partition never seen before starts at + // the earliest available record so requests published before the consumer's + // first deploy are not lost. + let mut next_offset = loop { + let restored = match registry.get_kafka_consumer_offset(&topic, partition).await { + Ok(restored) => restored, + Err(err) => { + tracing::error!(partition, error = %err, "failed to load consumer offset, retrying"); + if sleep_or_shutdown(&mut shutdown_rx, RETRY_BACKOFF).await { + return; + } + continue; + } + }; + match restored { + Some(offset) => break offset, + None => match consumer.offset(partition, OffsetAt::Earliest).await { + Ok(earliest) => break earliest, + Err(err) => { + tracing::error!(partition, error = %err, "failed to query earliest offset, retrying"); + if sleep_or_shutdown(&mut shutdown_rx, RETRY_BACKOFF).await { + return; + } + } + }, + } + }; + tracing::debug!(partition, next_offset, "partition consumer resuming"); + + loop { + let fetch = tokio::select! { + _ = shutdown_rx.changed() => return, + fetch = consumer.fetch(partition, next_offset, config.fetch_max_bytes, max_wait_ms) => fetch, + }; + + let (records, high_watermark) = match fetch { + Ok((records, high_watermark)) => (records, high_watermark), + Err(err) if err.is_offset_out_of_range() => { + // The broker refuses a cursor outside its retained range: below + // the log start when retention deleted records, or above the + // log end when the topic was recreated or truncated. Clamp to + // the nearest live edge; always rewinding to earliest would + // replay the whole retained topic in the truncation case. + let range = match consumer.offset(partition, OffsetAt::Earliest).await { + Ok(earliest) => match consumer.offset(partition, OffsetAt::Latest).await { + Ok(latest) => Some((earliest, latest)), + Err(err) => { + tracing::error!(partition, error = %err, "failed to query latest offset"); + None + } + }, + Err(err) => { + tracing::error!(partition, error = %err, "failed to query earliest offset"); + None + } + }; + match range { + Some((earliest, latest)) => { + let re_anchored = if next_offset < earliest { + earliest + } else { + latest + }; + tracing::warn!( + partition, + stale_offset = next_offset, + earliest, + latest, + re_anchored, + "consumer offset fell outside the broker's retained range; re-anchoring" + ); + next_offset = re_anchored; + persist_offset(®istry, &topic, partition, next_offset).await; + } + None => { + if sleep_or_shutdown(&mut shutdown_rx, RETRY_BACKOFF).await { + return; + } + } + } + continue; + } + Err(err) => { + tracing::error!(partition, error = %err, "fetch failed, backing off"); + if sleep_or_shutdown(&mut shutdown_rx, RETRY_BACKOFF).await { + return; + } + continue; + } + }; + + let records_was_empty = records.is_empty(); + for record_and_offset in records { + let offset = record_and_offset.offset; + match decode_propose(record_and_offset.record.value.as_deref(), &expected_network) { + Err(reason) => { + tracing::warn!( + topic, + partition, + offset, + reason, + key = ?record_and_offset.record.key.as_deref().map(String::from_utf8_lossy), + "skipping unprocessable indexing request record" + ); + } + Ok(target) => { + tracing::debug!( + event_id = %target.event_id, + deployment_id = %target.deployment_id, + deployment_chain_id = target.deployment_chain_id, + num_candidates = target.num_candidates, + "consumed indexing request propose event" + ); + let applied = apply_target_with_retries( + ®istry, + &worker_queue, + &events, + protocol_chain_id, + config.requested_by, + max_candidates, + &target, + partition, + offset, + &mut shutdown_rx, + ) + .await; + if !applied { + // Shutdown arrived before the record took effect; the + // unadvanced offset redelivers it on the next run. + return; + } + } + } + + next_offset = offset + 1; + persist_offset(®istry, &topic, partition, next_offset).await; + } + + // An empty fetch below the high watermark (e.g. a batch of records the + // client filtered out) would otherwise loop again instantly: pause so + // a run of them cannot spin hot. + if records_was_empty && high_watermark > next_offset { + tracing::debug!( + partition, + next_offset, + high_watermark, + "fetch below the high watermark returned no records; pausing" + ); + if sleep_or_shutdown(&mut shutdown_rx, EMPTY_FETCH_PAUSE).await { + return; + } + } + } +} + +/// Persist consumer progress. Failure is logged but does not halt consumption: +/// the in-memory cursor stays correct and a later write covers the gap, at the +/// cost of some redelivery after a restart (which is safe). +async fn persist_offset( + registry: &R, + topic: &str, + partition: i32, + next_offset: i64, +) { + if let Err(err) = registry + .set_kafka_consumer_offset(topic, partition, next_offset) + .await + { + tracing::error!( + topic, + partition, + next_offset, + error = %err, + "failed to persist consumer offset; progress will be re-delivered after a restart" + ); + } +} + +/// Wait out a backoff, returning `true` when shutdown was requested instead. +async fn sleep_or_shutdown(shutdown_rx: &mut watch::Receiver, delay: Duration) -> bool { + tokio::select! { + _ = shutdown_rx.changed() => true, + _ = tokio::time::sleep(delay) => false, + } +} + +/// Apply a decoded propose until it fully takes effect, returning `false` if +/// shutdown was requested first. A registry failure retries the whole apply +/// (nothing was committed); a queue failure retries only the job push, because +/// the row change is already committed and re-running the apply would land on +/// the registry's no-op path and silently drop the reassessment. +#[allow(clippy::too_many_arguments)] +async fn apply_target_with_retries( + registry: &R, + worker_queue: &W, + events: &Arc, + protocol_chain_id: ChainId, + requested_by: Address, + max_candidates: usize, + target: &ProposedTarget, + partition: i32, + offset: i64, + shutdown_rx: &mut watch::Receiver, +) -> bool +where + R: IndexingRequestRegistry + Send + Sync, + W: WorkerQueue + Send + Sync, +{ + // Cap the count from the wire: this door has no signature to vouch for the + // sender, so an absurd target must not reach indexer selection unclamped. + // 0 passes through untouched, since it is the cancellation shape. + let num_candidates = if target.num_candidates > max_candidates { + tracing::warn!( + partition, + offset, + requested = target.num_candidates, + max_candidates, + "clamping the requested indexer count to the configured maximum" + ); + max_candidates + } else { + target.num_candidates + }; + + let queue_retry_id = loop { + let result = apply_set_indexing_target( + registry, + worker_queue, + events, + protocol_chain_id, + SetIndexingTarget { + requested_by, + deployment_id: target.deployment_id, + deployment_chain_id: target.deployment_chain_id, + num_candidates, + // Interactive: a developer just asked for this in Studio. + priority: JobPriority::Interactive, + }, + ) + .await; + + match result { + Ok(_) => return true, + Err(err @ ApplyError::Registry(_)) => { + tracing::error!( + partition, + offset, + error = %err, + "failed to apply indexing request, retrying" + ); + if sleep_or_shutdown(shutdown_rx, RETRY_BACKOFF).await { + return false; + } + } + Err(ApplyError::QueueReassess { id, .. }) => break id, + } + }; + + loop { + if sleep_or_shutdown(shutdown_rx, RETRY_BACKOFF).await { + // One last immediate attempt: the row change is already committed, + // so leaving without the job strands the request until the daily + // reassignment sweep next queues it. + return retry_reassess_push(worker_queue, queue_retry_id, num_candidates, target).await; + } + if retry_reassess_push(worker_queue, queue_retry_id, num_candidates, target).await { + return true; + } + } +} + +/// One attempt at the reassessment push that failed inside the apply. +async fn retry_reassess_push( + worker_queue: &W, + id: dipper_core::ids::IndexingRequestId, + num_candidates: usize, + target: &ProposedTarget, +) -> bool +where + W: WorkerQueue + Send + Sync, +{ + match worker_queue + .reassess_indexing_request( + id, + target.deployment_id, + target.deployment_chain_id, + num_candidates, + JobPriority::Interactive, + ) + .await + { + Ok(_) => true, + Err(err) => { + tracing::error!( + indexing_request_id = %id, + error = ?err, + "retrying the reassessment job push" + ); + false + } + } +} + +/// A validated propose event, reduced to what the registry call needs. +#[derive(Debug, PartialEq, Eq)] +struct ProposedTarget { + event_id: String, + deployment_id: DeploymentId, + deployment_chain_id: ChainId, + num_candidates: usize, +} + +/// Decode a record value into a propose target, or the reason to skip it. +/// Unknown event types are tolerated by design: Studio may add types before +/// the dipper learns them, and they must not wedge the partition. +fn decode_propose( + value: Option<&[u8]>, + expected_network: &str, +) -> Result { + let Some(value) = value else { + return Err("empty record value"); + }; + + let event = match studio::SubgraphIndexingRequestEvent::decode(value) { + Ok(event) => event, + Err(_) => return Err("undecodable protobuf"), + }; + + match event.event_type.as_str() { + EVENT_TYPE_PROPOSE => {} + EVENT_TYPE_TERMINATE => return Err("terminate event (cancellation is a propose with 0)"), + _ => return Err("unknown event type"), + } + + if event.the_graph_network_caip2id != expected_network { + return Err("event is for a different protocol network"); + } + + let Some(studio::subgraph_indexing_request_event::Payload::SubgraphIndexingRequestPropose( + propose, + )) = event.payload + else { + return Err("propose event without a propose payload"); + }; + + let Ok(deployment_id) = event.subgraph_deployment_qm_hash.parse::() else { + return Err("invalid subgraph deployment hash"); + }; + + let Some(deployment_chain_id) = parse_eip155_caip2(&propose.indexed_network_caip2id) else { + // The field is the dipper's addition to Studio's schema; until Studio + // sends it, every message lands here and this warn is the signal. + return Err("missing or invalid indexed network caip2 id"); + }; + + let Ok(num_candidates) = usize::try_from(propose.indexing_agreements_requested) else { + return Err("negative indexing agreements requested"); + }; + + Ok(ProposedTarget { + event_id: event.event_id, + deployment_id, + deployment_chain_id, + num_candidates, + }) +} + +/// Parse an `eip155:{chain_id}` CAIP-2 identifier into its numeric chain id. +fn parse_eip155_caip2(value: &str) -> Option { + value.strip_prefix("eip155:")?.parse::().ok() +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use dipper_core::ids::{IndexingAgreementId, IndexingRequestId}; + use thegraph_core::{DeploymentId, deployment_id}; + use url::Url; + + use super::*; + use crate::{ + registry::{ + IndexingRequest as IndexingRequestRecord, Result as RegistryResult, SetTargetOutcome, + }, + test_support::{CapturedEvent, CapturingEventsProducer}, + worker::queue::JobId, + }; + + /// The protocol (signer) chain id, distinct from the indexed chain below + /// so assertions can tell the 2 apart. + const PROTOCOL_CHAIN_ID: ChainId = 42161; + + /// The chain the test deployment indexes. + const INDEXED_CHAIN_ID: ChainId = 1; + + const QM_HASH: &str = "QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"; + + /// Ceiling on requested indexer counts in these tests. + const TEST_MAX_CANDIDATES: usize = 10; + + fn requester() -> Address { + "0x8f8c426f956876325b1e037c6eae9b189952994c" + .parse() + .expect("valid address") + } + + /// Encode a propose envelope the way Studio's producer does. + fn encode_propose( + event_type: &str, + network: &str, + qm_hash: &str, + indexed_network: &str, + count: i32, + ) -> Vec { + let event = studio::SubgraphIndexingRequestEvent { + event_id: "01912345-6789-7abc-def0-123456789abc".to_string(), + event_type: event_type.to_string(), + event_version: "1.0".to_string(), + timestamp: "2026-08-24T10:30:00.123Z".to_string(), + subgraph_deployment_qm_hash: qm_hash.to_string(), + the_graph_network_caip2id: network.to_string(), + payload: Some( + studio::subgraph_indexing_request_event::Payload::SubgraphIndexingRequestPropose( + studio::SubgraphIndexingRequestPropose { + indexing_agreements_requested: count, + indexed_network_caip2id: indexed_network.to_string(), + }, + ), + ), + }; + event.encode_to_vec() + } + + fn valid_propose_bytes() -> Vec { + encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "eip155:1", 3) + } + + // -------- decode_propose -------- + + #[test] + fn decodes_a_valid_propose_event() { + let bytes = valid_propose_bytes(); + let target = decode_propose(Some(&bytes), "eip155:42161").expect("decodes"); + + assert_eq!(target.deployment_id, deployment_id!(QM_HASH)); + assert_eq!(target.deployment_chain_id, INDEXED_CHAIN_ID); + assert_eq!(target.num_candidates, 3); + assert_eq!(target.event_id, "01912345-6789-7abc-def0-123456789abc"); + } + + #[test] + fn tolerates_unknown_fields_appended_to_the_envelope() { + // A future schema revision adds fields this consumer does not know: + // field 15, wire type 2 (length-delimited), 3 bytes of payload. + let mut bytes = valid_propose_bytes(); + bytes.extend_from_slice(&[0x7A, 0x03, b'a', b'b', b'c']); + + let target = decode_propose(Some(&bytes), "eip155:42161").expect("decodes"); + assert_eq!(target.num_candidates, 3); + } + + #[test] + fn skips_a_zero_count_as_a_valid_cancellation() { + // Count 0 is not a skip: it is the agreed cancellation shape. + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "eip155:1", 0); + let target = decode_propose(Some(&bytes), "eip155:42161").expect("decodes"); + assert_eq!(target.num_candidates, 0); + } + + #[test] + fn rejects_an_empty_record_value() { + assert_eq!( + decode_propose(None, "eip155:42161"), + Err("empty record value") + ); + } + + #[test] + fn rejects_undecodable_bytes() { + // 0xFF is a field-15 wire-type-7 tag; wire type 7 does not exist. + let garbage = [0xFF, 0xFF, 0xFF]; + assert_eq!( + decode_propose(Some(&garbage), "eip155:42161"), + Err("undecodable protobuf") + ); + } + + #[test] + fn rejects_an_unknown_event_type() { + let bytes = encode_propose( + "subgraph.indexing.request.some_future_type", + "eip155:42161", + QM_HASH, + "eip155:1", + 3, + ); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("unknown event type") + ); + } + + #[test] + fn rejects_a_terminate_event() { + let bytes = encode_propose(EVENT_TYPE_TERMINATE, "eip155:42161", QM_HASH, "eip155:1", 3); + assert!(matches!( + decode_propose(Some(&bytes), "eip155:42161"), + Err(reason) if reason.contains("terminate") + )); + } + + #[test] + fn rejects_an_event_for_another_protocol_network() { + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:421614", QM_HASH, "eip155:1", 3); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("event is for a different protocol network") + ); + } + + #[test] + fn rejects_a_propose_without_a_payload() { + let event = studio::SubgraphIndexingRequestEvent { + event_id: "e".to_string(), + event_type: EVENT_TYPE_PROPOSE.to_string(), + event_version: "1.0".to_string(), + timestamp: "t".to_string(), + subgraph_deployment_qm_hash: QM_HASH.to_string(), + the_graph_network_caip2id: "eip155:42161".to_string(), + payload: None, + }; + assert_eq!( + decode_propose(Some(&event.encode_to_vec()), "eip155:42161"), + Err("propose event without a propose payload") + ); + } + + #[test] + fn rejects_an_invalid_deployment_hash() { + let bytes = encode_propose( + EVENT_TYPE_PROPOSE, + "eip155:42161", + "not-a-deployment-hash", + "eip155:1", + 3, + ); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("invalid subgraph deployment hash") + ); + } + + #[test] + fn rejects_a_missing_indexed_network() { + // Studio has not added the field yet: it decodes as an empty string. + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "", 3); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("missing or invalid indexed network caip2 id") + ); + } + + #[test] + fn rejects_a_malformed_indexed_network() { + for indexed in ["cosmos:hub", "eip155:", "eip155:abc", "1"] { + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, indexed, 3); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("missing or invalid indexed network caip2 id"), + "indexed network {indexed:?} should be rejected" + ); + } + } + + #[test] + fn rejects_a_negative_candidate_count() { + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "eip155:1", -1); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("negative indexing agreements requested") + ); + } + + #[test] + fn parses_eip155_caip2_ids() { + assert_eq!(parse_eip155_caip2("eip155:1"), Some(1)); + assert_eq!(parse_eip155_caip2("eip155:42161"), Some(42161)); + assert_eq!(parse_eip155_caip2("eip155:"), None); + assert_eq!(parse_eip155_caip2("eip155:1x"), None); + assert_eq!(parse_eip155_caip2("solana:1"), None); + assert_eq!(parse_eip155_caip2(""), None); + } + + // -------- handle_record -------- + + type SetTargetCall = (Address, DeploymentId, ChainId, usize); + type ReassessCall = (IndexingRequestId, DeploymentId, ChainId, usize); + + /// A registry whose `set_indexing_target_candidates` pops the next scripted + /// response (erroring once the script runs out) and records its arguments. + #[derive(Clone)] + struct MockRegistry { + script: Arc>>>, + calls: Arc>>, + } + + impl MockRegistry { + /// `None` entries are errors; after the script is exhausted every call errors. + fn scripted(script: Vec>) -> Self { + Self { + script: Arc::new(Mutex::new(script)), + calls: Arc::new(Mutex::new(Vec::new())), + } + } + + fn returning(outcome: SetTargetOutcome) -> Self { + Self::scripted(vec![Some(outcome)]) + } + + fn calls(&self) -> Vec { + self.calls.lock().expect("poisoned").clone() + } + } + + #[async_trait] + impl IndexingRequestRegistry for MockRegistry { + async fn set_indexing_target_candidates( + &self, + requested_by: Address, + deployment_id: DeploymentId, + deployment_chain_id: ChainId, + num_candidates: usize, + ) -> RegistryResult { + self.calls.lock().expect("poisoned").push(( + requested_by, + deployment_id, + deployment_chain_id, + num_candidates, + )); + let mut script = self.script.lock().expect("poisoned"); + match if script.is_empty() { + None + } else { + script.remove(0) + } { + Some(outcome) => Ok(outcome), + None => Err(crate::registry::Error::NoRecordsUpdated), + } + } + + async fn get_all_indexing_requests(&self) -> RegistryResult> { + unimplemented!() + } + + async fn get_indexing_request_by_id( + &self, + _id: &IndexingRequestId, + ) -> RegistryResult> { + unimplemented!() + } + + async fn get_indexing_requests_by_deployment_id( + &self, + _deployment_id: &DeploymentId, + ) -> RegistryResult> { + unimplemented!() + } + + async fn get_open_indexing_requests_for_reassessment( + &self, + _min_age_seconds: i64, + _batch_size: i64, + ) -> RegistryResult> { + unimplemented!() + } + } + + /// A worker queue that records reassessment jobs, optionally failing the + /// first N pushes to exercise the queue-retry path. + #[derive(Clone)] + struct MockWorker { + reassessments: Arc>>, + failures_left: Arc>, + } + + impl MockWorker { + fn new() -> Self { + Self::failing_pushes(0) + } + + fn failing_pushes(failures: usize) -> Self { + Self { + reassessments: Arc::new(Mutex::new(Vec::new())), + failures_left: Arc::new(Mutex::new(failures)), + } + } + + fn reassessments(&self) -> Vec { + self.reassessments.lock().expect("poisoned").clone() + } + } + + #[async_trait] + impl WorkerQueue for MockWorker { + async fn send_indexing_agreement_proposal( + &self, + _candidate_url: Url, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + + async fn reassess_indexing_request( + &self, + indexing_request_id: IndexingRequestId, + deployment_id: DeploymentId, + deployment_chain_id: ChainId, + num_candidates: usize, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + { + let mut failures_left = self.failures_left.lock().expect("poisoned"); + if *failures_left > 0 { + *failures_left -= 1; + anyhow::bail!("scripted queue failure"); + } + } + self.reassessments.lock().expect("poisoned").push(( + indexing_request_id, + deployment_id, + deployment_chain_id, + num_candidates, + )); + Ok(JobId::default()) + } + + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + + async fn submit_offer( + &self, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _indexer_url: Url, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + } + + /// Drive `apply_target_with_retries` for the standard valid propose, with + /// no shutdown pending, returning its result and the captured events. + async fn run_apply( + registry: &MockRegistry, + worker: &MockWorker, + num_candidates: usize, + ) -> (bool, Vec) { + let events_capture = CapturingEventsProducer::new(); + let events: Arc = + Arc::new(events_capture.clone()); + let (_shutdown_tx, mut shutdown_rx) = watch::channel(false); + + let target = ProposedTarget { + event_id: "01912345-6789-7abc-def0-123456789abc".to_string(), + deployment_id: deployment_id!(QM_HASH), + deployment_chain_id: INDEXED_CHAIN_ID, + num_candidates, + }; + + let applied = apply_target_with_retries( + registry, + worker, + &events, + PROTOCOL_CHAIN_ID, + requester(), + TEST_MAX_CANDIDATES, + &target, + 0, + 0, + &mut shutdown_rx, + ) + .await; + + (applied, events_capture.events()) + } + + #[tokio::test] + async fn an_inserted_outcome_applies_emits_and_queues_reassessment() { + let registry = MockRegistry::returning(SetTargetOutcome::Inserted { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::new(); + + let (applied, events) = run_apply(®istry, &worker, 3).await; + + assert!(applied); + assert_eq!( + registry.calls(), + vec![(requester(), deployment_id!(QM_HASH), INDEXED_CHAIN_ID, 3)] + ); + + let reassessments = worker.reassessments(); + assert_eq!(reassessments.len(), 1); + assert_eq!(reassessments[0].1, deployment_id!(QM_HASH)); + assert_eq!(reassessments[0].2, INDEXED_CHAIN_ID); + assert_eq!(reassessments[0].3, 3); + + assert_eq!(events.len(), 1, "expected 1 request-received event"); + match &events[0] { + CapturedEvent::RequestReceived { + deployment, + chain_id, + event, + } => { + assert_eq!(*deployment, deployment_id!(QM_HASH)); + assert_eq!( + *chain_id, PROTOCOL_CHAIN_ID, + "the event carries the protocol chain id, not the indexed chain" + ); + assert_eq!(event.agreements_requested, 3); + } + other => panic!("expected RequestReceived, got {other:?}"), + } + } + + #[tokio::test] + async fn a_noop_outcome_applies_without_events_or_reassessment() { + let registry = MockRegistry::returning(SetTargetOutcome::NoOp { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::new(); + + let (applied, events) = run_apply(®istry, &worker, 3).await; + + assert!(applied); + assert!(worker.reassessments().is_empty(), "no-op must not reassess"); + assert!(events.is_empty(), "no-op must not emit events"); + } + + #[tokio::test] + async fn a_zero_count_cancellation_reassesses_to_zero_without_events() { + let registry = MockRegistry::returning(SetTargetOutcome::Canceled { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::new(); + + let (applied, events) = run_apply(®istry, &worker, 0).await; + + assert!(applied); + assert_eq!(registry.calls()[0].3, 0, "the registry saw the 0 count"); + + let reassessments = worker.reassessments(); + assert_eq!(reassessments.len(), 1); + assert_eq!( + reassessments[0].3, 0, + "reassessment with 0 drives the shrink that cancels agreements" + ); + assert!( + events.is_empty(), + "cancellation must not emit request-received" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_registry_failure_retries_the_whole_apply_until_it_succeeds() { + // 1st call errors (nothing committed), the retry lands the insert. + let registry = MockRegistry::scripted(vec![ + None, + Some(SetTargetOutcome::Inserted { + id: IndexingRequestId::new(), + }), + ]); + let worker = MockWorker::new(); + + let (applied, events) = run_apply(®istry, &worker, 3).await; + + assert!(applied); + assert_eq!(registry.calls().len(), 2, "the apply was retried once"); + assert_eq!(worker.reassessments().len(), 1); + assert_eq!(events.len(), 1); + } + + #[tokio::test(start_paused = true)] + async fn a_queue_failure_retries_only_the_push_never_the_registry() { + // If the retry re-ran the registry call, the outcome would be a no-op + // and the reassessment job would be silently lost. + let registry = MockRegistry::returning(SetTargetOutcome::Inserted { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::failing_pushes(2); + + let (applied, events) = run_apply(®istry, &worker, 3).await; + + assert!(applied); + assert_eq!( + registry.calls().len(), + 1, + "the committed row change must not be re-applied" + ); + assert_eq!( + worker.reassessments().len(), + 1, + "the push eventually landed" + ); + assert_eq!( + events.len(), + 1, + "the lifecycle event is emitted exactly once" + ); + } + + #[tokio::test] + async fn an_oversized_count_is_clamped_to_the_configured_maximum() { + let registry = MockRegistry::returning(SetTargetOutcome::Inserted { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::new(); + + let (applied, _) = run_apply(®istry, &worker, 5_000_000).await; + + assert!(applied); + assert_eq!( + registry.calls()[0].3, + TEST_MAX_CANDIDATES, + "the registry must see the clamped count, not the wire value" + ); + assert_eq!(worker.reassessments()[0].3, TEST_MAX_CANDIDATES); + } + + #[tokio::test] + async fn a_pending_shutdown_stops_retrying_without_applying() { + // Every registry call fails, so only shutdown can end the retry loop. + let registry = MockRegistry::scripted(vec![]); + let worker = MockWorker::new(); + + let events_capture = CapturingEventsProducer::new(); + let events: Arc = + Arc::new(events_capture.clone()); + let (shutdown_tx, mut shutdown_rx) = watch::channel(false); + shutdown_tx.send(true).expect("send shutdown"); + + let target = ProposedTarget { + event_id: "e".to_string(), + deployment_id: deployment_id!(QM_HASH), + deployment_chain_id: INDEXED_CHAIN_ID, + num_candidates: 3, + }; + + let applied = apply_target_with_retries( + ®istry, + &worker, + &events, + PROTOCOL_CHAIN_ID, + requester(), + TEST_MAX_CANDIDATES, + &target, + 0, + 0, + &mut shutdown_rx, + ) + .await; + + assert!(!applied, "shutdown must win over the retry loop"); + assert!(worker.reassessments().is_empty()); + assert!(events_capture.events().is_empty()); + } + + // -------- Redpanda-backed service test -------- + + use dipper_producer::{ + events::SubgraphIndexingAgreementsEventsEmitter, + kafka::{KafkaConfig, KafkaConsumerConfig, KafkaProducer}, + }; + + use crate::registry::RegistryProvider; + + fn redpanda_brokers() -> Option> { + match std::env::var("REDPANDA_BROKERS") { + Ok(value) if !value.trim().is_empty() => Some( + value + .split(',') + .map(|broker| broker.trim().to_string()) + .collect(), + ), + // REQUIRE_REDPANDA turns the silent skip into a failure, so CI + // cannot go green while accidentally testing nothing. + _ if std::env::var("REQUIRE_REDPANDA").is_ok() => { + panic!("REQUIRE_REDPANDA is set but REDPANDA_BROKERS is not") + } + _ => { + eprintln!("skipping Redpanda-backed test: REDPANDA_BROKERS is not set"); + None + } + } + } + + fn unique_topic() -> String { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock before unix epoch") + .as_nanos(); + format!("dipper.test.consumer.{}.{nanos}", std::process::id()) + } + + async fn create_topic(brokers: &[String], topic: &str) { + let client = rskafka::client::ClientBuilder::new(brokers.to_vec()) + .build() + .await + .expect("connect to broker"); + client + .controller_client() + .expect("controller client") + .create_topic(topic, 1, 1, 5_000) + .await + .expect("create topic"); + } + + fn propose_bytes(qm_hash: &str, count: i32) -> Vec { + studio::SubgraphIndexingRequestEvent { + event_id: "01912345-6789-7abc-def0-123456789abc".to_string(), + event_type: EVENT_TYPE_PROPOSE.to_string(), + event_version: "1.0".to_string(), + timestamp: "2026-08-24T10:30:00.123Z".to_string(), + subgraph_deployment_qm_hash: qm_hash.to_string(), + the_graph_network_caip2id: "eip155:42161".to_string(), + payload: Some( + studio::subgraph_indexing_request_event::Payload::SubgraphIndexingRequestPropose( + studio::SubgraphIndexingRequestPropose { + indexing_agreements_requested: count, + indexed_network_caip2id: "eip155:1".to_string(), + }, + ), + ), + } + .encode_to_vec() + } + + async fn wait_until(what: &str, mut check: impl AsyncFnMut() -> bool) { + for _ in 0..300 { + if check().await { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + panic!("timed out waiting for {what}"); + } + + #[tokio::test(flavor = "multi_thread")] + #[expect( + clippy::too_many_lines, + reason = "predates this lint; fix when next touched" + )] + async fn it_consumer_resumes_from_persisted_offsets_across_restarts() { + let Some(brokers) = redpanda_brokers() else { + return; + }; + + let topic = unique_topic(); + create_topic(&brokers, &topic).await; + + let temp_db = pgtemp::PgTempDB::new(); + let db = sqlx::Pool::connect(&temp_db.connection_uri()) + .await + .expect("connect to temp db"); + dipper_pgregistry::run_db_migrations(&db) + .await + .expect("run migrations"); + let provider = RegistryProvider::new(db.clone()); + + let requested_by: Address = "0x8f8c426f956876325b1e037c6eae9b189952994c" + .parse() + .expect("valid address"); + let deployment_a = deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"); + let deployment_b = deployment_id!("QmXbNL4EMkQ6DAPUcBjYSDXZJdpu1Kb1XkKvNvS8JdT7Hs"); + + let producer_config: KafkaConfig = serde_json::from_value(serde_json::json!({ + "brokers": brokers, + "topic": topic, + "partitions": 1, + })) + .expect("valid producer config"); + let producer = KafkaProducer::new(&producer_config) + .await + .expect("producer connects"); + + let consumer_config = IndexingRequestConsumerConfig { + enabled: true, + kafka: KafkaConsumerConfig { + brokers: brokers.clone(), + topic: topic.clone(), + sasl_mechanism: None, + sasl_username: None, + sasl_password: None, + tls_enabled: false, + tls_ca_cert_path: None, + connect_timeout_secs: 60, + }, + requested_by, + max_wait: Duration::from_secs(1), + fetch_max_bytes: 1_048_576, + }; + + let run_service = |worker: MockWorker| { + let events: Arc = + Arc::new(SubgraphIndexingAgreementsEventsEmitter::disabled()); + let (handle, service) = new(Ctx { + registry: provider.clone(), + worker_queue: worker, + events, + protocol_chain_id: 42161, + max_candidates: 10, + config: consumer_config.clone(), + }); + (handle, tokio::spawn(service)) + }; + + // A propose published before the consumer ever ran must still be + // picked up: a fresh partition starts from the earliest record. + producer + .send( + "eip155:42161/QmUzRg.../request", + &propose_bytes("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv", 2), + ) + .await + .expect("produce event A"); + + let worker_run_1 = MockWorker::new(); + let (handle, task) = run_service(worker_run_1.clone()); + wait_until("request A to be registered", async || { + !provider + .get_indexing_requests_by_deployment_id(&deployment_a) + .await + .expect("query requests") + .is_empty() + }) + .await; + wait_until("offset 1 to be persisted", async || { + provider + .get_kafka_consumer_offset(&topic, 0) + .await + .expect("query offset") + == Some(1) + }) + .await; + handle.stop().await; + task.await.expect("service task").expect("service result"); + + assert_eq!( + worker_run_1.reassessments().len(), + 1, + "run 1 applied exactly the 1 produced event" + ); + + // Published while the consumer is down; run 2 must pick it up from the + // persisted offset without re-applying event A. + producer + .send( + "eip155:42161/QmXbNL.../request", + &propose_bytes("QmXbNL4EMkQ6DAPUcBjYSDXZJdpu1Kb1XkKvNvS8JdT7Hs", 3), + ) + .await + .expect("produce event B"); + + let worker_run_2 = MockWorker::new(); + let (handle, task) = run_service(worker_run_2.clone()); + wait_until("request B to be registered", async || { + !provider + .get_indexing_requests_by_deployment_id(&deployment_b) + .await + .expect("query requests") + .is_empty() + }) + .await; + handle.stop().await; + task.await.expect("service task").expect("service result"); + + let run_2_reassessments = worker_run_2.reassessments(); + assert_eq!( + run_2_reassessments.len(), + 1, + "run 2 resumed past event A and applied only event B: {run_2_reassessments:?}" + ); + assert_eq!(run_2_reassessments[0].1, deployment_b); + assert_eq!(run_2_reassessments[0].3, 3); + + assert_eq!( + provider + .get_kafka_consumer_offset(&topic, 0) + .await + .expect("query offset"), + Some(2), + "both records are committed" + ); + assert_eq!( + provider + .get_indexing_requests_by_deployment_id(&deployment_a) + .await + .expect("query requests") + .len(), + 1, + "event A was applied exactly once across both runs" + ); + } +} diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index bb07b366..ddee7237 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -935,6 +935,34 @@ impl crate::network::service::chain_listener::ChainListenerStateRegistry for Reg } } +#[async_trait] +impl crate::network::service::indexing_request_consumer::KafkaConsumerOffsetRegistry + for RegistryProvider +{ + async fn get_kafka_consumer_offset( + &self, + topic: &str, + partition_id: i32, + ) -> RegistryResult> { + self.inner + .get_kafka_consumer_offset(topic, partition_id) + .await + .map_err(Into::into) + } + + async fn set_kafka_consumer_offset( + &self, + topic: &str, + partition_id: i32, + next_offset: i64, + ) -> RegistryResult<()> { + self.inner + .set_kafka_consumer_offset(topic, partition_id, next_offset) + .await + .map_err(Into::into) + } +} + #[cfg(test)] mod tests { use dipper_core::ids::IndexingAgreementId; diff --git a/bin/dipper-service/src/set_indexing_target.rs b/bin/dipper-service/src/set_indexing_target.rs new file mode 100644 index 00000000..d5274936 --- /dev/null +++ b/bin/dipper-service/src/set_indexing_target.rs @@ -0,0 +1,183 @@ +//! Applies a set-indexing-target request end to end: the registry upsert, the +//! request-received lifecycle event, and the follow-up reassessment job. Both +//! front doors share it: the admin RPC handler and the Kafka request consumer. + +use std::sync::Arc; + +use dipper_core::ids::IndexingRequestId; +use dipper_producer::{events::SubgraphIndexingAgreementEventsProducer, proto}; +use thegraph_core::{ + DeploymentId, + alloy::primitives::{Address, ChainId}, +}; + +use crate::{ + registry::{IndexingRequestRegistry, SetTargetOutcome}, + worker::service::{JobPriority, WorkerQueue}, +}; + +/// A set-indexing-target request, independent of which front door it came in +/// through. +pub struct SetIndexingTarget { + /// Who asked (recovered RPC signer, or the consumer's configured identity). + pub requested_by: Address, + /// The subgraph deployment to index. + pub deployment_id: DeploymentId, + /// The chain the deployment indexes (its data source), keying the request. + pub deployment_chain_id: ChainId, + /// The target number of indexers; 0 cancels the request. + pub num_candidates: usize, + /// Priority for the follow-up reassessment job. + pub priority: JobPriority, +} + +/// Errors from applying a set-indexing-target request. Both variants are +/// already logged with full context when returned. +#[derive(Debug, thiserror::Error)] +pub enum ApplyError { + /// The registry upsert failed; nothing was changed or queued. + #[error("failed to set indexing target candidates")] + Registry(#[source] crate::registry::Error), + + /// The row change is committed but the reassessment job was not queued. + /// Retry the queue push itself, not the whole apply: a repeated apply + /// lands on the registry's no-op path and never queues the job. + #[error("failed to queue reassessment for indexing request {id}")] + QueueReassess { + id: IndexingRequestId, + #[source] + source: anyhow::Error, + }, +} + +/// Upserts the indexing request, emits the request-received lifecycle event on +/// a genuinely new request, and queues reassessment when the row changed. +/// Returns the request id, or `None` when there was nothing to act on. +#[expect( + clippy::cast_possible_truncation, + clippy::cast_possible_wrap, + clippy::cognitive_complexity, + reason = "predates this lint; fix when next touched" +)] +pub async fn apply_set_indexing_target( + registry: &R, + worker: &W, + events: &Arc, + the_graph_network: ChainId, + request: SetIndexingTarget, +) -> Result, ApplyError> +where + R: IndexingRequestRegistry + Send + Sync, + W: WorkerQueue + Send + Sync, +{ + let SetIndexingTarget { + requested_by, + deployment_id, + deployment_chain_id, + num_candidates, + priority, + } = request; + + let outcome = match registry + .set_indexing_target_candidates( + requested_by, + deployment_id, + deployment_chain_id, + num_candidates, + ) + .await + { + Ok(outcome) => outcome, + Err(err) => { + tracing::error!(error=?err, "Failed to set indexing target candidates"); + return Err(ApplyError::Registry(err)); + } + }; + + // Translate the outcome into the appropriate follow-up worker job and the + // request id to hand back. + let (id_opt, reassess_count): (Option, Option) = match outcome { + SetTargetOutcome::Inserted { id } => { + tracing::info!( + indexing_request_id = %id, + %requested_by, + %deployment_id, + chain_id = %deployment_chain_id, + num_candidates, + "Inserted new indexing request" + ); + + // Only `Inserted` is a genuinely new request, so only it emits the + // lifecycle event; `the_graph_network` is the protocol network, + // not the deployment's data-source `chain_id`. + events.produce_subgraph_indexing_agreement_request_received( + deployment_id, + the_graph_network, + proto::SubgraphIndexingAgreementRequestReceived { + agreements_requested: num_candidates as i32, + }, + ); + + (Some(id), Some(num_candidates)) + } + SetTargetOutcome::Updated { + id, + new_num_candidates, + } => { + tracing::info!( + indexing_request_id = %id, + %requested_by, + %deployment_id, + chain_id = %deployment_chain_id, + num_candidates = new_num_candidates, + "Updated num_candidates on open indexing request" + ); + (Some(id), Some(new_num_candidates)) + } + SetTargetOutcome::NoOp { id } => { + tracing::debug!( + indexing_request_id = %id, + "Set target candidates is a no-op (count unchanged)" + ); + (Some(id), None) + } + SetTargetOutcome::Canceled { id } => { + tracing::info!( + indexing_request_id = %id, + %requested_by, + %deployment_id, + chain_id = %deployment_chain_id, + "Canceled indexing request (target candidates set to zero)" + ); + (Some(id), Some(0)) + } + SetTargetOutcome::NoOpAlreadyEmpty => { + tracing::warn!( + %requested_by, + %deployment_id, + chain_id = %deployment_chain_id, + "set_indexing_target_candidates with num_candidates=0 against a key with no open request \ + - nothing to cancel" + ); + (None, None) + } + }; + + // Queue reassessment if the row changed: it diffs the IISA target group of + // size `num_candidates` against the active agreements and grows or shrinks + // to match; 0 shrinks to nothing, cancelling every agreement on-chain. + if let (Some(id), Some(count)) = (id_opt, reassess_count) + && let Err(err) = worker + .reassess_indexing_request(id, deployment_id, deployment_chain_id, count, priority) + .await + { + tracing::error!( + indexing_request_id = %id, + error = ?err, + "Failed to queue task: 'reassess_indexing_request'" + ); + return Err(ApplyError::QueueReassess { id, source: err }); + } + + Ok(id_opt) +} diff --git a/dipper-pgregistry/migrations/20260824000000_add_kafka_consumer_offsets.sql b/dipper-pgregistry/migrations/20260824000000_add_kafka_consumer_offsets.sql new file mode 100644 index 00000000..f419f87c --- /dev/null +++ b/dipper-pgregistry/migrations/20260824000000_add_kafka_consumer_offsets.sql @@ -0,0 +1,22 @@ +-- Kafka consumer offsets +-- +-- The dipper consumes subgraph indexing request events that Studio produces on +-- a Redpanda topic. The Kafka client in use (rskafka) has no consumer groups, +-- so the broker cannot store consumer progress; the dipper records it here +-- instead, after each record is processed, and resumes from it on restart. + +CREATE TABLE IF NOT EXISTS dipper_kafka_consumer_offsets ( + -- Topic name; the topic is deploy-time config, so it is part of the key. + topic TEXT NOT NULL, + -- Kafka partition id within the topic + partition_id INT NOT NULL, + -- The next offset to fetch: 1 past the last fully processed record + next_offset BIGINT NOT NULL, + -- Timestamps for auditing + created_at TIMESTAMPTZ NOT NULL DEFAULT timezone('UTC', now()), + updated_at TIMESTAMPTZ NOT NULL DEFAULT timezone('UTC', now()), + PRIMARY KEY (topic, partition_id) +); + +COMMENT ON TABLE dipper_kafka_consumer_offsets IS 'Per-partition consumer progress for Kafka topics the dipper consumes; written after processing so delivery is at-least-once'; +COMMENT ON COLUMN dipper_kafka_consumer_offsets.next_offset IS 'The next offset to fetch: 1 past the last fully processed record'; diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 3a36f0bf..d5ea79be 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -2320,6 +2320,61 @@ impl PgRegistry { Ok(()) } + // ========================================================================= + // Kafka consumer offset operations + // ========================================================================= + + /// Get the next offset to fetch for a topic partition. + /// Returns `None` if no offset was recorded yet (first run). + pub async fn get_kafka_consumer_offset( + &self, + topic: &str, + partition_id: i32, + ) -> Result, Error> { + let row: Option<(i64,)> = sqlx::query_as( + r#" + SELECT next_offset + FROM dipper_kafka_consumer_offsets + WHERE topic = $1 AND partition_id = $2 + "#, + ) + .bind(topic) + .bind(partition_id) + .fetch_optional(&self.pool) + .await?; + + Ok(row.map(|(next_offset,)| next_offset)) + } + + /// Record the next offset to fetch for a topic partition (upsert). Called + /// after a record is fully processed, so a crash between processing and + /// this write redelivers the record (at-least-once). + pub async fn set_kafka_consumer_offset( + &self, + topic: &str, + partition_id: i32, + next_offset: i64, + ) -> Result<(), Error> { + sqlx::query( + r#" + INSERT INTO dipper_kafka_consumer_offsets + (topic, partition_id, next_offset, updated_at) + VALUES ($1, $2, $3, timezone('UTC', now())) + ON CONFLICT (topic, partition_id) + DO UPDATE SET + next_offset = EXCLUDED.next_offset, + updated_at = EXCLUDED.updated_at + "#, + ) + .bind(topic) + .bind(partition_id) + .bind(next_offset) + .execute(&self.pool) + .await?; + + Ok(()) + } + // -- Pending cancellations -- /// Register a new agreement and record a pending cancellation in a single diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index a09aa096..b3a7586f 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -3507,6 +3507,75 @@ async fn count_created_agreements_by_indexer_counts_only_created() { assert_eq!(global, 3, "global counts only the 3 Created rows"); } +#[tokio::test] +async fn kafka_consumer_offsets_roundtrip_and_upsert() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + let registry = PgRegistry::new(db); + + let topic = "studio.subgraph.indexing.requests"; + + //* When / Then - no offset recorded yet + let offset = registry + .get_kafka_consumer_offset(topic, 0) + .await + .expect("get offset"); + assert_eq!(offset, None, "a fresh partition has no recorded offset"); + + //* When / Then - first write inserts + registry + .set_kafka_consumer_offset(topic, 0, 5) + .await + .expect("set offset"); + let offset = registry + .get_kafka_consumer_offset(topic, 0) + .await + .expect("get offset"); + assert_eq!(offset, Some(5)); + + //* When / Then - second write updates in place + registry + .set_kafka_consumer_offset(topic, 0, 42) + .await + .expect("update offset"); + let offset = registry + .get_kafka_consumer_offset(topic, 0) + .await + .expect("get offset"); + assert_eq!(offset, Some(42)); + + //* When / Then - partitions and topics are independent keys + registry + .set_kafka_consumer_offset(topic, 7, 1) + .await + .expect("set offset on another partition"); + registry + .set_kafka_consumer_offset("another.topic", 0, 9) + .await + .expect("set offset on another topic"); + assert_eq!( + registry + .get_kafka_consumer_offset(topic, 0) + .await + .expect("get offset"), + Some(42) + ); + assert_eq!( + registry + .get_kafka_consumer_offset(topic, 7) + .await + .expect("get offset"), + Some(1) + ); + assert_eq!( + registry + .get_kafka_consumer_offset("another.topic", 0) + .await + .expect("get offset"), + Some(9) + ); +} + #[tokio::test] async fn count_created_agreements_by_indexer_counts_offers_being_withdrawn() { //* Given diff --git a/dipper-producer/Cargo.toml b/dipper-producer/Cargo.toml index 02dc06d6..505fcccb 100644 --- a/dipper-producer/Cargo.toml +++ b/dipper-producer/Cargo.toml @@ -20,6 +20,10 @@ tracing.workspace = true uuid = { workspace = true, features = ["v7"] } webpki-roots = "1.0.8" +[dev-dependencies] +serde_json.workspace = true +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } + # Protobuf code generation dependencies # Run with: RUSTFLAGS="--cfg gen_event_proto" cargo check -p dipper-producer [target.'cfg(gen_event_proto)'.build-dependencies] diff --git a/dipper-producer/README.md b/dipper-producer/README.md index 32ee5100..6e6b5ccf 100644 --- a/dipper-producer/README.md +++ b/dipper-producer/README.md @@ -1,16 +1,18 @@ # dipper-producer -This crate contains protobuf definitions for dipper indexer agreement event streaming. -The generated Rust bindings are committed to the repository and only need to be regenerated when the `.proto` file changes. +This crate contains the protobuf definitions and Kafka plumbing for dipper event streaming: the producer side for the agreement lifecycle events the dipper emits, and the consumer side for the subgraph indexing request events Studio emits. The generated Rust bindings are committed to the repository and only need to be regenerated when a `.proto` file changes. ## Protobuf Generation -The build script uses a configuration flag `gen_event_proto` that enables protobuf code generation via `prost-build`. When enabled, the build script compiles `proto/indexing-agreement-events.proto` into Rust types under `src/proto/`. +The build script uses a configuration flag `gen_event_proto` that enables protobuf code generation via `prost-build`. When enabled, the build script compiles the schemas under `proto/` into Rust types under `src/proto/`: + +- `proto/indexing-agreement-events.proto`, owned by this repo, generates `src/proto/dipper.subgraph.indexing.agreement.events.v1.rs`. +- `proto/subgraph-indexing-request-events.proto`, vendored from the subgraph-studio repo (`packages/shared/src/helpers/dips/proto/SubgraphIndexingRequest.proto`), generates `src/proto/studio.subgraph.indexing.requests.events.v1.rs`. When Studio's copy changes, re-vendor it here and regenerate. To regenerate protobuf bindings, run: ```bash -just gen-indexing-agreement-events-proto +just gen-event-protos ``` Or using the full `cargo` command: @@ -18,5 +20,3 @@ Or using the full `cargo` command: ```bash RUSTFLAGS="--cfg gen_event_proto" cargo check -p dipper-producer ``` - -This will regenerate `src/proto/dipper.subgraph.indexing.agreement.events.v1.rs` from `proto/indexing-agreement-events.proto`. diff --git a/dipper-producer/build.rs b/dipper-producer/build.rs index c894d68b..c4c5f531 100644 --- a/dipper-producer/build.rs +++ b/dipper-producer/build.rs @@ -7,7 +7,13 @@ fn main() -> Result<(), Box> { config.out_dir("src/proto"); config.protoc_arg("--experimental_allow_proto3_optional"); - config.compile_protos(&["proto/indexing-agreement-events.proto"], &["proto/"])?; + config.compile_protos( + &[ + "proto/indexing-agreement-events.proto", + "proto/subgraph-indexing-request-events.proto", + ], + &["proto/"], + )?; // Instruct cargo to rerun this build script if any of the proto files change println!("cargo:rerun-if-changed=proto"); diff --git a/dipper-producer/proto/subgraph-indexing-request-events.proto b/dipper-producer/proto/subgraph-indexing-request-events.proto new file mode 100644 index 00000000..d9bc7233 --- /dev/null +++ b/dipper-producer/proto/subgraph-indexing-request-events.proto @@ -0,0 +1,88 @@ +// Subgraph Indexing Requests Events Protocol Buffer Schema +// +// Vendored copy of the schema owned by Subgraph Studio: +// repo: edgeandnode/subgraph-studio +// path: packages/shared/src/helpers/dips/proto/SubgraphIndexingRequest.proto +// commit: a8140613 +// +// Studio produces these messages on a Redpanda topic; the dipper consumes them +// and turns each propose event into a set_indexing_target_candidates call. +// When Studio's copy changes, re-vendor it here and regenerate the bindings +// (see the crate README). +// +// Local divergence from the vendored source, kept intentionally: +// - SubgraphIndexingRequestPropose.indexed_network_caip2id (field 2) is the +// network the subgraph indexes, which the dipper needs to key the indexing +// request and Studio has at every call site. It is a backward-compatible +// proto3 addition agreed as the preferred fix for that gap; drop this note +// once Studio's copy carries the field. +// +// Event flow: +// 1. subgraph.indexing.request.propose - Emitted when the developer sends a request to have their Subgraph indexed +// 2. subgraph.indexing.agreements.terminate - Emitted when the developer sends a request to terminate active Indexing agreements +// +// Partition key format: {the_graph_network_caip2id}/{subgraph_deployment_qm_hash}/{request/terminate} +// +// - the_graph_network_caip2id -> CAIP2 ID of The Graph network where the Subgraph is published +// - subgraph_deployment_qm_hash -> Qm hash of the Subgraph deployment requesting to be indexed +// Example: eip155:42161/QmTXzATwNfgGVukV1fX2T6xw9f6LAYRVWpsdXyRWzUR2H9/request + +syntax = "proto3"; + +package studio.subgraph.indexing.requests.events.v1; + +// SubgraphIndexingRequestEvent is the envelope that wraps all the Subgraph Indexing Request event types. +// +// Provides: +// - event_id -> unique identifier (uuid) +// - event_type -> event type discrimination +// - event_version -> versioning for schema evolution awareness to the consumer +// - timestamp -> when the event occurred, useful for ordering/debugging +// - subgraph_deployment_qm_hash -> Qm hash of the Subgraph deployment +// - the_graph_network_caip2id -> CAIP-2 chain id of The Graph protocol network the Subgraph Deployment is published to (e.g. eip155:42161) +// - payload -> the event payload, determined by the event_type +message SubgraphIndexingRequestEvent { + // Unique event identifier. + // Format: UUID v7 (time-ordered) for natural chronological sorting. + // Example: "01912345-6789-7abc-def0-123456789abc" + string event_id = 1; + + // Event type discriminator for routing and filtering. + // Values: "subgraph.indexing.request.propose", "subgraph.indexing.agreements.terminate" + string event_type = 2; + + // Schema version for forward compatibility. + // Consumers should handle unknown fields gracefully. + // Current version: "1.0" + string event_version = 3; + + // Event timestamp in RFC 3339 format. + // Example: "2024-01-15T10:30:00.123Z" + string timestamp = 4; + + // Qm hash of the Subgraph deployment with a submitted indexing agreement event. + string subgraph_deployment_qm_hash = 5; + + // CAIP-2 chain id of The Graph protocol network where the Subgraph was published to. + // Format: "eip155:{chain_id}". Examples: "eip155:42161" (arbitrum), "eip155:421614" (arbitrum-sepolia) + string the_graph_network_caip2id = 6; + + // Event payload - exactly one of the specific event types. + // Use event_type field to determine which payload is present. + oneof payload { + SubgraphIndexingRequestPropose subgraph_indexing_request_propose = 7; + } +} + +// SubgraphIndexingRequestPropose is emitted when the Subgraph developer initiates a request to have the Subgraph indexed +// +// Event type: subgraph.indexing.request.propose +message SubgraphIndexingRequestPropose { + // Number of requested Indexing agreements to find + int32 indexing_agreements_requested = 1; + + // CAIP-2 chain id of the network the Subgraph indexes (its data source), + // distinct from the_graph_network_caip2id on the envelope. + // Format: "eip155:{chain_id}". Example: "eip155:1" for a subgraph indexing Ethereum mainnet. + string indexed_network_caip2id = 2; +} diff --git a/dipper-producer/src/events/subgraph_indexing_agreements_events_emitter.rs b/dipper-producer/src/events/subgraph_indexing_agreements_events_emitter.rs index 2c1a45a1..7dd2ed9e 100644 --- a/dipper-producer/src/events/subgraph_indexing_agreements_events_emitter.rs +++ b/dipper-producer/src/events/subgraph_indexing_agreements_events_emitter.rs @@ -711,6 +711,7 @@ mod tests { sasl_password: None, tls_enabled: false, tls_ca_cert_path: None, + connect_timeout_secs: 60, } } diff --git a/dipper-producer/src/kafka.rs b/dipper-producer/src/kafka.rs index 25714154..789bd36a 100644 --- a/dipper-producer/src/kafka.rs +++ b/dipper-producer/src/kafka.rs @@ -1,27 +1,13 @@ -//! Kafka client for subgraph indexing agreement event streaming. -//! -//! This module provides a Kafka producer for emitting subgraph indexing agreement lifecycle events -//! to a kafka topic -//! -//! Events are encoded using Protocol Buffers for compact, schema-enforced messages. -//! -//! # Example -//! -//! ```ignore -//! use dipper_producer::kafka::{KafkaConfig, KafkaProducer}; -//! -//! let config = KafkaConfig { -//! brokers: vec!["localhost:9092".to_string()], -//! topic: "dipper.subgraph.indexing.agreement.events".to_string(), -//! partitions: 16, -//! }; -//! -//! let producer = KafkaProducer::new(&config).await?; -//! -//! // Send an event with partition key and protobuf payload -//! producer.send("QmT329Bej8AwSLahmgnmi6fdYkj3rorYAcCes45gDv9aJ4", &encoded_event).await?; -//! ``` +//! Kafka clients for dipper event streaming: a producer for the agreement +//! lifecycle events the dipper emits and a consumer for the indexing request +//! events Studio emits. Events are Protocol Buffers encoded. +mod connection; +mod consumer; mod producer; +pub use connection::ConnectionError; +pub use consumer::{ConsumerError, KafkaConsumer, KafkaConsumerConfig}; pub use producer::{Error, KafkaConfig, KafkaProducer}; +// Re-exported so callers name fetch positions without depending on rskafka. +pub use rskafka::client::partition::OffsetAt; diff --git a/dipper-producer/src/kafka/connection.rs b/dipper-producer/src/kafka/connection.rs new file mode 100644 index 00000000..db407626 --- /dev/null +++ b/dipper-producer/src/kafka/connection.rs @@ -0,0 +1,224 @@ +//! Broker connection helpers shared by the Kafka producer and consumer: +//! SASL/TLS setup and the client bootstrap built from them. + +use std::{ + path::Path, + sync::{Arc, Once}, +}; + +use rskafka::client::{Client, ClientBuilder, Credentials, SaslConfig}; +use rustls::ClientConfig; + +static RUSTLS_CRYPTO_PROVIDER: Once = Once::new(); + +/// Broker connection parameters, borrowed from the producer or consumer config. +pub(crate) struct ConnectOptions<'a> { + pub brokers: &'a [String], + pub sasl_mechanism: Option<&'a str>, + pub sasl_username: Option<&'a str>, + pub sasl_password: Option<&'a str>, + pub tls_enabled: bool, + pub tls_ca_cert_path: Option<&'a Path>, +} + +/// Connects a Kafka client with the given SASL/TLS settings. +pub(crate) async fn connect(opts: ConnectOptions<'_>) -> Result { + if opts.brokers.is_empty() { + return Err(ConnectionError::MissingBrokers); + } + + let mut builder = ClientBuilder::new(opts.brokers.to_vec()); + + if let Some(mechanism_str) = opts.sasl_mechanism { + let mechanism: SaslMechanism = mechanism_str.parse()?; + let sasl_config = build_sasl_config(mechanism, opts.sasl_username, opts.sasl_password)?; + builder = builder.sasl_config(sasl_config); + } + + if opts.tls_enabled { + let tls_config = build_tls_config(opts.tls_ca_cert_path)?; + builder = builder.tls_config(tls_config); + } + + builder.build().await.map_err(ConnectionError::Connection) +} + +/// Builds SASL configuration from the provided mechanism and credentials. +pub(crate) fn build_sasl_config( + mechanism: SaslMechanism, + username: Option<&str>, + password: Option<&str>, +) -> Result { + let username = username.ok_or(ConnectionError::MissingSaslUsername)?; + let password = password.ok_or(ConnectionError::MissingSaslPassword)?; + + let credentials = Credentials::new(username.to_string(), password.to_string()); + + Ok(match mechanism { + SaslMechanism::Plain => SaslConfig::Plain(credentials), + SaslMechanism::ScramSha256 => SaslConfig::ScramSha256(credentials), + SaslMechanism::ScramSha512 => SaslConfig::ScramSha512(credentials), + }) +} + +/// Builds TLS configuration. A custom CA certificate path makes the client +/// trust that CA for broker verification; otherwise system roots are used. +pub(crate) fn build_tls_config( + ca_cert_path: Option<&Path>, +) -> Result, ConnectionError> { + install_rustls_crypto_provider(); + + let root_store = match ca_cert_path { + Some(path) => { + let ca_pem = + fs_err::read(path).map_err(|e| ConnectionError::TlsCaCert { source: e })?; + let mut reader = std::io::BufReader::new(&ca_pem[..]); + let certs: Vec<_> = rustls_pemfile::certs(&mut reader) + .collect::>() + .map_err(|e| ConnectionError::TlsCaCert { source: e })?; + + let mut store = rustls::RootCertStore::empty(); + for cert in certs { + store.add(cert).map_err(|e| ConnectionError::TlsCaCert { + source: std::io::Error::new(std::io::ErrorKind::InvalidData, e), + })?; + } + store + } + None => rustls::RootCertStore { + roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(), + }, + }; + + let tls_config = ClientConfig::builder() + .with_root_certificates(root_store) + .with_no_client_auth(); + + Ok(Arc::new(tls_config)) +} + +fn install_rustls_crypto_provider() { + RUSTLS_CRYPTO_PROVIDER.call_once(|| { + // Necessary for the Kafka client: it builds a Rustls TLS config directly, + // so install a provider before `ClientConfig::builder()` tries to infer one. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + }); +} + +/// Errors that can occur while establishing a broker connection. +#[derive(Debug, thiserror::Error)] +pub enum ConnectionError { + /// Failed to connect to Kafka brokers + #[error("failed to connect to Kafka brokers")] + Connection(#[source] rskafka::client::error::Error), + + /// The brokers list is empty + #[error("brokers must list at least 1 broker address")] + MissingBrokers, + + /// Unsupported SASL mechanism + #[error("unsupported SASL mechanism '{0}', supported: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512")] + UnsupportedSaslMechanism(String), + + /// Missing SASL username + #[error("sasl_username is required when sasl_mechanism is set")] + MissingSaslUsername, + + /// Missing SASL password + #[error("sasl_password is required when sasl_mechanism is set")] + MissingSaslPassword, + + /// Failed to load TLS CA certificate + #[error("failed to load TLS CA certificate")] + TlsCaCert { + #[source] + source: std::io::Error, + }, +} + +/// Supported SASL authentication mechanisms. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum SaslMechanism { + Plain, + ScramSha256, + ScramSha512, +} + +impl std::str::FromStr for SaslMechanism { + type Err = ConnectionError; + + fn from_str(s: &str) -> Result { + match s.to_uppercase().as_str() { + "PLAIN" => Ok(Self::Plain), + "SCRAM-SHA-256" => Ok(Self::ScramSha256), + "SCRAM-SHA-512" => Ok(Self::ScramSha512), + _ => Err(ConnectionError::UnsupportedSaslMechanism(s.to_string())), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn should_parse_sasl_mechanism_correctly() { + // Case-insensitive parsing + assert_eq!( + "PLAIN".parse::().unwrap(), + SaslMechanism::Plain + ); + assert_eq!( + "plain".parse::().unwrap(), + SaslMechanism::Plain + ); + assert_eq!( + "SCRAM-SHA-256".parse::().unwrap(), + SaslMechanism::ScramSha256 + ); + assert_eq!( + "scram-sha-256".parse::().unwrap(), + SaslMechanism::ScramSha256 + ); + assert_eq!( + "SCRAM-SHA-512".parse::().unwrap(), + SaslMechanism::ScramSha512 + ); + + // Unsupported mechanism + assert!("GSSAPI".parse::().is_err()); + } + + #[test] + fn should_build_sasl_plain_sasl_config() { + let result = build_sasl_config(SaslMechanism::Plain, Some("user"), Some("pass")); + assert!(matches!(result, Ok(SaslConfig::Plain(_)))); + } + + #[test] + fn should_build_scram_sha_256_sasl_config() { + let result = build_sasl_config(SaslMechanism::ScramSha256, Some("user"), Some("pass")); + assert!(matches!(result, Ok(SaslConfig::ScramSha256(_)))); + } + + #[test] + fn should_build_sasl_sha_512_sasl_config() { + let result = build_sasl_config(SaslMechanism::ScramSha512, Some("user"), Some("pass")); + assert!(matches!(result, Ok(SaslConfig::ScramSha512(_)))); + } + + #[test] + fn should_throw_err_on_missing_credentials() { + // Missing username + assert!(matches!( + build_sasl_config(SaslMechanism::Plain, None, Some("pass")), + Err(ConnectionError::MissingSaslUsername) + )); + + // Missing password + assert!(matches!( + build_sasl_config(SaslMechanism::Plain, Some("user"), None), + Err(ConnectionError::MissingSaslPassword) + )); + } +} diff --git a/dipper-producer/src/kafka/consumer.rs b/dipper-producer/src/kafka/consumer.rs new file mode 100644 index 00000000..38fa767e --- /dev/null +++ b/dipper-producer/src/kafka/consumer.rs @@ -0,0 +1,390 @@ +//! Kafka consumer for the subgraph indexing request events Studio produces. +//! Deliberately a thin fetch layer: rskafka has no consumer groups, so offset +//! tracking belongs to the caller (the dipper persists offsets in its own DB). + +use std::{collections::BTreeMap, path::PathBuf, sync::Arc, time::Duration}; + +use rskafka::{ + client::{ + Client, + partition::{OffsetAt, PartitionClient, UnknownTopicHandling}, + }, + record::RecordAndOffset, +}; + +use super::connection::{self, ConnectOptions, ConnectionError}; + +/// Kafka consumer configuration. +#[derive(Clone, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct KafkaConsumerConfig { + /// Kafka broker addresses. + pub brokers: Vec, + /// Kafka topic to consume. Deliberately has no default: a consumer pointed + /// at a missing or misnamed topic must fail at startup, not idle on nothing. + pub topic: String, + /// SASL authentication mechanism (e.g., "PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"). + #[serde(default)] + pub sasl_mechanism: Option, + /// SASL username. + #[serde(default)] + pub sasl_username: Option, + /// SASL password. + #[serde(default)] + pub sasl_password: Option, + /// Enable TLS encryption. + #[serde(default)] + pub tls_enabled: bool, + /// Path to a PEM-encoded CA certificate file for TLS verification. + #[serde(default)] + pub tls_ca_cert_path: Option, + /// Seconds allowed for the initial connect and topic discovery (default: + /// 60). Load-bearing: the underlying client retries an unreachable broker + /// forever, so without this bound `connect` would never return. + #[serde(default = "default_connect_timeout_secs")] + pub connect_timeout_secs: u64, +} + +/// Default number of seconds allowed for connect and topic discovery. +pub fn default_connect_timeout_secs() -> u64 { + 60 +} + +// Manual impl instead of derive: the service logs the whole config with Debug +// formatting at startup, so the SASL password must never reach the output. +impl std::fmt::Debug for KafkaConsumerConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("KafkaConsumerConfig") + .field("brokers", &self.brokers) + .field("topic", &self.topic) + .field("sasl_mechanism", &self.sasl_mechanism) + .field("sasl_username", &self.sasl_username) + .field( + "sasl_password", + &self.sasl_password.as_ref().map(|_| ""), + ) + .field("tls_enabled", &self.tls_enabled) + .field("tls_ca_cert_path", &self.tls_ca_cert_path) + .field("connect_timeout_secs", &self.connect_timeout_secs) + .finish() + } +} + +/// Kafka consumer bound to a single topic, with a partition client per +/// discovered partition. Thread-safe; share across tasks via `Arc`. +pub struct KafkaConsumer { + client: Client, + topic: String, + partition_clients: BTreeMap>, +} + +impl KafkaConsumer { + /// Margin added on top of a fetch's `max_wait_ms` before the whole call is + /// abandoned, covering broker round-trip and retry time. + const FETCH_TIMEOUT_MARGIN: Duration = Duration::from_secs(30); + + /// Timeout for offset queries, which carry no server-side wait. + const OFFSET_TIMEOUT: Duration = Duration::from_secs(30); + + /// Connects to the brokers and binds to the configured topic. Partitions + /// are discovered from broker metadata, so there is no partition count to + /// configure; a topic the credentials cannot see is an error. The whole + /// sequence is bounded by `connect_timeout_secs`, since the underlying + /// client would otherwise retry an unreachable broker forever. + pub async fn connect(config: &KafkaConsumerConfig) -> Result { + tokio::time::timeout( + Duration::from_secs(config.connect_timeout_secs), + Self::connect_inner(config), + ) + .await + .map_err(|_| ConsumerError::Timeout)? + } + + async fn connect_inner(config: &KafkaConsumerConfig) -> Result { + let client = connection::connect(ConnectOptions { + brokers: &config.brokers, + sasl_mechanism: config.sasl_mechanism.as_deref(), + sasl_username: config.sasl_username.as_deref(), + sasl_password: config.sasl_password.as_deref(), + tls_enabled: config.tls_enabled, + tls_ca_cert_path: config.tls_ca_cert_path.as_deref(), + }) + .await?; + + let partitions = discover_partitions(&client, &config.topic).await?; + + let mut partition_clients = BTreeMap::new(); + for partition in partitions { + let partition_client = client + .partition_client(&config.topic, partition, UnknownTopicHandling::Error) + .await + .map_err(ConsumerError::PartitionClient)?; + partition_clients.insert(partition, Arc::new(partition_client)); + } + + Ok(Self { + client, + topic: config.topic.clone(), + partition_clients, + }) + } + + /// The topic this consumer is bound to. + pub fn topic(&self) -> &str { + &self.topic + } + + /// The partition ids discovered for the topic, in ascending order. + pub fn partitions(&self) -> Vec { + self.partition_clients.keys().copied().collect() + } + + /// Re-reads broker metadata and returns the topic's current partition + /// count, so callers can notice a topic growing partitions after connect + /// (this consumer keeps serving only the set discovered at connect). + pub async fn current_partition_count(&self) -> Result { + tokio::time::timeout( + Self::OFFSET_TIMEOUT, + discover_partitions(&self.client, &self.topic), + ) + .await + .map_err(|_| ConsumerError::Timeout)? + .map(|partitions| partitions.len()) + } + + /// Fetches records from one partition starting at `offset`, waiting up to + /// `max_wait_ms` for data to arrive. Returns the records (with their + /// offsets) and the partition's current high watermark. + pub async fn fetch( + &self, + partition: i32, + offset: i64, + max_bytes: i32, + max_wait_ms: i32, + ) -> Result<(Vec, i64), ConsumerError> { + let partition_client = self.partition_client(partition)?; + // A negative wait means no wait. + let max_wait = Duration::from_millis(u64::try_from(max_wait_ms).unwrap_or(0)); + let timeout = max_wait + Self::FETCH_TIMEOUT_MARGIN; + + // rskafka encodes the range's exclusive end minus 1 as the wire-level + // max_bytes, so widen by 1 to request the full `max_bytes` budget. + tokio::time::timeout( + timeout, + partition_client.fetch_records(offset, 1..max_bytes.saturating_add(1), max_wait_ms), + ) + .await + .map_err(|_| ConsumerError::Timeout)? + .map_err(ConsumerError::Fetch) + } + + /// Queries one partition's earliest or latest offset. + pub async fn offset(&self, partition: i32, at: OffsetAt) -> Result { + let partition_client = self.partition_client(partition)?; + + tokio::time::timeout(Self::OFFSET_TIMEOUT, partition_client.get_offset(at)) + .await + .map_err(|_| ConsumerError::Timeout)? + .map_err(ConsumerError::Offset) + } + + fn partition_client(&self, partition: i32) -> Result<&Arc, ConsumerError> { + self.partition_clients + .get(&partition) + .ok_or(ConsumerError::UnknownPartition { + partition, + topic: self.topic.clone(), + }) + } +} + +/// Lists the topic's partitions from broker metadata; a topic the broker does +/// not report (missing, or invisible to the credentials) is an error. +async fn discover_partitions( + client: &Client, + topic: &str, +) -> Result, ConsumerError> { + let topics = client + .list_topics() + .await + .map_err(ConsumerError::Metadata)?; + topics + .into_iter() + .find(|t| t.name == topic) + .map(|t| t.partitions) + .ok_or_else(|| ConsumerError::TopicNotFound { + topic: topic.to_string(), + }) +} + +/// Errors that can occur when working with the Kafka consumer. +#[derive(Debug, thiserror::Error)] +pub enum ConsumerError { + /// Failed to establish the broker connection (SASL, TLS, or bootstrap) + #[error(transparent)] + Connection(#[from] ConnectionError), + + /// Failed to list topics from the broker + #[error("failed to list topics from the broker")] + Metadata(#[source] rskafka::client::error::Error), + + /// The configured topic does not exist (or is not visible to the credentials) + #[error("topic '{topic}' does not exist on the broker or is not visible to the credentials")] + TopicNotFound { topic: String }, + + /// Failed to get partition client + #[error("failed to get partition client")] + PartitionClient(#[source] rskafka::client::error::Error), + + /// The requested partition is not part of the bound topic + #[error("partition {partition} is not part of topic '{topic}'")] + UnknownPartition { partition: i32, topic: String }, + + /// Failed to fetch records from Kafka + #[error("failed to fetch records from Kafka")] + Fetch(#[source] rskafka::client::error::Error), + + /// Failed to query a partition offset + #[error("failed to query partition offset")] + Offset(#[source] rskafka::client::error::Error), + + /// Kafka operation timed out + #[error("Kafka operation timed out")] + Timeout, +} + +impl ConsumerError { + /// Whether this is the broker refusing a fetch offset outside its retained + /// range. Callers recover by re-anchoring to the earliest available offset + /// (records below it were deleted by retention). + pub fn is_offset_out_of_range(&self) -> bool { + matches!( + self, + Self::Fetch(rskafka::client::error::Error::ServerError { + protocol_error: rskafka::client::error::ProtocolError::OffsetOutOfRange, + .. + }) + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn config_deserializes_with_topic_and_defaults() { + let json = r#"{ + "brokers": ["localhost:9092"], + "topic": "studio.subgraph.indexing.requests" + }"#; + + let config: KafkaConsumerConfig = serde_json::from_str(json).expect("valid config"); + assert_eq!(config.brokers, vec!["localhost:9092".to_string()]); + assert_eq!(config.topic, "studio.subgraph.indexing.requests"); + assert_eq!(config.sasl_mechanism, None); + assert_eq!(config.sasl_username, None); + assert_eq!(config.sasl_password, None); + assert!(!config.tls_enabled); + assert_eq!(config.tls_ca_cert_path, None); + assert_eq!(config.connect_timeout_secs, 60); + } + + #[tokio::test] + async fn connect_gives_up_after_the_configured_timeout() { + // Port 1 refuses connections, which the underlying client retries + // forever; the configured bound must turn that into an error. + let config = KafkaConsumerConfig { + brokers: vec!["127.0.0.1:1".to_string()], + topic: "test".to_string(), + sasl_mechanism: None, + sasl_username: None, + sasl_password: None, + tls_enabled: false, + tls_ca_cert_path: None, + connect_timeout_secs: 1, + }; + let started = std::time::Instant::now(); + let result = KafkaConsumer::connect(&config).await; + assert!( + matches!(result, Err(ConsumerError::Timeout)), + "expected a timeout, got {:?}", + result.err() + ); + assert!( + started.elapsed() < Duration::from_secs(30), + "connect must give up promptly" + ); + } + + #[tokio::test] + async fn connect_rejects_an_empty_brokers_list() { + let config = KafkaConsumerConfig { + brokers: Vec::new(), + topic: "test".to_string(), + sasl_mechanism: None, + sasl_username: None, + sasl_password: None, + tls_enabled: false, + tls_ca_cert_path: None, + connect_timeout_secs: 60, + }; + assert!(matches!( + KafkaConsumer::connect(&config).await, + Err(ConsumerError::Connection( + super::super::connection::ConnectionError::MissingBrokers + )) + )); + } + + #[test] + fn config_without_topic_is_rejected() { + // No default topic on purpose: a missing name must fail configuration + // loading, not silently consume from nowhere. + let json = r#"{ "brokers": ["localhost:9092"] }"#; + + let err = serde_json::from_str::(json).unwrap_err(); + assert!( + err.to_string().contains("topic"), + "error should name the missing field: {err}" + ); + } + + #[test] + fn config_with_unknown_fields_is_rejected() { + let json = r#"{ + "brokers": ["localhost:9092"], + "topic": "t", + "partitions": 16 + }"#; + + let err = serde_json::from_str::(json).unwrap_err(); + assert!( + err.to_string().contains("partitions"), + "error should name the unknown field: {err}" + ); + } + + #[test] + fn debug_output_redacts_the_sasl_password() { + let config = KafkaConsumerConfig { + brokers: vec!["localhost:9092".to_string()], + topic: "test".to_string(), + sasl_mechanism: Some("PLAIN".to_string()), + sasl_username: Some("user".to_string()), + sasl_password: Some("hunter2".to_string()), + tls_enabled: false, + tls_ca_cert_path: None, + connect_timeout_secs: 60, + }; + let rendered = format!("{config:?}"); + assert!( + !rendered.contains("hunter2"), + "debug output must not contain the password: {rendered}" + ); + assert!( + rendered.contains(""), + "debug output should mark the password as redacted: {rendered}" + ); + } +} diff --git a/dipper-producer/src/kafka/producer.rs b/dipper-producer/src/kafka/producer.rs index 4a3f3afb..303a206e 100644 --- a/dipper-producer/src/kafka/producer.rs +++ b/dipper-producer/src/kafka/producer.rs @@ -1,21 +1,13 @@ //! Kafka producer for sending dipper events on a configured topic -use std::{ - path::{Path, PathBuf}, - sync::{Arc, Once}, - time::Duration, -}; +use std::{path::PathBuf, sync::Arc, time::Duration}; use rskafka::{ - client::{ - ClientBuilder, Credentials, SaslConfig, - partition::{Compression, PartitionClient, UnknownTopicHandling}, - }, + client::partition::{Compression, PartitionClient, UnknownTopicHandling}, record::Record, }; -use rustls::ClientConfig; -static RUSTLS_CRYPTO_PROVIDER: Once = Once::new(); +use super::connection::{self, ConnectOptions, ConnectionError}; /// Kafka producer configuration. #[derive(Clone, serde::Deserialize)] @@ -46,6 +38,11 @@ pub struct KafkaConfig { /// Path to a PEM-encoded CA certificate file for TLS verification. #[serde(default)] pub tls_ca_cert_path: Option, + /// Seconds allowed for the initial connect and partition binding (default: + /// 60). Load-bearing: the underlying client retries an unreachable broker + /// forever, so without this bound `new` would never return. + #[serde(default = "super::consumer::default_connect_timeout_secs")] + pub connect_timeout_secs: u64, } // Manual impl instead of derive: the service logs the whole config with Debug @@ -64,6 +61,7 @@ impl std::fmt::Debug for KafkaConfig { ) .field("tls_enabled", &self.tls_enabled) .field("tls_ca_cert_path", &self.tls_ca_cert_path) + .field("connect_timeout_secs", &self.connect_timeout_secs) .finish() } } @@ -89,32 +87,36 @@ pub struct KafkaProducer { impl KafkaProducer { const PRODUCE_TIMEOUT: Duration = Duration::from_secs(30); - /// Creates a new Kafka producer with the given configuration. + /// Creates a new Kafka producer with the given configuration. Bounded by + /// `connect_timeout_secs`, since the underlying client retries an + /// unreachable broker forever. + pub async fn new(config: &KafkaConfig) -> Result { + tokio::time::timeout( + Duration::from_secs(config.connect_timeout_secs), + Self::new_inner(config), + ) + .await + .map_err(|_| Error::Timeout)? + } + #[expect( clippy::cast_possible_wrap, reason = "predates this lint; fix when next touched" )] - pub async fn new(config: &KafkaConfig) -> Result { + async fn new_inner(config: &KafkaConfig) -> Result { if config.partitions == 0 { return Err(Error::InvalidPartitionCount); } - let mut builder = ClientBuilder::new(config.brokers.clone()); - - // Configure SASL authentication if mechanism is specified - if let Some(mechanism_str) = &config.sasl_mechanism { - let mechanism: SaslMechanism = mechanism_str.parse()?; - let sasl_config = Self::build_sasl_config(mechanism, config)?; - builder = builder.sasl_config(sasl_config); - } - - // Configure TLS if enabled - if config.tls_enabled { - let tls_config = Self::build_tls_config(config.tls_ca_cert_path.as_deref())?; - builder = builder.tls_config(tls_config); - } - - let client = builder.build().await.map_err(Error::Connection)?; + let client = connection::connect(ConnectOptions { + brokers: &config.brokers, + sasl_mechanism: config.sasl_mechanism.as_deref(), + sasl_username: config.sasl_username.as_deref(), + sasl_password: config.sasl_password.as_deref(), + tls_enabled: config.tls_enabled, + tls_ca_cert_path: config.tls_ca_cert_path.as_deref(), + }) + .await?; let client = Arc::new(client); let mut partition_clients = Vec::with_capacity(config.partitions as usize); @@ -134,68 +136,8 @@ impl KafkaProducer { }) } - /// Builds SASL configuration from the provided mechanism and credentials. - fn build_sasl_config( - mechanism: SaslMechanism, - config: &KafkaConfig, - ) -> Result { - let username = config - .sasl_username - .clone() - .ok_or(Error::MissingSaslUsername)?; - let password = config - .sasl_password - .clone() - .ok_or(Error::MissingSaslPassword)?; - - let credentials = Credentials::new(username, password); - - Ok(match mechanism { - SaslMechanism::Plain => SaslConfig::Plain(credentials), - SaslMechanism::ScramSha256 => SaslConfig::ScramSha256(credentials), - SaslMechanism::ScramSha512 => SaslConfig::ScramSha512(credentials), - }) - } - - /// Builds TLS configuration. - /// - /// If a custom CA certificate path is provided, the client will trust that CA - /// for verifying broker connections. Otherwise, system root certificates are used. - fn build_tls_config(ca_cert_path: Option<&Path>) -> Result, Error> { - install_rustls_crypto_provider(); - - let root_store = match ca_cert_path { - Some(path) => { - let ca_pem = fs_err::read(path).map_err(|e| Error::TlsCaCert { source: e })?; - let mut reader = std::io::BufReader::new(&ca_pem[..]); - let certs: Vec<_> = rustls_pemfile::certs(&mut reader) - .collect::>() - .map_err(|e| Error::TlsCaCert { source: e })?; - - let mut store = rustls::RootCertStore::empty(); - for cert in certs { - store.add(cert).map_err(|e| Error::TlsCaCert { - source: std::io::Error::new(std::io::ErrorKind::InvalidData, e), - })?; - } - store - } - None => rustls::RootCertStore { - roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(), - }, - }; - - let tls_config = ClientConfig::builder() - .with_root_certificates(root_store) - .with_no_client_auth(); - - Ok(Arc::new(tls_config)) - } - - /// Sends an event to Kafka. - /// - /// Events are partitioned by the partition key (table discriminator) before being written to - /// Kafka. The produce attempt times out after 30 seconds. + /// Sends an event to Kafka, partitioned by the partition key (table + /// discriminator). The produce attempt times out after 30 seconds. #[expect( clippy::cast_sign_loss, reason = "predates this lint; fix when next touched" @@ -221,12 +163,9 @@ impl KafkaProducer { .map(|_| ()) } - /// Computes the partition for a given key. - /// - /// Uses a deterministic FNV-1a hash modulo the partition count so that a given - /// key always maps to the same partition across restarts and instances, - /// preserving per-key ordering. The partition count is configured via - /// `KafkaConfig::partitions`. + /// Computes the partition for a key: deterministic FNV-1a hash modulo + /// `KafkaConfig::partitions`, so a key maps to the same partition across + /// restarts and instances, preserving per-key ordering. #[expect( clippy::cast_possible_wrap, reason = "predates this lint; fix when next touched" @@ -243,20 +182,12 @@ impl KafkaProducer { } } -fn install_rustls_crypto_provider() { - RUSTLS_CRYPTO_PROVIDER.call_once(|| { - // Necessary for the Kafka client: it builds a Rustls TLS config directly, - // so install a provider before `ClientConfig::builder()` tries to infer one. - let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); - }); -} - /// Errors that can occur when working with the Kafka producer. #[derive(Debug, thiserror::Error)] pub enum Error { - /// Failed to connect to Kafka brokers - #[error("failed to connect to Kafka brokers")] - Connection(#[source] rskafka::client::error::Error), + /// Failed to establish the broker connection (SASL, TLS, or bootstrap) + #[error(transparent)] + Connection(#[from] ConnectionError), /// Failed to get partition client #[error("failed to get partition client")] @@ -273,126 +204,12 @@ pub enum Error { /// Partition count must be greater than zero #[error("partitions must be greater than zero")] InvalidPartitionCount, - - /// Unsupported SASL mechanism - #[error("unsupported SASL mechanism '{0}', supported: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512")] - UnsupportedSaslMechanism(String), - - /// Missing SASL username - #[error("sasl_username is required when sasl_mechanism is set")] - MissingSaslUsername, - - /// Missing SASL password - #[error("sasl_password is required when sasl_mechanism is set")] - MissingSaslPassword, - - /// Failed to load TLS CA certificate - #[error("failed to load TLS CA certificate")] - TlsCaCert { - #[source] - source: std::io::Error, - }, -} - -/// Supported SASL authentication mechanisms. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum SaslMechanism { - Plain, - ScramSha256, - ScramSha512, -} - -impl std::str::FromStr for SaslMechanism { - type Err = Error; - - fn from_str(s: &str) -> Result { - match s.to_uppercase().as_str() { - "PLAIN" => Ok(Self::Plain), - "SCRAM-SHA-256" => Ok(Self::ScramSha256), - "SCRAM-SHA-512" => Ok(Self::ScramSha512), - _ => Err(Error::UnsupportedSaslMechanism(s.to_string())), - } - } } #[cfg(test)] mod tests { use super::*; - #[test] - fn should_parse_sasl_mechanism_correctly() { - // Case-insensitive parsing - assert_eq!( - "PLAIN".parse::().unwrap(), - SaslMechanism::Plain - ); - assert_eq!( - "plain".parse::().unwrap(), - SaslMechanism::Plain - ); - assert_eq!( - "SCRAM-SHA-256".parse::().unwrap(), - SaslMechanism::ScramSha256 - ); - assert_eq!( - "scram-sha-256".parse::().unwrap(), - SaslMechanism::ScramSha256 - ); - assert_eq!( - "SCRAM-SHA-512".parse::().unwrap(), - SaslMechanism::ScramSha512 - ); - - // Unsupported mechanism - assert!("GSSAPI".parse::().is_err()); - } - - #[test] - fn should_build_sasl_plain_sasl_config() { - let config = make_kafka_config(Some("PLAIN"), Some("user".into()), Some("pass".into())); - let result = KafkaProducer::build_sasl_config(SaslMechanism::Plain, &config); - assert!(matches!(result, Ok(SaslConfig::Plain(_)))); - } - - #[test] - fn should_build_scram_sha_256_sasl_config() { - let config = make_kafka_config( - Some("SCRAM-SHA-256"), - Some("user".into()), - Some("pass".into()), - ); - let result = KafkaProducer::build_sasl_config(SaslMechanism::ScramSha256, &config); - assert!(matches!(result, Ok(SaslConfig::ScramSha256(_)))); - } - - #[test] - fn should_build_sasl_sha_512_sasl_config() { - let config = make_kafka_config( - Some("SCRAM-SHA-512"), - Some("user".into()), - Some("pass".into()), - ); - let result = KafkaProducer::build_sasl_config(SaslMechanism::ScramSha512, &config); - assert!(matches!(result, Ok(SaslConfig::ScramSha512(_)))); - } - - #[test] - fn should_throw_err_on_missing_credentials() { - // Missing username - let config = make_kafka_config(Some("PLAIN"), None, Some("pass".into())); - assert!(matches!( - KafkaProducer::build_sasl_config(SaslMechanism::Plain, &config), - Err(Error::MissingSaslUsername) - )); - - // Missing password - let config = make_kafka_config(Some("PLAIN"), Some("user".into()), None); - assert!(matches!( - KafkaProducer::build_sasl_config(SaslMechanism::Plain, &config), - Err(Error::MissingSaslPassword) - )); - } - #[test] fn debug_output_redacts_the_sasl_password() { let config = make_kafka_config(Some("PLAIN"), Some("user".into()), Some("hunter2".into())); @@ -434,6 +251,7 @@ mod tests { sasl_password: sasl_pass, tls_enabled: false, tls_ca_cert_path: None, + connect_timeout_secs: 60, } } } diff --git a/dipper-producer/src/lib.rs b/dipper-producer/src/lib.rs index 71079c9c..1ac830ec 100644 --- a/dipper-producer/src/lib.rs +++ b/dipper-producer/src/lib.rs @@ -1,3 +1,7 @@ pub mod events; pub mod kafka; pub mod proto; + +// Re-exported so consumers of the generated types decode without pinning +// their own copy of prost. +pub use prost; diff --git a/dipper-producer/src/proto/dipper.subgraph.indexing.agreement.events.v1.rs b/dipper-producer/src/proto/dipper.subgraph.indexing.agreement.events.v1.rs index be209e64..100b909d 100644 --- a/dipper-producer/src/proto/dipper.subgraph.indexing.agreement.events.v1.rs +++ b/dipper-producer/src/proto/dipper.subgraph.indexing.agreement.events.v1.rs @@ -178,7 +178,7 @@ pub struct SubgraphIndexingAgreementTerminated { /// the one just terminated. /// 0 -> the Subgraph no longer has any active, accepted indexing agreements. /// -1 -> the count was unavailable when the event was emitted (e.g. a transient - /// datastore error); treat as unknown, not as zero. + /// datastore error); treat as unknown, not as zero. #[prost(int32, tag = "5")] pub remaining_accepted_indexing_agreements: i32, } diff --git a/dipper-producer/src/proto/mod.rs b/dipper-producer/src/proto/mod.rs index 75c540c5..7709ba0c 100644 --- a/dipper-producer/src/proto/mod.rs +++ b/dipper-producer/src/proto/mod.rs @@ -1,4 +1,11 @@ // Generated protobuf types for indexing agreement events // -// To regenerate, run: just gen-indexing-agreement-events-proto +// To regenerate, run: just gen-event-protos include!("dipper.subgraph.indexing.agreement.events.v1.rs"); + +/// Subgraph indexing request events that Studio produces and the dipper consumes. +/// Schema vendored in `proto/subgraph-indexing-request-events.proto`. +/// To regenerate, run: just gen-event-protos +pub mod studio { + include!("studio.subgraph.indexing.requests.events.v1.rs"); +} diff --git a/dipper-producer/src/proto/studio.subgraph.indexing.requests.events.v1.rs b/dipper-producer/src/proto/studio.subgraph.indexing.requests.events.v1.rs new file mode 100644 index 00000000..760abe2a --- /dev/null +++ b/dipper-producer/src/proto/studio.subgraph.indexing.requests.events.v1.rs @@ -0,0 +1,67 @@ +// This file is @generated by prost-build. +/// SubgraphIndexingRequestEvent is the envelope that wraps all the Subgraph Indexing Request event types. +/// +/// Provides: +/// - event_id -> unique identifier (uuid) +/// - event_type -> event type discrimination +/// - event_version -> versioning for schema evolution awareness to the consumer +/// - timestamp -> when the event occurred, useful for ordering/debugging +/// - subgraph_deployment_qm_hash -> Qm hash of the Subgraph deployment +/// - the_graph_network_caip2id -> CAIP-2 chain id of The Graph protocol network the Subgraph Deployment is published to (e.g. eip155:42161) +/// - payload -> the event payload, determined by the event_type +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct SubgraphIndexingRequestEvent { + /// Unique event identifier. + /// Format: UUID v7 (time-ordered) for natural chronological sorting. + /// Example: "01912345-6789-7abc-def0-123456789abc" + #[prost(string, tag = "1")] + pub event_id: ::prost::alloc::string::String, + /// Event type discriminator for routing and filtering. + /// Values: "subgraph.indexing.request.propose", "subgraph.indexing.agreements.terminate" + #[prost(string, tag = "2")] + pub event_type: ::prost::alloc::string::String, + /// Schema version for forward compatibility. + /// Consumers should handle unknown fields gracefully. + /// Current version: "1.0" + #[prost(string, tag = "3")] + pub event_version: ::prost::alloc::string::String, + /// Event timestamp in RFC 3339 format. + /// Example: "2024-01-15T10:30:00.123Z" + #[prost(string, tag = "4")] + pub timestamp: ::prost::alloc::string::String, + /// Qm hash of the Subgraph deployment with a submitted indexing agreement event. + #[prost(string, tag = "5")] + pub subgraph_deployment_qm_hash: ::prost::alloc::string::String, + /// CAIP-2 chain id of The Graph protocol network where the Subgraph was published to. + /// Format: "eip155:{chain_id}". Examples: "eip155:42161" (arbitrum), "eip155:421614" (arbitrum-sepolia) + #[prost(string, tag = "6")] + pub the_graph_network_caip2id: ::prost::alloc::string::String, + /// Event payload - exactly one of the specific event types. + /// Use event_type field to determine which payload is present. + #[prost(oneof = "subgraph_indexing_request_event::Payload", tags = "7")] + pub payload: ::core::option::Option, +} +/// Nested message and enum types in `SubgraphIndexingRequestEvent`. +pub mod subgraph_indexing_request_event { + /// Event payload - exactly one of the specific event types. + /// Use event_type field to determine which payload is present. + #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)] + pub enum Payload { + #[prost(message, tag = "7")] + SubgraphIndexingRequestPropose(super::SubgraphIndexingRequestPropose), + } +} +/// SubgraphIndexingRequestPropose is emitted when the Subgraph developer initiates a request to have the Subgraph indexed +/// +/// Event type: subgraph.indexing.request.propose +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct SubgraphIndexingRequestPropose { + /// Number of requested Indexing agreements to find + #[prost(int32, tag = "1")] + pub indexing_agreements_requested: i32, + /// CAIP-2 chain id of the network the Subgraph indexes (its data source), + /// distinct from the_graph_network_caip2id on the envelope. + /// Format: "eip155:{chain_id}". Example: "eip155:1" for a subgraph indexing Ethereum mainnet. + #[prost(string, tag = "2")] + pub indexed_network_caip2id: ::prost::alloc::string::String, +} diff --git a/dipper-producer/tests/it_kafka_redpanda.rs b/dipper-producer/tests/it_kafka_redpanda.rs new file mode 100644 index 00000000..3667916f --- /dev/null +++ b/dipper-producer/tests/it_kafka_redpanda.rs @@ -0,0 +1,173 @@ +//! Kafka producer/consumer roundtrip tests against a real Redpanda broker. +//! Gated on `REDPANDA_BROKERS` (e.g. `localhost:9092`): unset, every test +//! skips with a note; CI starts a Redpanda container and always runs them. + +use dipper_producer::kafka::{ + ConsumerError, KafkaConfig, KafkaConsumer, KafkaConsumerConfig, KafkaProducer, OffsetAt, +}; + +fn brokers() -> Option> { + match std::env::var("REDPANDA_BROKERS") { + Ok(value) if !value.trim().is_empty() => Some( + value + .split(',') + .map(|broker| broker.trim().to_string()) + .collect(), + ), + // REQUIRE_REDPANDA turns the silent skip into a failure, so CI cannot + // go green while accidentally testing nothing. + _ if std::env::var("REQUIRE_REDPANDA").is_ok() => { + panic!("REQUIRE_REDPANDA is set but REDPANDA_BROKERS is not") + } + _ => { + eprintln!("skipping Redpanda-backed test: REDPANDA_BROKERS is not set"); + None + } + } +} + +/// A topic name unique to this test process and call site, so parallel tests +/// on a shared broker never read each other's records. +#[expect( + clippy::expect_used, + reason = "a test helper fails its test on any error" +)] +fn unique_topic(label: &str) -> String { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock before unix epoch") + .as_nanos(); + format!("dipper.test.{label}.{}.{nanos}", std::process::id()) +} + +#[expect( + clippy::expect_used, + reason = "a test helper fails its test on any error" +)] +async fn create_topic(brokers: &[String], topic: &str, partitions: i32) { + let client = rskafka::client::ClientBuilder::new(brokers.to_vec()) + .build() + .await + .expect("connect to broker"); + client + .controller_client() + .expect("controller client") + .create_topic(topic, partitions, 1, 5_000) + .await + .expect("create topic"); +} + +#[expect( + clippy::expect_used, + reason = "a test helper fails its test on any error" +)] +fn producer_config(brokers: Vec, topic: &str, partitions: u32) -> KafkaConfig { + serde_json::from_value(serde_json::json!({ + "brokers": brokers, + "topic": topic, + "partitions": partitions, + })) + .expect("valid producer config") +} + +#[expect( + clippy::expect_used, + reason = "a test helper fails its test on any error" +)] +fn consumer_config(brokers: Vec, topic: &str) -> KafkaConsumerConfig { + serde_json::from_value(serde_json::json!({ + "brokers": brokers, + "topic": topic, + })) + .expect("valid consumer config") +} + +#[tokio::test] +async fn produce_then_consume_roundtrip_across_partitions() { + let Some(brokers) = brokers() else { return }; + let topic = unique_topic("roundtrip"); + create_topic(&brokers, &topic, 2).await; + + // The producer compresses batches with GZIP, like Studio's producer does, + // so a successful roundtrip also proves fetch-side decompression. + let producer = KafkaProducer::new(&producer_config(brokers.clone(), &topic, 2)) + .await + .expect("producer connects"); + let payloads: Vec<(String, Vec)> = (0..5) + .map(|i| (format!("key-{i}"), format!("payload-{i}").into_bytes())) + .collect(); + for (key, payload) in &payloads { + producer.send(key, payload).await.expect("produce"); + } + + let consumer = KafkaConsumer::connect(&consumer_config(brokers, &topic)) + .await + .expect("consumer connects"); + assert_eq!(consumer.partitions(), vec![0, 1]); + + let mut consumed: Vec> = Vec::new(); + for partition in consumer.partitions() { + let mut offset = consumer + .offset(partition, OffsetAt::Earliest) + .await + .expect("earliest offset"); + let end = consumer + .offset(partition, OffsetAt::Latest) + .await + .expect("latest offset"); + while offset < end { + let (records, _) = consumer + .fetch(partition, offset, 1_048_576, 500) + .await + .expect("fetch"); + assert!(!records.is_empty(), "records expected below the watermark"); + for record in records { + consumed.push(record.record.value.expect("record value")); + offset = record.offset + 1; + } + } + } + + let mut expected: Vec> = payloads.into_iter().map(|(_, payload)| payload).collect(); + expected.sort(); + consumed.sort(); + assert_eq!(consumed, expected, "every produced payload comes back"); +} + +#[tokio::test] +async fn consumer_refuses_a_missing_topic() { + let Some(brokers) = brokers() else { return }; + let topic = unique_topic("never-created"); + + let Err(err) = KafkaConsumer::connect(&consumer_config(brokers, &topic)).await else { + panic!("a missing topic must fail the connect"); + }; + assert!( + matches!(err, ConsumerError::TopicNotFound { .. }), + "expected TopicNotFound, got {err:?}" + ); +} + +#[tokio::test] +async fn fetching_beyond_the_retained_range_is_detectable() { + let Some(brokers) = brokers() else { return }; + let topic = unique_topic("out-of-range"); + create_topic(&brokers, &topic, 1).await; + + let producer = KafkaProducer::new(&producer_config(brokers.clone(), &topic, 1)) + .await + .expect("producer connects"); + producer.send("key", b"payload").await.expect("produce"); + + let consumer = KafkaConsumer::connect(&consumer_config(brokers, &topic)) + .await + .expect("consumer connects"); + let err = consumer + .fetch(0, 5_000, 1_048_576, 500) + .await + .expect_err("an offset far past the watermark must error"); + assert!( + err.is_offset_out_of_range(), + "expected an offset-out-of-range error, got {err:?}" + ); +} diff --git a/justfile b/justfile index 500af5f6..72d93f7a 100644 --- a/justfile +++ b/justfile @@ -117,7 +117,9 @@ remove-git-hooks: # Remove the pre-commit hooks pre-commit uninstall --config .github/pre-commit-config.yaml -# Generate job events protobuf bindings (RUSTFLAGS="--cfg gen_event_proto" cargo check) +# Generate event protobuf bindings (RUSTFLAGS="--cfg gen_event_proto" cargo check) [group: 'codegen'] -gen-indexing-agreement-events-proto: - RUSTFLAGS="--cfg gen_event_proto" cargo check -p dipper-producer \ No newline at end of file +gen-event-protos: + RUSTFLAGS="--cfg gen_event_proto" cargo check -p dipper-producer + +alias gen-indexing-agreement-events-proto := gen-event-protos \ No newline at end of file