From 230e4bddca636467fe11bf30900354e0a947e2b1 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 24 Aug 2026 17:06:47 +0300 Subject: [PATCH 1/7] feat(pgregistry): store Kafka consumer offsets per topic partition The Kafka client the dipper uses has no consumer groups, so the broker cannot remember how far a consumer got. A new table records the next offset to fetch per topic partition, written after each record is processed, so a restarted dipper resumes where it left off. --- ...60824000000_add_kafka_consumer_offsets.sql | 22 ++++++ dipper-pgregistry/src/postgres.rs | 55 +++++++++++++++ .../tests/it_registry_postgres.rs | 69 +++++++++++++++++++ 3 files changed, 146 insertions(+) create mode 100644 dipper-pgregistry/migrations/20260824000000_add_kafka_consumer_offsets.sql 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 8f2254fc..0e287421 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -1926,6 +1926,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 84a63bcb..8507307d 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -3256,3 +3256,72 @@ 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) + ); +} From d6e022cdac7a20506d4f373f32f0b60e0f277f23 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 24 Aug 2026 17:09:04 +0300 Subject: [PATCH 2/7] refactor(service): share the set-indexing-target logic beyond the RPC Setting an indexing target does 3 things: upsert the request, emit the request-received event for a new one, and queue a reassessment. That sequence lived inside the admin RPC handler; it moves to a shared function so the upcoming Kafka consumer applies requests identically. --- .../handlers/indexing_requests.rs | 147 +++------------ bin/dipper-service/src/main.rs | 1 + bin/dipper-service/src/set_indexing_target.rs | 176 ++++++++++++++++++ 3 files changed, 205 insertions(+), 119 deletions(-) create mode 100644 bin/dipper-service/src/set_indexing_target.rs 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 e5c26605..97442110 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}, }; @@ -109,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) + }) } } @@ -267,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, }; @@ -379,8 +290,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( @@ -399,10 +309,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/main.rs b/bin/dipper-service/src/main.rs index 93a4460b..37a768a2 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -25,6 +25,7 @@ mod health; mod indexer_rpc_client; mod network; mod registry; +mod set_indexing_target; mod signing; mod supervisor; #[cfg(test)] 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..22405f1a --- /dev/null +++ b/bin/dipper-service/src/set_indexing_target.rs @@ -0,0 +1,176 @@ +//! 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 changed but the reassessment job could not be queued; a later + /// call with the same target is a registry no-op yet queues the job again. + #[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. +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) +} From c931bb61971c9328488d70dd32f0630cc4280c0e Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 24 Aug 2026 17:15:34 +0300 Subject: [PATCH 3/7] feat(service): consume Studio's indexing requests from Redpanda A new background service reads the propose events Studio publishes and applies each through the same path as the admin RPC, so a developer's request in Studio reaches indexer selection with no RPC call. It resumes from offsets stored in the dipper's database and skips bad messages. --- bin/dipper-service/src/config.rs | 79 +- bin/dipper-service/src/main.rs | 38 + bin/dipper-service/src/network/service.rs | 1 + .../service/indexing_request_consumer.rs | 999 ++++++++++++++++++ bin/dipper-service/src/registry.rs | 28 + dipper-producer/src/kafka.rs | 2 + dipper-producer/src/lib.rs | 4 + 7 files changed, 1150 insertions(+), 1 deletion(-) create mode 100644 bin/dipper-service/src/network/service/indexing_request_consumer.rs diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index d9e0bad9..66fa5123 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. @@ -1369,6 +1373,79 @@ 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: true when the section is present) + #[serde(default = "default_indexing_request_consumer_enabled")] + 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_enabled() -> bool { + true +} + +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. + pub fn validate(&self) -> Result<(), String> { + if !self.enabled { + return Ok(()); + } + // 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(), + ); + } + if self.fetch_max_bytes <= 0 { + return Err(format!( + "indexing_request_consumer.fetch_max_bytes ({}) must be positive", + 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::*; diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 37a768a2..93efe7a2 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -82,6 +82,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}"); } @@ -568,6 +573,24 @@ 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, + config: consumer_conf.clone(), + }; + let (handle, service) = network::service::indexing_request_consumer::new(ctx); + Some((handle, service)) + } + _ => None, + }; + //- The admin RPC service let (admin_rpc_handle, admin_rpc_service) = { let config = admin_rpc_server::service::Config { @@ -679,6 +702,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); @@ -755,6 +788,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 86f9bb3e..8d51f4bc 100644 --- a/bin/dipper-service/src/network/service.rs +++ b/bin/dipper-service/src/network/service.rs @@ -5,5 +5,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..7f17e671 --- /dev/null +++ b/bin/dipper-service/src/network/service/indexing_request_consumer.rs @@ -0,0 +1,999 @@ +//! 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::{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); + +/// 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, + /// 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.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. + let result = tokio::select! { + _ = rx_stop.recv() => Ok(()), + joined = tasks.join_next() => 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" + )), + }, + }; + + 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 (at-least-once; redelivery is +/// safe because an unchanged target count is a registry no-op). +#[allow(clippy::too_many_arguments)] +async fn partition_loop( + consumer: Arc, + partition: i32, + registry: R, + worker_queue: W, + events: Arc, + protocol_chain_id: ChainId, + 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 = config.max_wait.as_millis().min(i32::MAX as u128) as i32; + + // 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 = match fetch { + Ok((records, _high_watermark)) => records, + Err(err) if err.is_offset_out_of_range() => { + // Retention deleted records under the cursor; re-anchor to the + // earliest still-available record rather than spinning forever. + match consumer.offset(partition, OffsetAt::Earliest).await { + Ok(earliest) => { + tracing::warn!( + partition, + stale_offset = next_offset, + earliest, + "consumer offset fell outside the broker's retained range; re-anchoring" + ); + next_offset = earliest; + persist_offset(®istry, &topic, partition, next_offset).await; + } + Err(err) => { + tracing::error!(partition, error = %err, "failed to query earliest offset"); + 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; + } + }; + + for record_and_offset in records { + let offset = record_and_offset.offset; + // Apply failures (registry or queue down) retry the same record + // rather than skip it: a propose must eventually take effect. + loop { + let disposition = handle_record( + ®istry, + &worker_queue, + &events, + protocol_chain_id, + config.requested_by, + &expected_network, + record_and_offset.record.value.as_deref(), + ) + .await; + + match disposition { + Ok(Disposition::Applied) => break, + Ok(Disposition::Skipped(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" + ); + break; + } + Err(err) => { + tracing::error!(partition, offset, error = %err, "failed to apply indexing request, retrying"); + if sleep_or_shutdown(&mut shutdown_rx, RETRY_BACKOFF).await { + return; + } + } + } + } + + next_offset = offset + 1; + persist_offset(®istry, &topic, partition, next_offset).await; + } + } +} + +/// 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, + } +} + +/// What became of one record. +#[derive(Debug, PartialEq, Eq)] +enum Disposition { + /// The propose was applied through the shared set-indexing-target path. + Applied, + /// The record cannot be processed and was deliberately skipped. + Skipped(&'static str), +} + +/// Decode and apply a single record. `Err` means a transient processing +/// failure the caller should retry; skips are a successful `Disposition`. +async fn handle_record( + registry: &R, + worker_queue: &W, + events: &Arc, + protocol_chain_id: ChainId, + requested_by: Address, + expected_network: &str, + value: Option<&[u8]>, +) -> Result +where + R: IndexingRequestRegistry + Send + Sync, + W: WorkerQueue + Send + Sync, +{ + let target = match decode_propose(value, expected_network) { + Ok(target) => target, + Err(reason) => return Ok(Disposition::Skipped(reason)), + }; + + 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" + ); + + 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: target.num_candidates, + // Interactive: a developer just asked for this in Studio. + priority: JobPriority::Interactive, + }, + ) + .await?; + + Ok(Disposition::Applied) +} + +/// 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"; + + 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` returns a configured + /// outcome (or errors) and records the arguments it was called with. + #[derive(Clone)] + struct MockRegistry { + outcome: Option, + calls: Arc>>, + } + + impl MockRegistry { + fn returning(outcome: SetTargetOutcome) -> Self { + Self { + outcome: Some(outcome), + calls: Arc::new(Mutex::new(Vec::new())), + } + } + + fn erroring() -> Self { + Self { + outcome: None, + calls: Arc::new(Mutex::new(Vec::new())), + } + } + + 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, + )); + match &self.outcome { + Some(outcome) => Ok(outcome.clone()), + 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. + #[derive(Clone)] + struct MockWorker { + reassessments: Arc>>, + } + + impl MockWorker { + fn new() -> Self { + Self { + reassessments: Arc::new(Mutex::new(Vec::new())), + } + } + + 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 { + 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, + _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!() + } + } + + async fn run_handle_record( + registry: &MockRegistry, + worker: &MockWorker, + bytes: Option<&[u8]>, + ) -> ( + Result, + Vec, + ) { + let events_capture = CapturingEventsProducer::new(); + let events: Arc = + Arc::new(events_capture.clone()); + + let result = handle_record( + registry, + worker, + &events, + PROTOCOL_CHAIN_ID, + requester(), + "eip155:42161", + bytes, + ) + .await; + + (result, 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 bytes = valid_propose_bytes(); + let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + + assert!(matches!(result, Ok(Disposition::Applied)), "{result:?}"); + 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 bytes = valid_propose_bytes(); + let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + + assert!(matches!(result, Ok(Disposition::Applied)), "{result:?}"); + 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 bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "eip155:1", 0); + let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + + assert!(matches!(result, Ok(Disposition::Applied)), "{result:?}"); + 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] + async fn an_unprocessable_record_is_skipped_without_touching_the_registry() { + let registry = MockRegistry::returning(SetTargetOutcome::NoOp { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::new(); + + let bytes = encode_propose(EVENT_TYPE_TERMINATE, "eip155:42161", QM_HASH, "eip155:1", 3); + let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + + assert!(matches!(result, Ok(Disposition::Skipped(_))), "{result:?}"); + assert!(registry.calls().is_empty()); + assert!(worker.reassessments().is_empty()); + assert!(events.is_empty()); + } + + #[tokio::test] + async fn a_registry_failure_is_a_retryable_error_not_a_skip() { + let registry = MockRegistry::erroring(); + let worker = MockWorker::new(); + + let bytes = valid_propose_bytes(); + let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + + assert!(result.is_err(), "a registry failure must surface for retry"); + assert!(worker.reassessments().is_empty()); + assert!(events.is_empty()); + } +} diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index 0509cb78..c4b139a6 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -798,3 +798,31 @@ impl crate::network::service::chain_listener::ChainListenerStateRegistry for Reg .map_err(Into::into) } } + +#[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) + } +} diff --git a/dipper-producer/src/kafka.rs b/dipper-producer/src/kafka.rs index a6dfe341..789bd36a 100644 --- a/dipper-producer/src/kafka.rs +++ b/dipper-producer/src/kafka.rs @@ -9,3 +9,5 @@ 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/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; From c75740386c59c68e828ed245fc184a37b680f992 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 24 Aug 2026 17:21:45 +0300 Subject: [PATCH 4/7] test(kafka): prove the consumer against a real Redpanda broker in CI CI starts a Redpanda container and 2 test layers use it: producer-to-consumer roundtrips with GZIP batches, and a full service run showing offsets persist after processing and a restarted consumer resumes without re-applying old events. Without a broker the tests skip with a note. --- .github/workflows/ci.yml | 20 ++ Cargo.lock | 2 + bin/dipper-service/Cargo.toml | 2 + .../service/indexing_request_consumer.rs | 226 ++++++++++++++++++ dipper-producer/Cargo.toml | 1 + dipper-producer/tests/it_kafka_redpanda.rs | 152 ++++++++++++ 6 files changed, 403 insertions(+) create mode 100644 dipper-producer/tests/it_kafka_redpanda.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7bf97145..76862e53 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -88,10 +88,29 @@ 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 with: env_file: .env run: cargo nextest run --all-features -E 'test(~tests::it_)' @@ -100,6 +119,7 @@ jobs: uses: LNSD/sops-exec-action@6da1fbca63459d9796097496d5f5e6233555b31a # v1 env: SOPS_AGE_KEY: ${{ secrets.IT_TESTS_AGE_KEY }} + REDPANDA_BROKERS: localhost:9092 with: env_file: .env run: cargo nextest run --all-features -E 'kind(test)' diff --git a/Cargo.lock b/Cargo.lock index b53ab07e..09f0c6cc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2497,8 +2497,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 47c588e6..ac02bd39 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/network/service/indexing_request_consumer.rs b/bin/dipper-service/src/network/service/indexing_request_consumer.rs index 7f17e671..f0897533 100644 --- a/bin/dipper-service/src/network/service/indexing_request_consumer.rs +++ b/bin/dipper-service/src/network/service/indexing_request_consumer.rs @@ -996,4 +996,230 @@ mod tests { assert!(worker.reassessments().is_empty()); assert!(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(), + ), + _ => { + 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")] + 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, + }, + 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, + 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/dipper-producer/Cargo.toml b/dipper-producer/Cargo.toml index c8a5432f..6f66b9ad 100644 --- a/dipper-producer/Cargo.toml +++ b/dipper-producer/Cargo.toml @@ -22,6 +22,7 @@ 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 diff --git a/dipper-producer/tests/it_kafka_redpanda.rs b/dipper-producer/tests/it_kafka_redpanda.rs new file mode 100644 index 00000000..9f6c0beb --- /dev/null +++ b/dipper-producer/tests/it_kafka_redpanda.rs @@ -0,0 +1,152 @@ +//! 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(), + ), + _ => { + 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. +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()) +} + +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"); +} + +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") +} + +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:?}" + ); +} From 89bcdc80563435b12cffb691918dd52da8f949ee Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 24 Aug 2026 18:08:31 +0300 Subject: [PATCH 5/7] fix(consumer): keep the reassessment job when only its queueing fails Setting an indexing target commits the row change first, then queues a reassessment job. If that second step failed, retrying the whole record would hit the registry's no-op path (count unchanged) and quietly drop the job; the consumer now retries just the queue push instead. --- bin/dipper-service/src/config.rs | 6 +- .../service/indexing_request_consumer.rs | 340 +++++++++++------- bin/dipper-service/src/set_indexing_target.rs | 5 +- 3 files changed, 223 insertions(+), 128 deletions(-) diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index 66fa5123..baaab895 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -1429,9 +1429,11 @@ impl IndexingRequestConsumerConfig { "indexing_request_consumer.requested_by must be a non-zero address".to_string(), ); } - if self.fetch_max_bytes <= 0 { + // 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 positive", + "indexing_request_consumer.fetch_max_bytes ({}) must be at least 1024", self.fetch_max_bytes )); } diff --git a/bin/dipper-service/src/network/service/indexing_request_consumer.rs b/bin/dipper-service/src/network/service/indexing_request_consumer.rs index f0897533..457982c7 100644 --- a/bin/dipper-service/src/network/service/indexing_request_consumer.rs +++ b/bin/dipper-service/src/network/service/indexing_request_consumer.rs @@ -23,7 +23,7 @@ use tokio::{ use crate::{ config::IndexingRequestConsumerConfig, registry::IndexingRequestRegistry, - set_indexing_target::{SetIndexingTarget, apply_set_indexing_target}, + set_indexing_target::{ApplyError, SetIndexingTarget, apply_set_indexing_target}, worker::service::{JobPriority, WorkerQueue}, }; @@ -294,38 +294,39 @@ async fn partition_loop( for record_and_offset in records { let offset = record_and_offset.offset; - // Apply failures (registry or queue down) retry the same record - // rather than skip it: a propose must eventually take effect. - loop { - let disposition = handle_record( - ®istry, - &worker_queue, - &events, - protocol_chain_id, - config.requested_by, - &expected_network, - record_and_offset.record.value.as_deref(), - ) - .await; - - match disposition { - Ok(Disposition::Applied) => break, - Ok(Disposition::Skipped(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" - ); - break; - } - Err(err) => { - tracing::error!(partition, offset, error = %err, "failed to apply indexing request, retrying"); - if sleep_or_shutdown(&mut shutdown_rx, RETRY_BACKOFF).await { - return; - } + 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, + &target, + &mut shutdown_rx, + ) + .await; + if !applied { + // Shutdown arrived before the record took effect; the + // unadvanced offset redelivers it on the next run. + return; } } } @@ -367,60 +368,76 @@ async fn sleep_or_shutdown(shutdown_rx: &mut watch::Receiver, delay: Durat } } -/// What became of one record. -#[derive(Debug, PartialEq, Eq)] -enum Disposition { - /// The propose was applied through the shared set-indexing-target path. - Applied, - /// The record cannot be processed and was deliberately skipped. - Skipped(&'static str), -} - -/// Decode and apply a single record. `Err` means a transient processing -/// failure the caller should retry; skips are a successful `Disposition`. -async fn handle_record( +/// 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. +async fn apply_target_with_retries( registry: &R, worker_queue: &W, events: &Arc, protocol_chain_id: ChainId, requested_by: Address, - expected_network: &str, - value: Option<&[u8]>, -) -> Result + target: &ProposedTarget, + shutdown_rx: &mut watch::Receiver, +) -> bool where R: IndexingRequestRegistry + Send + Sync, W: WorkerQueue + Send + Sync, { - let target = match decode_propose(value, expected_network) { - Ok(target) => target, - Err(reason) => return Ok(Disposition::Skipped(reason)), - }; + 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: target.num_candidates, + // Interactive: a developer just asked for this in Studio. + priority: JobPriority::Interactive, + }, + ) + .await; - 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" - ); - - 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: target.num_candidates, - // Interactive: a developer just asked for this in Studio. - priority: JobPriority::Interactive, - }, - ) - .await?; + match result { + Ok(_) => return true, + Err(ApplyError::Registry(_)) => { + if sleep_or_shutdown(shutdown_rx, RETRY_BACKOFF).await { + return false; + } + } + Err(ApplyError::QueueReassess { id, .. }) => break id, + } + }; - Ok(Disposition::Applied) + loop { + if sleep_or_shutdown(shutdown_rx, RETRY_BACKOFF).await { + return false; + } + match worker_queue + .reassess_indexing_request( + queue_retry_id, + target.deployment_id, + target.deployment_chain_id, + target.num_candidates, + JobPriority::Interactive, + ) + .await + { + Ok(_) => return true, + Err(err) => { + tracing::error!( + indexing_request_id = %queue_retry_id, + error = ?err, + "retrying the reassessment job push" + ); + } + } + } } /// A validated propose event, reduced to what the registry call needs. @@ -716,27 +733,25 @@ mod tests { type SetTargetCall = (Address, DeploymentId, ChainId, usize); type ReassessCall = (IndexingRequestId, DeploymentId, ChainId, usize); - /// A registry whose `set_indexing_target_candidates` returns a configured - /// outcome (or errors) and records the arguments it was called with. + /// 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 { - outcome: Option, + script: Arc>>>, calls: Arc>>, } impl MockRegistry { - fn returning(outcome: SetTargetOutcome) -> Self { + /// `None` entries are errors; after the script is exhausted every call errors. + fn scripted(script: Vec>) -> Self { Self { - outcome: Some(outcome), + script: Arc::new(Mutex::new(script)), calls: Arc::new(Mutex::new(Vec::new())), } } - fn erroring() -> Self { - Self { - outcome: None, - calls: Arc::new(Mutex::new(Vec::new())), - } + fn returning(outcome: SetTargetOutcome) -> Self { + Self::scripted(vec![Some(outcome)]) } fn calls(&self) -> Vec { @@ -759,8 +774,13 @@ mod tests { deployment_chain_id, num_candidates, )); - match &self.outcome { - Some(outcome) => Ok(outcome.clone()), + 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), } } @@ -792,16 +812,23 @@ mod tests { } } - /// A worker queue that records reassessment jobs. + /// 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)), } } @@ -832,6 +859,13 @@ mod tests { 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, @@ -862,30 +896,37 @@ mod tests { } } - async fn run_handle_record( + /// 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, - bytes: Option<&[u8]>, - ) -> ( - Result, - Vec, - ) { + 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 result = handle_record( + let applied = apply_target_with_retries( registry, worker, &events, PROTOCOL_CHAIN_ID, requester(), - "eip155:42161", - bytes, + &target, + &mut shutdown_rx, ) .await; - (result, events_capture.events()) + (applied, events_capture.events()) } #[tokio::test] @@ -895,10 +936,9 @@ mod tests { }); let worker = MockWorker::new(); - let bytes = valid_propose_bytes(); - let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + let (applied, events) = run_apply(®istry, &worker, 3).await; - assert!(matches!(result, Ok(Disposition::Applied)), "{result:?}"); + assert!(applied); assert_eq!( registry.calls(), vec![(requester(), deployment_id!(QM_HASH), INDEXED_CHAIN_ID, 3)] @@ -935,10 +975,9 @@ mod tests { }); let worker = MockWorker::new(); - let bytes = valid_propose_bytes(); - let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + let (applied, events) = run_apply(®istry, &worker, 3).await; - assert!(matches!(result, Ok(Disposition::Applied)), "{result:?}"); + assert!(applied); assert!(worker.reassessments().is_empty(), "no-op must not reassess"); assert!(events.is_empty(), "no-op must not emit events"); } @@ -950,10 +989,9 @@ mod tests { }); let worker = MockWorker::new(); - let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "eip155:1", 0); - let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + let (applied, events) = run_apply(®istry, &worker, 0).await; - assert!(matches!(result, Ok(Disposition::Applied)), "{result:?}"); + assert!(applied); assert_eq!(registry.calls()[0].3, 0, "the registry saw the 0 count"); let reassessments = worker.reassessments(); @@ -968,33 +1006,87 @@ mod tests { ); } - #[tokio::test] - async fn an_unprocessable_record_is_skipped_without_touching_the_registry() { - let registry = MockRegistry::returning(SetTargetOutcome::NoOp { + #[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::new(); + let worker = MockWorker::failing_pushes(2); - let bytes = encode_propose(EVENT_TYPE_TERMINATE, "eip155:42161", QM_HASH, "eip155:1", 3); - let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + let (applied, events) = run_apply(®istry, &worker, 3).await; - assert!(matches!(result, Ok(Disposition::Skipped(_))), "{result:?}"); - assert!(registry.calls().is_empty()); - assert!(worker.reassessments().is_empty()); - assert!(events.is_empty()); + 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 a_registry_failure_is_a_retryable_error_not_a_skip() { - let registry = MockRegistry::erroring(); + 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 bytes = valid_propose_bytes(); - let (result, events) = run_handle_record(®istry, &worker, Some(&bytes)).await; + 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(), + &target, + &mut shutdown_rx, + ) + .await; - assert!(result.is_err(), "a registry failure must surface for retry"); + assert!(!applied, "shutdown must win over the retry loop"); assert!(worker.reassessments().is_empty()); - assert!(events.is_empty()); + assert!(events_capture.events().is_empty()); } // -------- Redpanda-backed service test -------- diff --git a/bin/dipper-service/src/set_indexing_target.rs b/bin/dipper-service/src/set_indexing_target.rs index 22405f1a..839290fc 100644 --- a/bin/dipper-service/src/set_indexing_target.rs +++ b/bin/dipper-service/src/set_indexing_target.rs @@ -39,8 +39,9 @@ pub enum ApplyError { #[error("failed to set indexing target candidates")] Registry(#[source] crate::registry::Error), - /// The row changed but the reassessment job could not be queued; a later - /// call with the same target is a registry no-op yet queues the job again. + /// 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, From 6affafa8cc5e47fd3251e2b391c6a86a185c12d8 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 24 Aug 2026 18:33:07 +0300 Subject: [PATCH 6/7] fix(consumer): survive topic growth, log truncation, and bad counts Review hardening: a topic that gains partitions now restarts the consumer so nothing reads from the new ones unseen; a cursor past a truncated log re-anchors to the nearest live edge instead of replaying everything; wire counts are capped; the consumer now defaults to off. --- .github/workflows/ci.yml | 2 + bin/dipper-service/src/config.rs | 66 ++++- bin/dipper-service/src/main.rs | 10 + .../service/indexing_request_consumer.rs | 247 ++++++++++++++---- dipper-producer/tests/it_kafka_redpanda.rs | 5 + 5 files changed, 275 insertions(+), 55 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 76862e53..7a38f692 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -111,6 +111,7 @@ jobs: 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_)' @@ -120,6 +121,7 @@ jobs: 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/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index baaab895..432f7a6b 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -1380,8 +1380,11 @@ pub fn default_event_queue_capacity() -> NonZeroUsize { #[derive(Debug, Clone, serde::Deserialize)] #[serde(deny_unknown_fields)] pub struct IndexingRequestConsumerConfig { - /// Whether the consumer is enabled (default: true when the section is present) - #[serde(default = "default_indexing_request_consumer_enabled")] + /// 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 @@ -1404,10 +1407,6 @@ pub struct IndexingRequestConsumerConfig { pub fetch_max_bytes: i32, } -fn default_indexing_request_consumer_enabled() -> bool { - true -} - fn default_indexing_request_consumer_max_wait() -> Duration { Duration::from_secs(5) } @@ -1417,10 +1416,13 @@ fn default_indexing_request_consumer_fetch_max_bytes() -> i32 { } impl IndexingRequestConsumerConfig { - /// Reject a configuration the consumer cannot run with. + /// 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.enabled { - return Ok(()); + if self.kafka.brokers.is_empty() { + return Err( + "indexing_request_consumer.kafka.brokers must list at least 1 broker".to_string(), + ); } // A zero requester is almost certainly an unset value, and it would // silently key every consumed request under the zero address. @@ -1452,6 +1454,52 @@ impl IndexingRequestConsumerConfig { 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 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 93efe7a2..9386adf0 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -583,9 +583,19 @@ pub async fn main() -> anyhow::Result<()> { 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, diff --git a/bin/dipper-service/src/network/service/indexing_request_consumer.rs b/bin/dipper-service/src/network/service/indexing_request_consumer.rs index 457982c7..01562929 100644 --- a/bin/dipper-service/src/network/service/indexing_request_consumer.rs +++ b/bin/dipper-service/src/network/service/indexing_request_consumer.rs @@ -40,6 +40,13 @@ 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); + /// Handle for controlling the indexing request consumer lifecycle #[derive(Clone)] pub struct Handle { @@ -88,6 +95,10 @@ pub struct Ctx { /// 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, } @@ -128,6 +139,7 @@ where ctx.worker_queue.clone(), Arc::clone(&ctx.events), ctx.protocol_chain_id, + ctx.max_candidates, ctx.config.clone(), shutdown_rx.clone(), )); @@ -135,20 +147,39 @@ where // 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. - let result = tokio::select! { - _ = rx_stop.recv() => Ok(()), - joined = tasks.join_next() => 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" - )), - }, + // 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() => match consumer.current_partition_count().await { + 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() + )), + Ok(_) => {} + Err(err) => { + tracing::warn!(error = %err, "failed to re-check topic partition metadata"); + } + }, + } }; let _ = shutdown_tx.send(true); @@ -203,8 +234,11 @@ async fn connect_with_retries( } /// Consumes one partition sequentially: fetch from the persisted offset, apply -/// each record, then persist the offset past it (at-least-once; redelivery is -/// safe because an unchanged target count is a registry no-op). +/// 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. #[allow(clippy::too_many_arguments)] async fn partition_loop( consumer: Arc, @@ -213,6 +247,7 @@ async fn partition_loop( worker_queue: W, events: Arc, protocol_chain_id: ChainId, + max_candidates: usize, config: IndexingRequestConsumerConfig, mut shutdown_rx: watch::Receiver, ) where @@ -258,24 +293,46 @@ async fn partition_loop( fetch = consumer.fetch(partition, next_offset, config.fetch_max_bytes, max_wait_ms) => fetch, }; - let records = match fetch { - Ok((records, _high_watermark)) => records, + let (records, high_watermark) = match fetch { + Ok((records, high_watermark)) => (records, high_watermark), Err(err) if err.is_offset_out_of_range() => { - // Retention deleted records under the cursor; re-anchor to the - // earliest still-available record rather than spinning forever. - match consumer.offset(partition, OffsetAt::Earliest).await { - Ok(earliest) => { + // 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 = earliest; + next_offset = re_anchored; persist_offset(®istry, &topic, partition, next_offset).await; } - Err(err) => { - tracing::error!(partition, error = %err, "failed to query earliest offset"); + None => { if sleep_or_shutdown(&mut shutdown_rx, RETRY_BACKOFF).await { return; } @@ -292,6 +349,7 @@ async fn partition_loop( } }; + 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) { @@ -319,7 +377,10 @@ async fn partition_loop( &events, protocol_chain_id, config.requested_by, + max_candidates, &target, + partition, + offset, &mut shutdown_rx, ) .await; @@ -334,6 +395,21 @@ async fn partition_loop( 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; + } + } } } @@ -373,19 +449,39 @@ async fn sleep_or_shutdown(shutdown_rx: &mut watch::Receiver, delay: Durat /// (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, @@ -396,7 +492,7 @@ where requested_by, deployment_id: target.deployment_id, deployment_chain_id: target.deployment_chain_id, - num_candidates: target.num_candidates, + num_candidates, // Interactive: a developer just asked for this in Studio. priority: JobPriority::Interactive, }, @@ -405,7 +501,13 @@ where match result { Ok(_) => return true, - Err(ApplyError::Registry(_)) => { + 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; } @@ -416,26 +518,45 @@ where loop { if sleep_or_shutdown(shutdown_rx, RETRY_BACKOFF).await { - return false; + // 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; } - match worker_queue - .reassess_indexing_request( - queue_retry_id, - target.deployment_id, - target.deployment_chain_id, - target.num_candidates, - JobPriority::Interactive, - ) - .await - { - Ok(_) => return true, - Err(err) => { - tracing::error!( - indexing_request_id = %queue_retry_id, - error = ?err, - "retrying the reassessment job push" - ); - } + 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 } } } @@ -535,6 +656,9 @@ mod tests { const QM_HASH: &str = "QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"; + /// Ceiling on requested indexer counts in these tests. + const TEST_MAX_CANDIDATES: usize = 10; + fn requester() -> Address { "0x8f8c426f956876325b1e037c6eae9b189952994c" .parse() @@ -921,7 +1045,10 @@ mod tests { &events, PROTOCOL_CHAIN_ID, requester(), + TEST_MAX_CANDIDATES, &target, + 0, + 0, &mut shutdown_rx, ) .await; @@ -1054,6 +1181,24 @@ mod tests { ); } + #[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. @@ -1079,7 +1224,10 @@ mod tests { &events, PROTOCOL_CHAIN_ID, requester(), + TEST_MAX_CANDIDATES, &target, + 0, + 0, &mut shutdown_rx, ) .await; @@ -1106,6 +1254,11 @@ mod tests { .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 @@ -1208,6 +1361,7 @@ mod tests { sasl_password: None, tls_enabled: false, tls_ca_cert_path: None, + connect_timeout_secs: 60, }, requested_by, max_wait: Duration::from_secs(1), @@ -1222,6 +1376,7 @@ mod tests { worker_queue: worker, events, protocol_chain_id: 42161, + max_candidates: 10, config: consumer_config.clone(), }); (handle, tokio::spawn(service)) diff --git a/dipper-producer/tests/it_kafka_redpanda.rs b/dipper-producer/tests/it_kafka_redpanda.rs index 9f6c0beb..c1c7129c 100644 --- a/dipper-producer/tests/it_kafka_redpanda.rs +++ b/dipper-producer/tests/it_kafka_redpanda.rs @@ -14,6 +14,11 @@ fn brokers() -> Option> { .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 From 37815110d1edb016b7c5e6bef7bf7c906096dc0f Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Mon, 24 Aug 2026 18:52:42 +0300 Subject: [PATCH 7/7] fix(consumer): stop restarting the dipper over a stale metadata read Kafka partitions only ever grow, so a metadata response reporting fewer than we serve is a stale read, not a topology change; it is now logged instead of restarting the whole process. The read itself is bounded to 3 seconds so it cannot stall shutdown, and a 0 connect timeout is rejected. --- bin/dipper-service/src/config.rs | 19 +++++++++ .../service/indexing_request_consumer.rs | 41 +++++++++++++++---- 2 files changed, 51 insertions(+), 9 deletions(-) diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index 432f7a6b..f681685b 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -1424,6 +1424,14 @@ impl IndexingRequestConsumerConfig { "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 { @@ -1487,6 +1495,17 @@ mod tests { })); 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", diff --git a/bin/dipper-service/src/network/service/indexing_request_consumer.rs b/bin/dipper-service/src/network/service/indexing_request_consumer.rs index 01562929..8cb4bdb7 100644 --- a/bin/dipper-service/src/network/service/indexing_request_consumer.rs +++ b/bin/dipper-service/src/network/service/indexing_request_consumer.rs @@ -47,6 +47,10 @@ 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 { @@ -168,15 +172,34 @@ where "the indexing request consumer had no partition loops to run" )), }, - _ = metadata_check.tick() => match consumer.current_partition_count().await { - 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() - )), - Ok(_) => {} - Err(err) => { - tracing::warn!(error = %err, "failed to re-check topic partition metadata"); + _ = 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"); + } } }, }