chore(all): rename Nozzle to Amp

Author: isum

Cargo.toml

@@ -98,7 +98,7 @@ substreams-entity-change = "2"
 substreams-near-core = "=0.10.2"
 rand = { version = "0.9.2", features = ["os_rng"] }
 
-# Dependencies related to Nozzle Subgraphs
+# Dependencies related to Amp subgraphs
 ahash = "0.8.11"
 alloy = { version = "1.0.12", default-features = false, features = ["json-abi", "serde"] }
 arrow = { version = "=55.0.0" }

chain/ethereum/src/runtime/runtime_adapter.rs

@@ -182,7 +182,7 @@ impl blockchain::RuntimeAdapter<Chain> for RuntimeAdapter {
                 create_host_fns(abis, archive, call_cache, eth_adapters, eth_call_gas)
             }
             data_source::DataSource::Offchain(_) => vec![],
-            data_source::DataSource::Nozzle(_) => vec![],
+            data_source::DataSource::Amp(_) => vec![],
         };
 
         Ok(host_fns)

core/src/nozzle_subgraph/manager.rs core/src/amp_subgraph/manager.rscore/src/nozzle_subgraph/manager.rs renamed to core/src/amp_subgraph/manager.rs

@@ -4,6 +4,7 @@ use alloy::primitives::BlockNumber;
 use anyhow::Context;
 use async_trait::async_trait;
 use graph::{
+    amp,
     components::{
         link_resolver::{LinkResolver, LinkResolverContext},
         metrics::MetricsRegistry,
@@ -12,44 +13,43 @@ use graph::{
     },
     env::EnvVars,
     log::factory::LoggerFactory,
-    nozzle,
     prelude::CheapClone,
 };
 use slog::{debug, error};
 use tokio_util::sync::CancellationToken;
 
 use super::{runner, Metrics, Monitor};
 
-/// Manages Nozzle subgraph runner futures.
+/// Manages Amp subgraph runner futures.
 ///
-/// Creates and schedules Nozzle subgraph runner futures for execution on demand.
-/// Also handles stopping previously started Nozzle subgraph runners.
+/// Creates and schedules Amp subgraph runner futures for execution on demand.
+/// Also handles stopping previously started Amp subgraph runners.
 pub struct Manager<SS, NC> {
     logger_factory: LoggerFactory,
     metrics_registry: Arc<MetricsRegistry>,
     env_vars: Arc<EnvVars>,
     monitor: Monitor,
     subgraph_store: Arc<SS>,
     link_resolver: Arc<dyn LinkResolver>,
-    nozzle_client: Arc<NC>,
+    amp_client: Arc<NC>,
 }
 
 impl<SS, NC> Manager<SS, NC>
 where
     SS: SubgraphStore,
-    NC: nozzle::Client,
+    NC: amp::Client,
 {
-    /// Creates a new Nozzle subgraph manager.
+    /// Creates a new Amp subgraph manager.
     pub fn new(
         logger_factory: &LoggerFactory,
         metrics_registry: Arc<MetricsRegistry>,
         env_vars: Arc<EnvVars>,
         cancel_token: &CancellationToken,
         subgraph_store: Arc<SS>,
         link_resolver: Arc<dyn LinkResolver>,
-        nozzle_client: Arc<NC>,
+        amp_client: Arc<NC>,
     ) -> Self {
-        let logger = logger_factory.component_logger("NozzleSubgraphManager", None);
+        let logger = logger_factory.component_logger("AmpSubgraphManager", None);
         let logger_factory = logger_factory.with_parent(logger);
 
         let monitor = Monitor::new(&logger_factory, cancel_token);
@@ -61,7 +61,7 @@ where
             monitor,
             subgraph_store,
             link_resolver,
-            nozzle_client,
+            amp_client,
         }
     }
 }
@@ -70,7 +70,7 @@ where
 impl<SS, NC> SubgraphInstanceManager for Manager<SS, NC>
 where
     SS: SubgraphStore,
-    NC: nozzle::Client + Send + Sync + 'static,
+    NC: amp::Client + Send + Sync + 'static,
 {
     async fn start_subgraph(
         self: Arc<Self>,
@@ -115,10 +115,10 @@ where
                     let raw_manifest = serde_yaml::from_slice(&manifest_bytes)
                         .context("failed to parse subgraph manifest")?;
 
-                    let mut manifest = nozzle::Manifest::resolve::<graph_chain_ethereum::Chain, _>(
+                    let mut manifest = amp::Manifest::resolve::<graph_chain_ethereum::Chain, _>(
                         &logger,
                         manager.link_resolver.cheap_clone(),
-                        manager.nozzle_client.cheap_clone(),
+                        manager.amp_client.cheap_clone(),
                         manager.env_vars.max_spec_version.cheap_clone(),
                         deployment.hash.cheap_clone(),
                         raw_manifest,
@@ -138,8 +138,8 @@ where
 
                     let runner_context = runner::Context::new(
                         &logger,
-                        &manager.env_vars.nozzle,
-                        manager.nozzle_client.cheap_clone(),
+                        &manager.env_vars.amp,
+                        manager.amp_client.cheap_clone(),
                         store,
                         deployment.hash.cheap_clone(),
                         manifest,

core/src/nozzle_subgraph/metrics.rs core/src/amp_subgraph/metrics.rscore/src/nozzle_subgraph/metrics.rs renamed to core/src/amp_subgraph/metrics.rs

@@ -26,7 +26,7 @@ impl Metrics {
         let stopwatch = StopwatchMetrics::new(
             logger.cheap_clone(),
             deployment,
-            "nozzle-process",
+            "amp-process",
             metrics_registry,
             store.shard().to_string(),
         );

core/src/nozzle_subgraph/mod.rs core/src/amp_subgraph/mod.rscore/src/nozzle_subgraph/mod.rs renamed to core/src/amp_subgraph/mod.rs

@@ -92,7 +92,7 @@ impl Monitor {
     /// A new cancel token is derived from the `cancel_token` and only the derived token is used by the
     /// subgraph monitor and its background process.
     pub(super) fn new(logger_factory: &LoggerFactory, cancel_token: &CancellationToken) -> Self {
-        let logger = logger_factory.component_logger("NozzleSubgraphMonitor", None);
+        let logger = logger_factory.component_logger("AmpSubgraphMonitor", None);
         let logger_factory = Arc::new(logger_factory.with_parent(logger));
 
         // A derived token makes sure it is not possible to accidentally cancel the parent token
@@ -221,7 +221,7 @@ impl Monitor {
                 _ = cancel_token.cancelled() => {
                     debug!(logger, "Stopping command processor");
 
-                    // All active Subgraphs will shutdown gracefully
+                    // All active subgraphs will shutdown gracefully
                     // because their cancel tokens are derived from this cancelled token.
                     return;
                 }

core/src/nozzle_subgraph/monitor.rs core/src/amp_subgraph/monitor.rscore/src/nozzle_subgraph/monitor.rs renamed to core/src/amp_subgraph/monitor.rs

@@ -2,21 +2,21 @@ use std::sync::Arc;
 
 use alloy::primitives::{BlockHash, BlockNumber};
 use graph::{
+    amp::{log::Logger as _, Codec, Manifest},
     cheap_clone::CheapClone,
     components::store::WritableStore,
     data::subgraph::DeploymentHash,
-    env::NozzleEnv,
-    nozzle::{log::Logger as _, Codec, Manifest},
+    env::AmpEnv,
     util::backoff::ExponentialBackoff,
 };
 use slog::Logger;
 
 use super::Compat;
-use crate::nozzle_subgraph::Metrics;
+use crate::amp_subgraph::Metrics;
 
-pub(in super::super) struct Context<NC> {
+pub(in super::super) struct Context<AC> {
     pub(super) logger: Logger,
-    pub(super) client: Arc<NC>,
+    pub(super) client: Arc<AC>,
     pub(super) store: Arc<dyn WritableStore>,
     pub(super) max_buffer_size: usize,
     pub(super) max_block_range: usize,
@@ -27,17 +27,17 @@ pub(in super::super) struct Context<NC> {
     pub(super) codec: Codec,
 }
 
-impl<NC> Context<NC> {
+impl<AC> Context<AC> {
     pub(in super::super) fn new(
         logger: &Logger,
-        env: &NozzleEnv,
-        client: Arc<NC>,
+        env: &AmpEnv,
+        client: Arc<AC>,
         store: Arc<dyn WritableStore>,
         deployment: DeploymentHash,
         manifest: Manifest,
         metrics: Metrics,
     ) -> Self {
-        let logger = logger.component("NozzleSubgraphRunner");
+        let logger = logger.component("AmpSubgraphRunner");
         let backoff = ExponentialBackoff::new(env.query_retry_min_delay, env.query_retry_max_delay);
         let codec = Codec::new(manifest.schema.cheap_clone());
 

core/src/nozzle_subgraph/runner/compat.rs core/src/amp_subgraph/runner/compat.rscore/src/nozzle_subgraph/runner/compat.rs renamed to core/src/amp_subgraph/runner/compat.rs

@@ -5,20 +5,20 @@ use anyhow::anyhow;
 use arrow::array::RecordBatch;
 use chrono::{DateTime, Utc};
 use graph::{
-    blockchain::block_stream::FirehoseCursor,
-    cheap_clone::CheapClone,
-    components::store::{EntityCache, ModificationsAndCache},
-    nozzle::{
+    amp::{
         codec::{utils::auto_block_timestamp_decoder, DecodeOutput, DecodedEntity, Decoder},
         stream_aggregator::{RecordBatchGroup, RecordBatchGroups, StreamRecordBatch},
     },
+    blockchain::block_stream::FirehoseCursor,
+    cheap_clone::CheapClone,
+    components::store::{EntityCache, ModificationsAndCache},
 };
 use slog::{debug, trace};
 
 use super::{data_stream::TablePtr, Compat, Context, Error};
 
-pub(super) async fn process_record_batch_groups<NC>(
-    cx: &mut Context<NC>,
+pub(super) async fn process_record_batch_groups<AC>(
+    cx: &mut Context<AC>,
     mut entity_cache: EntityCache,
     record_batch_groups: RecordBatchGroups,
     stream_table_ptr: Arc<[TablePtr]>,
@@ -72,8 +72,8 @@ pub(super) async fn process_record_batch_groups<NC>(
     Ok(entity_cache)
 }
 
-async fn process_record_batch_group<NC>(
-    cx: &mut Context<NC>,
+async fn process_record_batch_group<AC>(
+    cx: &mut Context<AC>,
     mut entity_cache: EntityCache,
     block_number: BlockNumber,
     block_hash: BlockHash,
@@ -146,8 +146,8 @@ async fn process_record_batch_group<NC>(
     ))
 }
 
-async fn process_record_batch<NC>(
-    cx: &mut Context<NC>,
+async fn process_record_batch<AC>(
+    cx: &mut Context<AC>,
     entity_cache: &mut EntityCache,
     block_number: BlockNumber,
     record_batch: RecordBatch,

…re/src/nozzle_subgraph/runner/context.rs core/src/amp_subgraph/runner/context.rscore/src/nozzle_subgraph/runner/context.rs renamed to core/src/amp_subgraph/runner/context.rs core/src/nozzle_subgraph/runner/context.rs renamed to core/src/amp_subgraph/runner/context.rs

@@ -7,25 +7,25 @@ use futures::{
     StreamExt, TryStreamExt,
 };
 use graph::{
-    cheap_clone::CheapClone,
-    nozzle::{
+    amp::{
         manifest::DataSource,
         stream_aggregator::{RecordBatchGroups, StreamAggregator},
         Client,
     },
+    cheap_clone::CheapClone,
 };
 use slog::{debug, warn};
 
 use super::{Context, Error};
 
 pub(super) type TablePtr = (usize, usize);
 
-pub(super) fn new_data_stream<NC>(
-    cx: &Context<NC>,
+pub(super) fn new_data_stream<AC>(
+    cx: &Context<AC>,
     latest_block: BlockNumber,
 ) -> BoxStream<'static, Result<(RecordBatchGroups, Arc<[TablePtr]>), Error>>
 where
-    NC: Client,
+    AC: Client,
 {
     let logger = cx.logger.new(slog::o!("process" => "new_data_stream"));
 
@@ -118,8 +118,8 @@ where
     merged_data_stream
 }
 
-fn next_block_ranges<NC>(
-    cx: &Context<NC>,
+fn next_block_ranges<AC>(
+    cx: &Context<AC>,
     latest_queried_block: Option<BlockNumber>,
     latest_block: BlockNumber,
 ) -> HashMap<usize, RangeInclusive<BlockNumber>> {
@@ -148,8 +148,8 @@ fn next_block_ranges<NC>(
         .collect()
 }
 
-fn next_block_range<NC>(
-    cx: &Context<NC>,
+fn next_block_range<AC>(
+    cx: &Context<AC>,
     data_source: &DataSource,
     latest_queried_block: Option<BlockNumber>,
     latest_block: BlockNumber,