diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7bf97145..7a38f692 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -88,10 +88,30 @@ jobs: sudo apt-get install -y postgresql-17 echo "/usr/lib/postgresql/$(pg_lsclusters | awk 'NR==2 {print $1}')/bin" >> $GITHUB_PATH + # The Kafka roundtrip and consumer tests skip themselves when + # REDPANDA_BROKERS is unset, so the broker must be up before they run. + - name: Start Redpanda for Kafka integration tests + run: | + docker run -d --name redpanda -p 9092:9092 redpandadata/redpanda:v24.2.7 \ + redpanda start --mode dev-container --smp 1 --memory 1G \ + --kafka-addr PLAINTEXT://0.0.0.0:9092 \ + --advertise-kafka-addr PLAINTEXT://localhost:9092 + for _ in $(seq 1 30); do + if docker exec redpanda rpk cluster health 2>/dev/null | grep -q 'Healthy:.*true'; then + exit 0 + fi + sleep 1 + done + echo "Redpanda did not become healthy in time" >&2 + docker logs redpanda >&2 + exit 1 + - name: Integration tests (in-tree) uses: LNSD/sops-exec-action@6da1fbca63459d9796097496d5f5e6233555b31a # v1 env: SOPS_AGE_KEY: ${{ secrets.IT_TESTS_AGE_KEY }} + REDPANDA_BROKERS: localhost:9092 + REQUIRE_REDPANDA: '1' with: env_file: .env run: cargo nextest run --all-features -E 'test(~tests::it_)' @@ -100,6 +120,8 @@ jobs: uses: LNSD/sops-exec-action@6da1fbca63459d9796097496d5f5e6233555b31a # v1 env: SOPS_AGE_KEY: ${{ secrets.IT_TESTS_AGE_KEY }} + REDPANDA_BROKERS: localhost:9092 + REQUIRE_REDPANDA: '1' with: env_file: .env run: cargo nextest run --all-features -E 'kind(test)' diff --git a/Cargo.lock b/Cargo.lock index 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/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/config.rs b/bin/dipper-service/src/config.rs index d9e0bad9..f681685b 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,10 +1373,152 @@ pub fn default_event_queue_capacity() -> NonZeroUsize { NonZeroUsize::new(1024).expect("default event queue capacity is non-zero") } +/// Configuration for the Studio indexing request consumer, which reads +/// subgraph indexing request events from a Redpanda topic and applies them +/// through the same path as the admin RPC. +#[serde_as] +#[derive(Debug, Clone, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct IndexingRequestConsumerConfig { + /// Whether the consumer is enabled (default: false). Leave off until + /// Studio's propose events carry `indexed_network_caip2id`: without the + /// field every consumed request is skipped and its offset committed, so + /// enabling early permanently discards real requests. + #[serde(default)] + pub enabled: bool, + + /// Kafka connection settings and the topic Studio produces on. The topic + /// is required with no default; a wrong or missing name fails at startup. + pub kafka: KafkaConsumerConfig, + + /// The identity recorded as the requester on consumed requests, since + /// Kafka messages carry no signature to recover one from. Use the address + /// Studio signs the admin RPC with, so both doors share request rows. + pub requested_by: Address, + + /// How long a fetch waits server-side for new records before returning + /// empty, in seconds (default: 5). + #[serde_as(as = "serde_with::DurationSeconds")] + #[serde(default = "default_indexing_request_consumer_max_wait")] + pub max_wait: Duration, + + /// Maximum bytes per fetch (default: 1,048,576). + #[serde(default = "default_indexing_request_consumer_fetch_max_bytes")] + pub fetch_max_bytes: i32, +} + +fn default_indexing_request_consumer_max_wait() -> Duration { + Duration::from_secs(5) +} + +fn default_indexing_request_consumer_fetch_max_bytes() -> i32 { + 1_048_576 +} + +impl IndexingRequestConsumerConfig { + /// Reject a configuration the consumer cannot run with. Checked even when + /// disabled, so a broken value cannot lie dormant until the flag flips. + pub fn validate(&self) -> Result<(), String> { + if self.kafka.brokers.is_empty() { + return Err( + "indexing_request_consumer.kafka.brokers must list at least 1 broker".to_string(), + ); + } + // 0 makes every connect time out instantly, so startup would fail with + // a message blaming the broker instead of the config typo. + if self.kafka.connect_timeout_secs == 0 { + return Err( + "indexing_request_consumer.kafka.connect_timeout_secs must be at least 1" + .to_string(), + ); + } + // A zero requester is almost certainly an unset value, and it would + // silently key every consumed request under the zero address. + if self.requested_by == Address::ZERO { + return Err( + "indexing_request_consumer.requested_by must be a non-zero address".to_string(), + ); + } + // Below ~1 KB a fetch cannot hold a whole record, so the consumer + // would poll forever without ever making progress. + if self.fetch_max_bytes < 1_024 { + return Err(format!( + "indexing_request_consumer.fetch_max_bytes ({}) must be at least 1024", + self.fetch_max_bytes + )); + } + if self.max_wait.is_zero() || self.max_wait.as_millis() > i32::MAX as u128 { + return Err(format!( + "indexing_request_consumer.max_wait ({}s) must be between 1 second and {} seconds", + self.max_wait.as_secs(), + i32::MAX / 1_000 + )); + } + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; + fn consumer_config(json: serde_json::Value) -> IndexingRequestConsumerConfig { + serde_json::from_value(json).expect("deserializes") + } + + #[test] + fn indexing_request_consumer_config_defaults_and_validation() { + let config = consumer_config(serde_json::json!({ + "kafka": { "brokers": ["localhost:9092"], "topic": "t" }, + "requested_by": "0x8f8c426f956876325b1e037c6eae9b189952994c", + })); + assert!( + !config.enabled, + "the consumer must be off unless opted into" + ); + assert_eq!(config.max_wait, Duration::from_secs(5)); + assert_eq!(config.fetch_max_bytes, 1_048_576); + assert!(config.validate().is_ok()); + + // Validation runs even for a disabled section, so broken values are + // caught at startup rather than the day the flag flips. + let broken = consumer_config(serde_json::json!({ + "enabled": false, + "kafka": { "brokers": ["localhost:9092"], "topic": "t" }, + "requested_by": "0x0000000000000000000000000000000000000000", + })); + assert!(broken.validate().unwrap_err().contains("requested_by")); + + let no_brokers = consumer_config(serde_json::json!({ + "kafka": { "brokers": [], "topic": "t" }, + "requested_by": "0x8f8c426f956876325b1e037c6eae9b189952994c", + })); + assert!(no_brokers.validate().unwrap_err().contains("brokers")); + + let zero_connect_timeout = consumer_config(serde_json::json!({ + "kafka": { "brokers": ["localhost:9092"], "topic": "t", "connect_timeout_secs": 0 }, + "requested_by": "0x8f8c426f956876325b1e037c6eae9b189952994c", + })); + assert!( + zero_connect_timeout + .validate() + .unwrap_err() + .contains("connect_timeout_secs") + ); + + let tiny_fetch = consumer_config(serde_json::json!({ + "kafka": { "brokers": ["localhost:9092"], "topic": "t" }, + "requested_by": "0x8f8c426f956876325b1e037c6eae9b189952994c", + "fetch_max_bytes": 512, + })); + assert!( + tiny_fetch + .validate() + .unwrap_err() + .contains("fetch_max_bytes") + ); + } + #[test] fn test_dips_agreement_config_deserialization() { //* Arrange - JSON config with all new field names diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 93a4460b..9386adf0 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)] @@ -81,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}"); } @@ -567,6 +573,34 @@ pub async fn main() -> anyhow::Result<()> { None }; + //- The Studio indexing request consumer service (optional, enabled by config) + // Reads subgraph indexing requests from Redpanda and applies them through + // the same path as the admin RPC. + let indexing_request_consumer_handle = match conf.indexing_request_consumer { + Some(ref consumer_conf) if consumer_conf.enabled => { + let ctx = network::service::indexing_request_consumer::Ctx { + registry: registry.clone(), + worker_queue: worker_handle.queue().clone(), + events: subgraph_indexing_agreements_events_emitter.clone(), + protocol_chain_id: chain_id, + max_candidates: DEFAULT_MAX_CANDIDATES, + config: consumer_conf.clone(), + }; + let (handle, service) = network::service::indexing_request_consumer::new(ctx); + // A reassessment push lost at the shutdown edge is only repaired by + // the periodic reassignment sweep; without it the request would sit + // open with no job behind it. + if !conf.reassignment.as_ref().is_some_and(|r| r.enabled) { + tracing::warn!( + "the indexing request consumer is enabled without the reassignment service; \ + a reassessment job lost during a shutdown would never be retried" + ); + } + Some((handle, service)) + } + _ => None, + }; + //- The admin RPC service let (admin_rpc_handle, admin_rpc_service) = { let config = admin_rpc_server::service::Config { @@ -678,6 +712,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); @@ -754,6 +798,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..8cb4bdb7 --- /dev/null +++ b/bin/dipper-service/src/network/service/indexing_request_consumer.rs @@ -0,0 +1,1495 @@ +//! Consumes the subgraph indexing request events Studio produces on Redpanda +//! and applies each one as a set-indexing-target change, making the topic a +//! second front door to the same path the admin RPC serves. + +use std::{future::Future, sync::Arc, time::Duration}; + +use async_trait::async_trait; +use dipper_producer::{ + events::SubgraphIndexingAgreementEventsProducer, + kafka::{ConsumerError, KafkaConsumer, OffsetAt}, + prost::Message as _, + proto::studio, +}; +use thegraph_core::{ + DeploymentId, + alloy::primitives::{Address, ChainId}, +}; +use tokio::{ + sync::{mpsc, watch}, + task::JoinSet, +}; + +use crate::{ + config::IndexingRequestConsumerConfig, + registry::IndexingRequestRegistry, + set_indexing_target::{ApplyError, SetIndexingTarget, apply_set_indexing_target}, + worker::service::{JobPriority, WorkerQueue}, +}; + +/// The propose event type Studio sends. +const EVENT_TYPE_PROPOSE: &str = "subgraph.indexing.request.propose"; + +/// Studio's producer also defines a terminate event, but nothing sends it: +/// cancellation is a propose with a count of 0. Logged and skipped if seen. +const EVENT_TYPE_TERMINATE: &str = "subgraph.indexing.agreements.terminate"; + +/// Extra connection attempts after the first before startup fails visibly. +const CONNECT_MAX_RETRIES: u32 = 5; + +/// Delay before retrying after a fetch or apply failure. +const RETRY_BACKOFF: Duration = Duration::from_secs(5); + +/// Pause after a fetch below the high watermark that returned no usable +/// records, so a run of dropped batches cannot spin the loop hot. +const EMPTY_FETCH_PAUSE: Duration = Duration::from_secs(1); + +/// How often to re-read topic metadata to notice a partition count change. +const PARTITION_METADATA_CHECK_INTERVAL: Duration = Duration::from_secs(300); + +/// Bound on one metadata re-read. Deliberately shorter than the 5-second stop +/// cap, since the stop channel is not polled while the read is in flight. +const PARTITION_METADATA_CHECK_TIMEOUT: Duration = Duration::from_secs(3); + +/// Handle for controlling the indexing request consumer lifecycle +#[derive(Clone)] +pub struct Handle { + tx_stop: mpsc::Sender<()>, +} + +impl Handle { + /// Stop the consumer gracefully + pub async fn stop(&self) { + if self.tx_stop.is_closed() { + return; + } + + let _ = self.tx_stop.send(()).await; + self.tx_stop.closed().await; + } +} + +/// Registry for persisting per-partition consumer progress. +#[async_trait] +pub trait KafkaConsumerOffsetRegistry { + /// Get the next offset to fetch for a topic partition, `None` on first run. + async fn get_kafka_consumer_offset( + &self, + topic: &str, + partition_id: i32, + ) -> Result, crate::registry::Error>; + + /// Record the next offset to fetch for a topic partition. + async fn set_kafka_consumer_offset( + &self, + topic: &str, + partition_id: i32, + next_offset: i64, + ) -> Result<(), crate::registry::Error>; +} + +/// Context required by the indexing request consumer service +pub struct Ctx { + /// Registry for indexing requests and consumer offsets + pub registry: R, + /// Worker queue for the reassessment jobs that follow an applied request + pub worker_queue: W, + /// Lifecycle events emitter (request-received on newly inserted requests) + pub events: Arc, + /// The protocol network chain id (signer chain id), for validating the + /// envelope's network and stamping emitted lifecycle events + pub protocol_chain_id: ChainId, + /// Ceiling on the indexer count a consumed request may ask for; larger + /// counts are clamped with a warning. Kafka records carry no signature, + /// so this door gets a cap the signed admin RPC does not need. + pub max_candidates: usize, + /// Service configuration + pub config: IndexingRequestConsumerConfig, +} + +/// Create a new indexing request consumer service: a control handle plus a +/// future to spawn that reads Studio's propose events from Kafka, applies each +/// as a set-indexing-target change, and records its progress per partition. +pub fn new(ctx: Ctx) -> (Handle, impl Future>) +where + R: IndexingRequestRegistry + KafkaConsumerOffsetRegistry + Clone + Send + Sync + 'static, + W: WorkerQueue + Clone + Send + Sync + 'static, +{ + let (tx_stop, mut rx_stop) = mpsc::channel(1); + + let service = async move { + let consumer = match connect_with_retries(&ctx.config, &mut rx_stop).await { + Ok(Some(consumer)) => Arc::new(consumer), + // Stop was requested while still connecting; a clean exit. + Ok(None) => return Ok(()), + Err(err) => return Err(err), + }; + + let partitions = consumer.partitions(); + tracing::info!( + topic = consumer.topic(), + partitions = partitions.len(), + requested_by = %ctx.config.requested_by, + "indexing request consumer started" + ); + + let (shutdown_tx, shutdown_rx) = watch::channel(false); + let mut tasks: JoinSet<()> = JoinSet::new(); + for partition in partitions { + tasks.spawn(partition_loop( + Arc::clone(&consumer), + partition, + ctx.registry.clone(), + ctx.worker_queue.clone(), + Arc::clone(&ctx.events), + ctx.protocol_chain_id, + ctx.max_candidates, + ctx.config.clone(), + shutdown_rx.clone(), + )); + } + + // Partition loops only return on shutdown, so one finishing early means + // it panicked or hit a bug; tear the service down so the process + // restarts instead of consuming a partial set of partitions. The + // metadata timer notices a topic growing partitions: new partitions + // would otherwise be consumed by nobody, silently losing requests, so + // that also restarts the service to pick up the full layout. + let serving = tasks.len(); + let mut metadata_check = tokio::time::interval(PARTITION_METADATA_CHECK_INTERVAL); + metadata_check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let result = loop { + tokio::select! { + _ = rx_stop.recv() => break Ok(()), + joined = tasks.join_next() => break match joined { + Some(Ok(())) => Err(anyhow::anyhow!( + "an indexing request consumer partition loop exited unexpectedly" + )), + Some(Err(err)) => Err(anyhow::anyhow!( + "an indexing request consumer partition loop panicked: {err}" + )), + None => Err(anyhow::anyhow!( + "the indexing request consumer had no partition loops to run" + )), + }, + _ = metadata_check.tick() => { + let count = tokio::time::timeout( + PARTITION_METADATA_CHECK_TIMEOUT, + consumer.current_partition_count(), + ) + .await; + match count { + Ok(Ok(count)) if count > serving => break Err(anyhow::anyhow!( + "topic '{}' now has {count} partitions but this consumer serves \ + {serving}; restarting to consume the full set", + consumer.topic() + )), + // Kafka partitions only ever grow, so a lower count is a + // stale or partial metadata read, never a real change; + // restarting the process over it would be a false alarm. + Ok(Ok(count)) if count < serving => tracing::warn!( + count, + serving, + "metadata reported fewer partitions than this consumer serves; \ + ignoring the stale read" + ), + Ok(Ok(_)) => {} + Ok(Err(err)) => { + tracing::warn!(error = %err, "failed to re-check topic partition metadata"); + } + Err(_) => { + tracing::warn!("topic partition metadata re-check timed out"); + } + } + }, + } + }; + + let _ = shutdown_tx.send(true); + while tasks.join_next().await.is_some() {} + + tracing::info!("indexing request consumer stopped"); + result + }; + + (Handle { tx_stop }, service) +} + +/// Connects to the brokers, retrying transient failures a bounded number of +/// times. A missing topic fails immediately: it is configuration, and reading +/// from a wrong or absent topic must be loud, not an idle consumer. +async fn connect_with_retries( + config: &IndexingRequestConsumerConfig, + rx_stop: &mut mpsc::Receiver<()>, +) -> anyhow::Result> { + let mut attempt: u32 = 0; + loop { + match KafkaConsumer::connect(&config.kafka).await { + Ok(consumer) => return Ok(Some(consumer)), + Err(err @ ConsumerError::TopicNotFound { .. }) => { + return Err(anyhow::anyhow!( + "indexing request consumer startup failed: {err}; check the configured topic \ + name against the topic Studio produces on" + )); + } + Err(err) if attempt < CONNECT_MAX_RETRIES => { + attempt += 1; + let delay = Duration::from_secs(2u64.pow(attempt.min(5))); + tracing::warn!( + attempt, + delay_secs = delay.as_secs(), + error = %err, + "indexing request consumer connect failed, retrying" + ); + tokio::select! { + _ = rx_stop.recv() => return Ok(None), + _ = tokio::time::sleep(delay) => {} + } + } + Err(err) => { + return Err(anyhow::anyhow!( + "indexing request consumer failed to connect after {} attempts: {err}", + CONNECT_MAX_RETRIES + 1 + )); + } + } + } +} + +/// Consumes one partition sequentially: fetch from the persisted offset, apply +/// each record, then persist the offset past it. Delivery is at-least-once; +/// redelivering an open request's count is a registry no-op, though a replay +/// reaching back past a cancellation briefly re-opens it until the cancel +/// replays too, which is why re-anchoring below never rewinds further than +/// the broker forces it to. +#[allow(clippy::too_many_arguments)] +async fn partition_loop( + consumer: Arc, + partition: i32, + registry: R, + worker_queue: W, + events: Arc, + protocol_chain_id: ChainId, + max_candidates: usize, + config: IndexingRequestConsumerConfig, + mut shutdown_rx: watch::Receiver, +) where + R: IndexingRequestRegistry + KafkaConsumerOffsetRegistry + Send + Sync, + W: WorkerQueue + Send + Sync, +{ + let topic = consumer.topic().to_string(); + let expected_network = format!("eip155:{protocol_chain_id}"); + let max_wait_ms = 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, high_watermark) = match fetch { + Ok((records, high_watermark)) => (records, high_watermark), + Err(err) if err.is_offset_out_of_range() => { + // The broker refuses a cursor outside its retained range: below + // the log start when retention deleted records, or above the + // log end when the topic was recreated or truncated. Clamp to + // the nearest live edge; always rewinding to earliest would + // replay the whole retained topic in the truncation case. + let range = match consumer.offset(partition, OffsetAt::Earliest).await { + Ok(earliest) => match consumer.offset(partition, OffsetAt::Latest).await { + Ok(latest) => Some((earliest, latest)), + Err(err) => { + tracing::error!(partition, error = %err, "failed to query latest offset"); + None + } + }, + Err(err) => { + tracing::error!(partition, error = %err, "failed to query earliest offset"); + None + } + }; + match range { + Some((earliest, latest)) => { + let re_anchored = if next_offset < earliest { + earliest + } else { + latest + }; + tracing::warn!( + partition, + stale_offset = next_offset, + earliest, + latest, + re_anchored, + "consumer offset fell outside the broker's retained range; re-anchoring" + ); + next_offset = re_anchored; + persist_offset(®istry, &topic, partition, next_offset).await; + } + None => { + if sleep_or_shutdown(&mut shutdown_rx, RETRY_BACKOFF).await { + return; + } + } + } + continue; + } + Err(err) => { + tracing::error!(partition, error = %err, "fetch failed, backing off"); + if sleep_or_shutdown(&mut shutdown_rx, RETRY_BACKOFF).await { + return; + } + continue; + } + }; + + let records_was_empty = records.is_empty(); + for record_and_offset in records { + let offset = record_and_offset.offset; + match decode_propose(record_and_offset.record.value.as_deref(), &expected_network) { + Err(reason) => { + tracing::warn!( + topic, + partition, + offset, + reason, + key = ?record_and_offset.record.key.as_deref().map(String::from_utf8_lossy), + "skipping unprocessable indexing request record" + ); + } + Ok(target) => { + tracing::debug!( + event_id = %target.event_id, + deployment_id = %target.deployment_id, + deployment_chain_id = target.deployment_chain_id, + num_candidates = target.num_candidates, + "consumed indexing request propose event" + ); + let applied = apply_target_with_retries( + ®istry, + &worker_queue, + &events, + protocol_chain_id, + config.requested_by, + max_candidates, + &target, + partition, + offset, + &mut shutdown_rx, + ) + .await; + if !applied { + // Shutdown arrived before the record took effect; the + // unadvanced offset redelivers it on the next run. + return; + } + } + } + + next_offset = offset + 1; + persist_offset(®istry, &topic, partition, next_offset).await; + } + + // An empty fetch below the high watermark (e.g. a batch of records the + // client filtered out) would otherwise loop again instantly: pause so + // a run of them cannot spin hot. + if records_was_empty && high_watermark > next_offset { + tracing::debug!( + partition, + next_offset, + high_watermark, + "fetch below the high watermark returned no records; pausing" + ); + if sleep_or_shutdown(&mut shutdown_rx, EMPTY_FETCH_PAUSE).await { + return; + } + } + } +} + +/// Persist consumer progress. Failure is logged but does not halt consumption: +/// the in-memory cursor stays correct and a later write covers the gap, at the +/// cost of some redelivery after a restart (which is safe). +async fn persist_offset( + registry: &R, + topic: &str, + partition: i32, + next_offset: i64, +) { + if let Err(err) = registry + .set_kafka_consumer_offset(topic, partition, next_offset) + .await + { + tracing::error!( + topic, + partition, + next_offset, + error = %err, + "failed to persist consumer offset; progress will be re-delivered after a restart" + ); + } +} + +/// Wait out a backoff, returning `true` when shutdown was requested instead. +async fn sleep_or_shutdown(shutdown_rx: &mut watch::Receiver, delay: Duration) -> bool { + tokio::select! { + _ = shutdown_rx.changed() => true, + _ = tokio::time::sleep(delay) => false, + } +} + +/// Apply a decoded propose until it fully takes effect, returning `false` if +/// shutdown was requested first. A registry failure retries the whole apply +/// (nothing was committed); a queue failure retries only the job push, because +/// the row change is already committed and re-running the apply would land on +/// the registry's no-op path and silently drop the reassessment. +#[allow(clippy::too_many_arguments)] +async fn apply_target_with_retries( + registry: &R, + worker_queue: &W, + events: &Arc, + protocol_chain_id: ChainId, + requested_by: Address, + max_candidates: usize, + target: &ProposedTarget, + partition: i32, + offset: i64, + shutdown_rx: &mut watch::Receiver, +) -> bool +where + R: IndexingRequestRegistry + Send + Sync, + W: WorkerQueue + Send + Sync, +{ + // Cap the count from the wire: this door has no signature to vouch for the + // sender, so an absurd target must not reach indexer selection unclamped. + // 0 passes through untouched, since it is the cancellation shape. + let num_candidates = if target.num_candidates > max_candidates { + tracing::warn!( + partition, + offset, + requested = target.num_candidates, + max_candidates, + "clamping the requested indexer count to the configured maximum" + ); + max_candidates + } else { + target.num_candidates + }; + + let queue_retry_id = loop { + let result = apply_set_indexing_target( + registry, + worker_queue, + events, + protocol_chain_id, + SetIndexingTarget { + requested_by, + deployment_id: target.deployment_id, + deployment_chain_id: target.deployment_chain_id, + num_candidates, + // Interactive: a developer just asked for this in Studio. + priority: JobPriority::Interactive, + }, + ) + .await; + + match result { + Ok(_) => return true, + Err(err @ ApplyError::Registry(_)) => { + tracing::error!( + partition, + offset, + error = %err, + "failed to apply indexing request, retrying" + ); + if sleep_or_shutdown(shutdown_rx, RETRY_BACKOFF).await { + return false; + } + } + Err(ApplyError::QueueReassess { id, .. }) => break id, + } + }; + + loop { + if sleep_or_shutdown(shutdown_rx, RETRY_BACKOFF).await { + // One last immediate attempt: the row change is already committed, + // so leaving without the job strands the request until the daily + // reassignment sweep next queues it. + return retry_reassess_push(worker_queue, queue_retry_id, num_candidates, target).await; + } + if retry_reassess_push(worker_queue, queue_retry_id, num_candidates, target).await { + return true; + } + } +} + +/// One attempt at the reassessment push that failed inside the apply. +async fn retry_reassess_push( + worker_queue: &W, + id: dipper_core::ids::IndexingRequestId, + num_candidates: usize, + target: &ProposedTarget, +) -> bool +where + W: WorkerQueue + Send + Sync, +{ + match worker_queue + .reassess_indexing_request( + id, + target.deployment_id, + target.deployment_chain_id, + num_candidates, + JobPriority::Interactive, + ) + .await + { + Ok(_) => true, + Err(err) => { + tracing::error!( + indexing_request_id = %id, + error = ?err, + "retrying the reassessment job push" + ); + false + } + } +} + +/// A validated propose event, reduced to what the registry call needs. +#[derive(Debug, PartialEq, Eq)] +struct ProposedTarget { + event_id: String, + deployment_id: DeploymentId, + deployment_chain_id: ChainId, + num_candidates: usize, +} + +/// Decode a record value into a propose target, or the reason to skip it. +/// Unknown event types are tolerated by design: Studio may add types before +/// the dipper learns them, and they must not wedge the partition. +fn decode_propose( + value: Option<&[u8]>, + expected_network: &str, +) -> Result { + let Some(value) = value else { + return Err("empty record value"); + }; + + let event = match studio::SubgraphIndexingRequestEvent::decode(value) { + Ok(event) => event, + Err(_) => return Err("undecodable protobuf"), + }; + + match event.event_type.as_str() { + EVENT_TYPE_PROPOSE => {} + EVENT_TYPE_TERMINATE => return Err("terminate event (cancellation is a propose with 0)"), + _ => return Err("unknown event type"), + } + + if event.the_graph_network_caip2id != expected_network { + return Err("event is for a different protocol network"); + } + + let Some(studio::subgraph_indexing_request_event::Payload::SubgraphIndexingRequestPropose( + propose, + )) = event.payload + else { + return Err("propose event without a propose payload"); + }; + + let Ok(deployment_id) = event.subgraph_deployment_qm_hash.parse::() else { + return Err("invalid subgraph deployment hash"); + }; + + let Some(deployment_chain_id) = parse_eip155_caip2(&propose.indexed_network_caip2id) else { + // The field is the dipper's addition to Studio's schema; until Studio + // sends it, every message lands here and this warn is the signal. + return Err("missing or invalid indexed network caip2 id"); + }; + + let Ok(num_candidates) = usize::try_from(propose.indexing_agreements_requested) else { + return Err("negative indexing agreements requested"); + }; + + Ok(ProposedTarget { + event_id: event.event_id, + deployment_id, + deployment_chain_id, + num_candidates, + }) +} + +/// Parse an `eip155:{chain_id}` CAIP-2 identifier into its numeric chain id. +fn parse_eip155_caip2(value: &str) -> Option { + value.strip_prefix("eip155:")?.parse::().ok() +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use dipper_core::ids::{IndexingAgreementId, IndexingRequestId}; + use thegraph_core::{DeploymentId, deployment_id}; + use url::Url; + + use super::*; + use crate::{ + registry::{ + IndexingRequest as IndexingRequestRecord, Result as RegistryResult, SetTargetOutcome, + }, + test_support::{CapturedEvent, CapturingEventsProducer}, + worker::queue::JobId, + }; + + /// The protocol (signer) chain id, distinct from the indexed chain below + /// so assertions can tell the 2 apart. + const PROTOCOL_CHAIN_ID: ChainId = 42161; + + /// The chain the test deployment indexes. + const INDEXED_CHAIN_ID: ChainId = 1; + + const QM_HASH: &str = "QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"; + + /// Ceiling on requested indexer counts in these tests. + const TEST_MAX_CANDIDATES: usize = 10; + + fn requester() -> Address { + "0x8f8c426f956876325b1e037c6eae9b189952994c" + .parse() + .expect("valid address") + } + + /// Encode a propose envelope the way Studio's producer does. + fn encode_propose( + event_type: &str, + network: &str, + qm_hash: &str, + indexed_network: &str, + count: i32, + ) -> Vec { + let event = studio::SubgraphIndexingRequestEvent { + event_id: "01912345-6789-7abc-def0-123456789abc".to_string(), + event_type: event_type.to_string(), + event_version: "1.0".to_string(), + timestamp: "2026-08-24T10:30:00.123Z".to_string(), + subgraph_deployment_qm_hash: qm_hash.to_string(), + the_graph_network_caip2id: network.to_string(), + payload: Some( + studio::subgraph_indexing_request_event::Payload::SubgraphIndexingRequestPropose( + studio::SubgraphIndexingRequestPropose { + indexing_agreements_requested: count, + indexed_network_caip2id: indexed_network.to_string(), + }, + ), + ), + }; + event.encode_to_vec() + } + + fn valid_propose_bytes() -> Vec { + encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "eip155:1", 3) + } + + // -------- decode_propose -------- + + #[test] + fn decodes_a_valid_propose_event() { + let bytes = valid_propose_bytes(); + let target = decode_propose(Some(&bytes), "eip155:42161").expect("decodes"); + + assert_eq!(target.deployment_id, deployment_id!(QM_HASH)); + assert_eq!(target.deployment_chain_id, INDEXED_CHAIN_ID); + assert_eq!(target.num_candidates, 3); + assert_eq!(target.event_id, "01912345-6789-7abc-def0-123456789abc"); + } + + #[test] + fn tolerates_unknown_fields_appended_to_the_envelope() { + // A future schema revision adds fields this consumer does not know: + // field 15, wire type 2 (length-delimited), 3 bytes of payload. + let mut bytes = valid_propose_bytes(); + bytes.extend_from_slice(&[0x7A, 0x03, b'a', b'b', b'c']); + + let target = decode_propose(Some(&bytes), "eip155:42161").expect("decodes"); + assert_eq!(target.num_candidates, 3); + } + + #[test] + fn skips_a_zero_count_as_a_valid_cancellation() { + // Count 0 is not a skip: it is the agreed cancellation shape. + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "eip155:1", 0); + let target = decode_propose(Some(&bytes), "eip155:42161").expect("decodes"); + assert_eq!(target.num_candidates, 0); + } + + #[test] + fn rejects_an_empty_record_value() { + assert_eq!( + decode_propose(None, "eip155:42161"), + Err("empty record value") + ); + } + + #[test] + fn rejects_undecodable_bytes() { + // 0xFF is a field-15 wire-type-7 tag; wire type 7 does not exist. + let garbage = [0xFF, 0xFF, 0xFF]; + assert_eq!( + decode_propose(Some(&garbage), "eip155:42161"), + Err("undecodable protobuf") + ); + } + + #[test] + fn rejects_an_unknown_event_type() { + let bytes = encode_propose( + "subgraph.indexing.request.some_future_type", + "eip155:42161", + QM_HASH, + "eip155:1", + 3, + ); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("unknown event type") + ); + } + + #[test] + fn rejects_a_terminate_event() { + let bytes = encode_propose(EVENT_TYPE_TERMINATE, "eip155:42161", QM_HASH, "eip155:1", 3); + assert!(matches!( + decode_propose(Some(&bytes), "eip155:42161"), + Err(reason) if reason.contains("terminate") + )); + } + + #[test] + fn rejects_an_event_for_another_protocol_network() { + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:421614", QM_HASH, "eip155:1", 3); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("event is for a different protocol network") + ); + } + + #[test] + fn rejects_a_propose_without_a_payload() { + let event = studio::SubgraphIndexingRequestEvent { + event_id: "e".to_string(), + event_type: EVENT_TYPE_PROPOSE.to_string(), + event_version: "1.0".to_string(), + timestamp: "t".to_string(), + subgraph_deployment_qm_hash: QM_HASH.to_string(), + the_graph_network_caip2id: "eip155:42161".to_string(), + payload: None, + }; + assert_eq!( + decode_propose(Some(&event.encode_to_vec()), "eip155:42161"), + Err("propose event without a propose payload") + ); + } + + #[test] + fn rejects_an_invalid_deployment_hash() { + let bytes = encode_propose( + EVENT_TYPE_PROPOSE, + "eip155:42161", + "not-a-deployment-hash", + "eip155:1", + 3, + ); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("invalid subgraph deployment hash") + ); + } + + #[test] + fn rejects_a_missing_indexed_network() { + // Studio has not added the field yet: it decodes as an empty string. + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "", 3); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("missing or invalid indexed network caip2 id") + ); + } + + #[test] + fn rejects_a_malformed_indexed_network() { + for indexed in ["cosmos:hub", "eip155:", "eip155:abc", "1"] { + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, indexed, 3); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("missing or invalid indexed network caip2 id"), + "indexed network {indexed:?} should be rejected" + ); + } + } + + #[test] + fn rejects_a_negative_candidate_count() { + let bytes = encode_propose(EVENT_TYPE_PROPOSE, "eip155:42161", QM_HASH, "eip155:1", -1); + assert_eq!( + decode_propose(Some(&bytes), "eip155:42161"), + Err("negative indexing agreements requested") + ); + } + + #[test] + fn parses_eip155_caip2_ids() { + assert_eq!(parse_eip155_caip2("eip155:1"), Some(1)); + assert_eq!(parse_eip155_caip2("eip155:42161"), Some(42161)); + assert_eq!(parse_eip155_caip2("eip155:"), None); + assert_eq!(parse_eip155_caip2("eip155:1x"), None); + assert_eq!(parse_eip155_caip2("solana:1"), None); + assert_eq!(parse_eip155_caip2(""), None); + } + + // -------- handle_record -------- + + type SetTargetCall = (Address, DeploymentId, ChainId, usize); + type ReassessCall = (IndexingRequestId, DeploymentId, ChainId, usize); + + /// A registry whose `set_indexing_target_candidates` pops the next scripted + /// response (erroring once the script runs out) and records its arguments. + #[derive(Clone)] + struct MockRegistry { + script: Arc>>>, + calls: Arc>>, + } + + impl MockRegistry { + /// `None` entries are errors; after the script is exhausted every call errors. + fn scripted(script: Vec>) -> Self { + Self { + script: Arc::new(Mutex::new(script)), + calls: Arc::new(Mutex::new(Vec::new())), + } + } + + fn returning(outcome: SetTargetOutcome) -> Self { + Self::scripted(vec![Some(outcome)]) + } + + fn calls(&self) -> Vec { + self.calls.lock().expect("poisoned").clone() + } + } + + #[async_trait] + impl IndexingRequestRegistry for MockRegistry { + async fn set_indexing_target_candidates( + &self, + requested_by: Address, + deployment_id: DeploymentId, + deployment_chain_id: ChainId, + num_candidates: usize, + ) -> RegistryResult { + self.calls.lock().expect("poisoned").push(( + requested_by, + deployment_id, + deployment_chain_id, + num_candidates, + )); + let mut script = self.script.lock().expect("poisoned"); + match if script.is_empty() { + None + } else { + script.remove(0) + } { + Some(outcome) => Ok(outcome), + None => Err(crate::registry::Error::NoRecordsUpdated), + } + } + + async fn get_all_indexing_requests(&self) -> RegistryResult> { + unimplemented!() + } + + async fn get_indexing_request_by_id( + &self, + _id: &IndexingRequestId, + ) -> RegistryResult> { + unimplemented!() + } + + async fn get_indexing_requests_by_deployment_id( + &self, + _deployment_id: &DeploymentId, + ) -> RegistryResult> { + unimplemented!() + } + + async fn get_open_indexing_requests_for_reassessment( + &self, + _min_age_seconds: i64, + _batch_size: i64, + ) -> RegistryResult> { + unimplemented!() + } + } + + /// A worker queue that records reassessment jobs, optionally failing the + /// first N pushes to exercise the queue-retry path. + #[derive(Clone)] + struct MockWorker { + reassessments: Arc>>, + failures_left: Arc>, + } + + impl MockWorker { + fn new() -> Self { + Self::failing_pushes(0) + } + + fn failing_pushes(failures: usize) -> Self { + Self { + reassessments: Arc::new(Mutex::new(Vec::new())), + failures_left: Arc::new(Mutex::new(failures)), + } + } + + fn reassessments(&self) -> Vec { + self.reassessments.lock().expect("poisoned").clone() + } + } + + #[async_trait] + impl WorkerQueue for MockWorker { + async fn send_indexing_agreement_proposal( + &self, + _candidate_url: Url, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + + async fn reassess_indexing_request( + &self, + indexing_request_id: IndexingRequestId, + deployment_id: DeploymentId, + deployment_chain_id: ChainId, + num_candidates: usize, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + { + let mut failures_left = self.failures_left.lock().expect("poisoned"); + if *failures_left > 0 { + *failures_left -= 1; + anyhow::bail!("scripted queue failure"); + } + } + self.reassessments.lock().expect("poisoned").push(( + indexing_request_id, + deployment_id, + deployment_chain_id, + num_candidates, + )); + Ok(JobId::default()) + } + + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + + async fn submit_offer( + &self, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _indexer_url: Url, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + } + + /// Drive `apply_target_with_retries` for the standard valid propose, with + /// no shutdown pending, returning its result and the captured events. + async fn run_apply( + registry: &MockRegistry, + worker: &MockWorker, + num_candidates: usize, + ) -> (bool, Vec) { + let events_capture = CapturingEventsProducer::new(); + let events: Arc = + Arc::new(events_capture.clone()); + let (_shutdown_tx, mut shutdown_rx) = watch::channel(false); + + let target = ProposedTarget { + event_id: "01912345-6789-7abc-def0-123456789abc".to_string(), + deployment_id: deployment_id!(QM_HASH), + deployment_chain_id: INDEXED_CHAIN_ID, + num_candidates, + }; + + let applied = apply_target_with_retries( + registry, + worker, + &events, + PROTOCOL_CHAIN_ID, + requester(), + TEST_MAX_CANDIDATES, + &target, + 0, + 0, + &mut shutdown_rx, + ) + .await; + + (applied, events_capture.events()) + } + + #[tokio::test] + async fn an_inserted_outcome_applies_emits_and_queues_reassessment() { + let registry = MockRegistry::returning(SetTargetOutcome::Inserted { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::new(); + + let (applied, events) = run_apply(®istry, &worker, 3).await; + + assert!(applied); + assert_eq!( + registry.calls(), + vec![(requester(), deployment_id!(QM_HASH), INDEXED_CHAIN_ID, 3)] + ); + + let reassessments = worker.reassessments(); + assert_eq!(reassessments.len(), 1); + assert_eq!(reassessments[0].1, deployment_id!(QM_HASH)); + assert_eq!(reassessments[0].2, INDEXED_CHAIN_ID); + assert_eq!(reassessments[0].3, 3); + + assert_eq!(events.len(), 1, "expected 1 request-received event"); + match &events[0] { + CapturedEvent::RequestReceived { + deployment, + chain_id, + event, + } => { + assert_eq!(*deployment, deployment_id!(QM_HASH)); + assert_eq!( + *chain_id, PROTOCOL_CHAIN_ID, + "the event carries the protocol chain id, not the indexed chain" + ); + assert_eq!(event.agreements_requested, 3); + } + other => panic!("expected RequestReceived, got {other:?}"), + } + } + + #[tokio::test] + async fn a_noop_outcome_applies_without_events_or_reassessment() { + let registry = MockRegistry::returning(SetTargetOutcome::NoOp { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::new(); + + let (applied, events) = run_apply(®istry, &worker, 3).await; + + assert!(applied); + assert!(worker.reassessments().is_empty(), "no-op must not reassess"); + assert!(events.is_empty(), "no-op must not emit events"); + } + + #[tokio::test] + async fn a_zero_count_cancellation_reassesses_to_zero_without_events() { + let registry = MockRegistry::returning(SetTargetOutcome::Canceled { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::new(); + + let (applied, events) = run_apply(®istry, &worker, 0).await; + + assert!(applied); + assert_eq!(registry.calls()[0].3, 0, "the registry saw the 0 count"); + + let reassessments = worker.reassessments(); + assert_eq!(reassessments.len(), 1); + assert_eq!( + reassessments[0].3, 0, + "reassessment with 0 drives the shrink that cancels agreements" + ); + assert!( + events.is_empty(), + "cancellation must not emit request-received" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_registry_failure_retries_the_whole_apply_until_it_succeeds() { + // 1st call errors (nothing committed), the retry lands the insert. + let registry = MockRegistry::scripted(vec![ + None, + Some(SetTargetOutcome::Inserted { + id: IndexingRequestId::new(), + }), + ]); + let worker = MockWorker::new(); + + let (applied, events) = run_apply(®istry, &worker, 3).await; + + assert!(applied); + assert_eq!(registry.calls().len(), 2, "the apply was retried once"); + assert_eq!(worker.reassessments().len(), 1); + assert_eq!(events.len(), 1); + } + + #[tokio::test(start_paused = true)] + async fn a_queue_failure_retries_only_the_push_never_the_registry() { + // If the retry re-ran the registry call, the outcome would be a no-op + // and the reassessment job would be silently lost. + let registry = MockRegistry::returning(SetTargetOutcome::Inserted { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::failing_pushes(2); + + let (applied, events) = run_apply(®istry, &worker, 3).await; + + assert!(applied); + assert_eq!( + registry.calls().len(), + 1, + "the committed row change must not be re-applied" + ); + assert_eq!( + worker.reassessments().len(), + 1, + "the push eventually landed" + ); + assert_eq!( + events.len(), + 1, + "the lifecycle event is emitted exactly once" + ); + } + + #[tokio::test] + async fn an_oversized_count_is_clamped_to_the_configured_maximum() { + let registry = MockRegistry::returning(SetTargetOutcome::Inserted { + id: IndexingRequestId::new(), + }); + let worker = MockWorker::new(); + + let (applied, _) = run_apply(®istry, &worker, 5_000_000).await; + + assert!(applied); + assert_eq!( + registry.calls()[0].3, + TEST_MAX_CANDIDATES, + "the registry must see the clamped count, not the wire value" + ); + assert_eq!(worker.reassessments()[0].3, TEST_MAX_CANDIDATES); + } + + #[tokio::test] + async fn a_pending_shutdown_stops_retrying_without_applying() { + // Every registry call fails, so only shutdown can end the retry loop. + let registry = MockRegistry::scripted(vec![]); + let worker = MockWorker::new(); + + let events_capture = CapturingEventsProducer::new(); + let events: Arc = + Arc::new(events_capture.clone()); + let (shutdown_tx, mut shutdown_rx) = watch::channel(false); + shutdown_tx.send(true).expect("send shutdown"); + + let target = ProposedTarget { + event_id: "e".to_string(), + deployment_id: deployment_id!(QM_HASH), + deployment_chain_id: INDEXED_CHAIN_ID, + num_candidates: 3, + }; + + let applied = apply_target_with_retries( + ®istry, + &worker, + &events, + PROTOCOL_CHAIN_ID, + requester(), + TEST_MAX_CANDIDATES, + &target, + 0, + 0, + &mut shutdown_rx, + ) + .await; + + assert!(!applied, "shutdown must win over the retry loop"); + assert!(worker.reassessments().is_empty()); + assert!(events_capture.events().is_empty()); + } + + // -------- Redpanda-backed service test -------- + + use dipper_producer::{ + events::SubgraphIndexingAgreementsEventsEmitter, + kafka::{KafkaConfig, KafkaConsumerConfig, KafkaProducer}, + }; + + use crate::registry::RegistryProvider; + + fn redpanda_brokers() -> Option> { + match std::env::var("REDPANDA_BROKERS") { + Ok(value) if !value.trim().is_empty() => Some( + value + .split(',') + .map(|broker| broker.trim().to_string()) + .collect(), + ), + // REQUIRE_REDPANDA turns the silent skip into a failure, so CI + // cannot go green while accidentally testing nothing. + _ if std::env::var("REQUIRE_REDPANDA").is_ok() => { + panic!("REQUIRE_REDPANDA is set but REDPANDA_BROKERS is not") + } + _ => { + eprintln!("skipping Redpanda-backed test: REDPANDA_BROKERS is not set"); + None + } + } + } + + fn unique_topic() -> String { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock before unix epoch") + .as_nanos(); + format!("dipper.test.consumer.{}.{nanos}", std::process::id()) + } + + async fn create_topic(brokers: &[String], topic: &str) { + let client = rskafka::client::ClientBuilder::new(brokers.to_vec()) + .build() + .await + .expect("connect to broker"); + client + .controller_client() + .expect("controller client") + .create_topic(topic, 1, 1, 5_000) + .await + .expect("create topic"); + } + + fn propose_bytes(qm_hash: &str, count: i32) -> Vec { + studio::SubgraphIndexingRequestEvent { + event_id: "01912345-6789-7abc-def0-123456789abc".to_string(), + event_type: EVENT_TYPE_PROPOSE.to_string(), + event_version: "1.0".to_string(), + timestamp: "2026-08-24T10:30:00.123Z".to_string(), + subgraph_deployment_qm_hash: qm_hash.to_string(), + the_graph_network_caip2id: "eip155:42161".to_string(), + payload: Some( + studio::subgraph_indexing_request_event::Payload::SubgraphIndexingRequestPropose( + studio::SubgraphIndexingRequestPropose { + indexing_agreements_requested: count, + indexed_network_caip2id: "eip155:1".to_string(), + }, + ), + ), + } + .encode_to_vec() + } + + async fn wait_until(what: &str, mut check: impl AsyncFnMut() -> bool) { + for _ in 0..300 { + if check().await { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + panic!("timed out waiting for {what}"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn it_consumer_resumes_from_persisted_offsets_across_restarts() { + let Some(brokers) = redpanda_brokers() else { + return; + }; + + let topic = unique_topic(); + create_topic(&brokers, &topic).await; + + let temp_db = pgtemp::PgTempDB::new(); + let db = sqlx::Pool::connect(&temp_db.connection_uri()) + .await + .expect("connect to temp db"); + dipper_pgregistry::run_db_migrations(&db) + .await + .expect("run migrations"); + let provider = RegistryProvider::new(db.clone()); + + let requested_by: Address = "0x8f8c426f956876325b1e037c6eae9b189952994c" + .parse() + .expect("valid address"); + let deployment_a = deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"); + let deployment_b = deployment_id!("QmXbNL4EMkQ6DAPUcBjYSDXZJdpu1Kb1XkKvNvS8JdT7Hs"); + + let producer_config: KafkaConfig = serde_json::from_value(serde_json::json!({ + "brokers": brokers, + "topic": topic, + "partitions": 1, + })) + .expect("valid producer config"); + let producer = KafkaProducer::new(&producer_config) + .await + .expect("producer connects"); + + let consumer_config = IndexingRequestConsumerConfig { + enabled: true, + kafka: KafkaConsumerConfig { + brokers: brokers.clone(), + topic: topic.clone(), + sasl_mechanism: None, + sasl_username: None, + sasl_password: None, + tls_enabled: false, + tls_ca_cert_path: None, + connect_timeout_secs: 60, + }, + requested_by, + max_wait: Duration::from_secs(1), + fetch_max_bytes: 1_048_576, + }; + + let run_service = |worker: MockWorker| { + let events: Arc = + Arc::new(SubgraphIndexingAgreementsEventsEmitter::disabled()); + let (handle, service) = new(Ctx { + registry: provider.clone(), + worker_queue: worker, + events, + protocol_chain_id: 42161, + max_candidates: 10, + config: consumer_config.clone(), + }); + (handle, tokio::spawn(service)) + }; + + // A propose published before the consumer ever ran must still be + // picked up: a fresh partition starts from the earliest record. + producer + .send( + "eip155:42161/QmUzRg.../request", + &propose_bytes("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv", 2), + ) + .await + .expect("produce event A"); + + let worker_run_1 = MockWorker::new(); + let (handle, task) = run_service(worker_run_1.clone()); + wait_until("request A to be registered", async || { + !provider + .get_indexing_requests_by_deployment_id(&deployment_a) + .await + .expect("query requests") + .is_empty() + }) + .await; + wait_until("offset 1 to be persisted", async || { + provider + .get_kafka_consumer_offset(&topic, 0) + .await + .expect("query offset") + == Some(1) + }) + .await; + handle.stop().await; + task.await.expect("service task").expect("service result"); + + assert_eq!( + worker_run_1.reassessments().len(), + 1, + "run 1 applied exactly the 1 produced event" + ); + + // Published while the consumer is down; run 2 must pick it up from the + // persisted offset without re-applying event A. + producer + .send( + "eip155:42161/QmXbNL.../request", + &propose_bytes("QmXbNL4EMkQ6DAPUcBjYSDXZJdpu1Kb1XkKvNvS8JdT7Hs", 3), + ) + .await + .expect("produce event B"); + + let worker_run_2 = MockWorker::new(); + let (handle, task) = run_service(worker_run_2.clone()); + wait_until("request B to be registered", async || { + !provider + .get_indexing_requests_by_deployment_id(&deployment_b) + .await + .expect("query requests") + .is_empty() + }) + .await; + handle.stop().await; + task.await.expect("service task").expect("service result"); + + let run_2_reassessments = worker_run_2.reassessments(); + assert_eq!( + run_2_reassessments.len(), + 1, + "run 2 resumed past event A and applied only event B: {run_2_reassessments:?}" + ); + assert_eq!(run_2_reassessments[0].1, deployment_b); + assert_eq!(run_2_reassessments[0].3, 3); + + assert_eq!( + provider + .get_kafka_consumer_offset(&topic, 0) + .await + .expect("query offset"), + Some(2), + "both records are committed" + ); + assert_eq!( + provider + .get_indexing_requests_by_deployment_id(&deployment_a) + .await + .expect("query requests") + .len(), + 1, + "event A was applied exactly once across both runs" + ); + } +} diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index 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/bin/dipper-service/src/set_indexing_target.rs b/bin/dipper-service/src/set_indexing_target.rs new file mode 100644 index 00000000..839290fc --- /dev/null +++ b/bin/dipper-service/src/set_indexing_target.rs @@ -0,0 +1,177 @@ +//! Applies a set-indexing-target request end to end: the registry upsert, the +//! request-received lifecycle event, and the follow-up reassessment job. Both +//! front doors share it: the admin RPC handler and the Kafka request consumer. + +use std::sync::Arc; + +use dipper_core::ids::IndexingRequestId; +use dipper_producer::{events::SubgraphIndexingAgreementEventsProducer, proto}; +use thegraph_core::{ + DeploymentId, + alloy::primitives::{Address, ChainId}, +}; + +use crate::{ + registry::{IndexingRequestRegistry, SetTargetOutcome}, + worker::service::{JobPriority, WorkerQueue}, +}; + +/// A set-indexing-target request, independent of which front door it came in +/// through. +pub struct SetIndexingTarget { + /// Who asked (recovered RPC signer, or the consumer's configured identity). + pub requested_by: Address, + /// The subgraph deployment to index. + pub deployment_id: DeploymentId, + /// The chain the deployment indexes (its data source), keying the request. + pub deployment_chain_id: ChainId, + /// The target number of indexers; 0 cancels the request. + pub num_candidates: usize, + /// Priority for the follow-up reassessment job. + pub priority: JobPriority, +} + +/// Errors from applying a set-indexing-target request. Both variants are +/// already logged with full context when returned. +#[derive(Debug, thiserror::Error)] +pub enum ApplyError { + /// The registry upsert failed; nothing was changed or queued. + #[error("failed to set indexing target candidates")] + Registry(#[source] crate::registry::Error), + + /// The row change is committed but the reassessment job was not queued. + /// Retry the queue push itself, not the whole apply: a repeated apply + /// lands on the registry's no-op path and never queues the job. + #[error("failed to queue reassessment for indexing request {id}")] + QueueReassess { + id: IndexingRequestId, + #[source] + source: anyhow::Error, + }, +} + +/// Upserts the indexing request, emits the request-received lifecycle event on +/// a genuinely new request, and queues reassessment when the row changed. +/// Returns the request id, or `None` when there was nothing to act on. +pub async fn apply_set_indexing_target( + registry: &R, + worker: &W, + events: &Arc, + the_graph_network: ChainId, + request: SetIndexingTarget, +) -> Result, ApplyError> +where + R: IndexingRequestRegistry + Send + Sync, + W: WorkerQueue + Send + Sync, +{ + let SetIndexingTarget { + requested_by, + deployment_id, + deployment_chain_id, + num_candidates, + priority, + } = request; + + let outcome = match registry + .set_indexing_target_candidates( + requested_by, + deployment_id, + deployment_chain_id, + num_candidates, + ) + .await + { + Ok(outcome) => outcome, + Err(err) => { + tracing::error!(error=?err, "Failed to set indexing target candidates"); + return Err(ApplyError::Registry(err)); + } + }; + + // Translate the outcome into the appropriate follow-up worker job and the + // request id to hand back. + let (id_opt, reassess_count): (Option, Option) = match outcome { + SetTargetOutcome::Inserted { id } => { + tracing::info!( + indexing_request_id = %id, + %requested_by, + %deployment_id, + chain_id = %deployment_chain_id, + num_candidates, + "Inserted new indexing request" + ); + + // Only `Inserted` is a genuinely new request, so only it emits the + // lifecycle event; `the_graph_network` is the protocol network, + // not the deployment's data-source `chain_id`. + events.produce_subgraph_indexing_agreement_request_received( + deployment_id, + the_graph_network, + proto::SubgraphIndexingAgreementRequestReceived { + agreements_requested: num_candidates as i32, + }, + ); + + (Some(id), Some(num_candidates)) + } + SetTargetOutcome::Updated { + id, + new_num_candidates, + } => { + tracing::info!( + indexing_request_id = %id, + %requested_by, + %deployment_id, + chain_id = %deployment_chain_id, + num_candidates = new_num_candidates, + "Updated num_candidates on open indexing request" + ); + (Some(id), Some(new_num_candidates)) + } + SetTargetOutcome::NoOp { id } => { + tracing::debug!( + indexing_request_id = %id, + "Set target candidates is a no-op (count unchanged)" + ); + (Some(id), None) + } + SetTargetOutcome::Canceled { id } => { + tracing::info!( + indexing_request_id = %id, + %requested_by, + %deployment_id, + chain_id = %deployment_chain_id, + "Canceled indexing request (target candidates set to zero)" + ); + (Some(id), Some(0)) + } + SetTargetOutcome::NoOpAlreadyEmpty => { + tracing::warn!( + %requested_by, + %deployment_id, + chain_id = %deployment_chain_id, + "set_indexing_target_candidates with num_candidates=0 against a key with no open request \ + - nothing to cancel" + ); + (None, None) + } + }; + + // Queue reassessment if the row changed: it diffs the IISA target group of + // size `num_candidates` against the active agreements and grows or shrinks + // to match; 0 shrinks to nothing, cancelling every agreement on-chain. + if let (Some(id), Some(count)) = (id_opt, reassess_count) + && let Err(err) = worker + .reassess_indexing_request(id, deployment_id, deployment_chain_id, count, priority) + .await + { + tracing::error!( + indexing_request_id = %id, + error = ?err, + "Failed to queue task: 'reassess_indexing_request'" + ); + return Err(ApplyError::QueueReassess { id, source: err }); + } + + Ok(id_opt) +} diff --git a/dipper-pgregistry/migrations/20260824000000_add_kafka_consumer_offsets.sql b/dipper-pgregistry/migrations/20260824000000_add_kafka_consumer_offsets.sql new file mode 100644 index 00000000..f419f87c --- /dev/null +++ b/dipper-pgregistry/migrations/20260824000000_add_kafka_consumer_offsets.sql @@ -0,0 +1,22 @@ +-- Kafka consumer offsets +-- +-- The dipper consumes subgraph indexing request events that Studio produces on +-- a Redpanda topic. The Kafka client in use (rskafka) has no consumer groups, +-- so the broker cannot store consumer progress; the dipper records it here +-- instead, after each record is processed, and resumes from it on restart. + +CREATE TABLE IF NOT EXISTS dipper_kafka_consumer_offsets ( + -- Topic name; the topic is deploy-time config, so it is part of the key. + topic TEXT NOT NULL, + -- Kafka partition id within the topic + partition_id INT NOT NULL, + -- The next offset to fetch: 1 past the last fully processed record + next_offset BIGINT NOT NULL, + -- Timestamps for auditing + created_at TIMESTAMPTZ NOT NULL DEFAULT timezone('UTC', now()), + updated_at TIMESTAMPTZ NOT NULL DEFAULT timezone('UTC', now()), + PRIMARY KEY (topic, partition_id) +); + +COMMENT ON TABLE dipper_kafka_consumer_offsets IS 'Per-partition consumer progress for Kafka topics the dipper consumes; written after processing so delivery is at-least-once'; +COMMENT ON COLUMN dipper_kafka_consumer_offsets.next_offset IS 'The next offset to fetch: 1 past the last fully processed record'; diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 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) + ); +} 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/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; diff --git a/dipper-producer/tests/it_kafka_redpanda.rs b/dipper-producer/tests/it_kafka_redpanda.rs new file mode 100644 index 00000000..c1c7129c --- /dev/null +++ b/dipper-producer/tests/it_kafka_redpanda.rs @@ -0,0 +1,157 @@ +//! Kafka producer/consumer roundtrip tests against a real Redpanda broker. +//! Gated on `REDPANDA_BROKERS` (e.g. `localhost:9092`): unset, every test +//! skips with a note; CI starts a Redpanda container and always runs them. + +use dipper_producer::kafka::{ + ConsumerError, KafkaConfig, KafkaConsumer, KafkaConsumerConfig, KafkaProducer, OffsetAt, +}; + +fn brokers() -> Option> { + match std::env::var("REDPANDA_BROKERS") { + Ok(value) if !value.trim().is_empty() => Some( + value + .split(',') + .map(|broker| broker.trim().to_string()) + .collect(), + ), + // REQUIRE_REDPANDA turns the silent skip into a failure, so CI cannot + // go green while accidentally testing nothing. + _ if std::env::var("REQUIRE_REDPANDA").is_ok() => { + panic!("REQUIRE_REDPANDA is set but REDPANDA_BROKERS is not") + } + _ => { + eprintln!("skipping Redpanda-backed test: REDPANDA_BROKERS is not set"); + None + } + } +} + +/// A topic name unique to this test process and call site, so parallel tests +/// on a shared broker never read each other's records. +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:?}" + ); +}