From ca5f5f8e847e5247ea4131740a6fa977ce687aaf Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Wed, 30 Apr 2025 22:41:31 -0400
Subject: [PATCH 01/12] fix(amqp sink): attempt one reconnect when channel has
 errored

---
 src/sinks/amqp/channel.rs           | 67 +++++++++++++++++++++++++++++
 src/sinks/amqp/config.rs            |  5 ++-
 src/sinks/amqp/integration_tests.rs |  3 +-
 src/sinks/amqp/mod.rs               |  1 +
 src/sinks/amqp/service.rs           | 11 ++++-
 src/sinks/amqp/sink.rs              | 21 +++------
 6 files changed, 91 insertions(+), 17 deletions(-)
 create mode 100644 src/sinks/amqp/channel.rs

diff --git a/src/sinks/amqp/channel.rs b/src/sinks/amqp/channel.rs
new file mode 100644
index 0000000000000..4ac1ecee72083
--- /dev/null
+++ b/src/sinks/amqp/channel.rs
@@ -0,0 +1,67 @@
+use crate::amqp::AmqpConfig;
+use lapin::options::ConfirmSelectOptions;
+use tokio::sync::{RwLock, RwLockReadGuard};
+
+use super::service::AmqpError;
+
+/// A wrapper around the AMQP channel that handles reconnections.
+pub(crate) struct AmqpChannel {
+    channel: RwLock<lapin::Channel>,
+    config: AmqpConfig,
+}
+
+impl AmqpChannel {
+    /// Creates a new AMQP channel.
+    pub async fn new(config: &AmqpConfig) -> Result<Self, AmqpError> {
+        let channel = Self::new_channel(config).await?;
+
+        Ok(Self {
+            channel: RwLock::new(channel),
+            config: config.clone(),
+        })
+    }
+
+    /// Returns a read lock to the AMQP channel. If the current channel is in an error state,
+    /// it will attempt to reconnect and create a new channel.
+    pub async fn channel(&self) -> Result<RwLockReadGuard<'_, lapin::Channel>, AmqpError> {
+        let need_reconnect =
+            { self.channel.read().await.status().state() == lapin::ChannelState::Error };
+
+        if need_reconnect {
+            let mut channel = self.channel.write().await;
+
+            // Check if we still need to reconnect after acquiring the write lock.
+            if channel.status().state() != lapin::ChannelState::Error {
+                return Ok(channel.downgrade());
+            }
+
+            info!(
+                message = "Recovering broken connection to the AMQP broker.",
+                internal_log_rate_limit = true,
+            );
+
+            *channel = Self::new_channel(&self.config).await?;
+
+            info!(
+                message = "Recovered connection to the AMQP broker.",
+                internal_log_rate_limit = true,
+            );
+        }
+        Ok(self.channel.read().await)
+    }
+
+    async fn new_channel(config: &AmqpConfig) -> Result<lapin::Channel, AmqpError> {
+        let (_, channel) = config
+            .connect()
+            .await
+            .map_err(|e| AmqpError::ConnectFailed { error: e })?;
+
+        // Enable confirmations on the channel.
+        channel
+            .confirm_select(ConfirmSelectOptions::default())
+            .await
+            .map_err(|e| AmqpError::ConnectFailed { error: Box::new(e) })?;
+
+        Ok(channel)
+    }
+}
diff --git a/src/sinks/amqp/config.rs b/src/sinks/amqp/config.rs
index 50ea63104f836..78bccbda78feb 100644
--- a/src/sinks/amqp/config.rs
+++ b/src/sinks/amqp/config.rs
@@ -1,4 +1,5 @@
 //! Configuration functionality for the `AMQP` sink.
+use super::channel::AmqpChannel;
 use crate::{amqp::AmqpConfig, sinks::prelude::*};
 use lapin::{types::ShortString, BasicProperties};
 use std::sync::Arc;
@@ -135,9 +136,11 @@ impl SinkConfig for AmqpSinkConfig {
     }
 }
 
-pub(super) async fn healthcheck(channel: Arc<lapin::Channel>) -> crate::Result<()> {
+pub(super) async fn healthcheck(channel: Arc<AmqpChannel>) -> crate::Result<()> {
     trace!("Healthcheck started.");
 
+    let channel = channel.channel().await?;
+
     if !channel.status().connected() {
         return Err(Box::new(std::io::Error::new(
             std::io::ErrorKind::BrokenPipe,
diff --git a/src/sinks/amqp/integration_tests.rs b/src/sinks/amqp/integration_tests.rs
index e919c5b88e92e..1e877ab0a3547 100644
--- a/src/sinks/amqp/integration_tests.rs
+++ b/src/sinks/amqp/integration_tests.rs
@@ -3,6 +3,7 @@ use crate::{
     amqp::await_connection,
     config::{SinkConfig, SinkContext},
     shutdown::ShutdownSignal,
+    sinks::amqp::channel::AmqpChannel,
     template::{Template, UnsignedIntTemplate},
     test_util::{
         components::{run_and_assert_sink_compliance, SINK_TAGS},
@@ -37,7 +38,7 @@ async fn healthcheck() {
     let mut config = make_config();
     config.exchange = Template::try_from(exchange.as_str()).unwrap();
     await_connection(&config.connection).await;
-    let (_conn, channel) = config.connection.connect().await.unwrap();
+    let channel = AmqpChannel::new(&config.connection).await.unwrap();
     super::config::healthcheck(Arc::new(channel)).await.unwrap();
 }
 
diff --git a/src/sinks/amqp/mod.rs b/src/sinks/amqp/mod.rs
index 749f892f1ccd2..ef53974c44c4f 100644
--- a/src/sinks/amqp/mod.rs
+++ b/src/sinks/amqp/mod.rs
@@ -1,5 +1,6 @@
 //! `AMQP` sink.
 //! Handles version AMQP 0.9.1 which is used by RabbitMQ.
+mod channel;
 mod config;
 mod encoder;
 mod request_builder;
diff --git a/src/sinks/amqp/service.rs b/src/sinks/amqp/service.rs
index 44c475f208cae..1e0b956a99913 100644
--- a/src/sinks/amqp/service.rs
+++ b/src/sinks/amqp/service.rs
@@ -10,6 +10,8 @@ use std::{
     task::{Context, Poll},
 };
 
+use super::channel::AmqpChannel;
+
 /// The request contains the data to send to `AMQP` together
 /// with the information need to route the message.
 pub(super) struct AmqpRequest {
@@ -79,7 +81,7 @@ impl DriverResponse for AmqpResponse {
 
 /// The tower service that handles the actual sending of data to `AMQP`.
 pub(super) struct AmqpService {
-    pub(super) channel: Arc<lapin::Channel>,
+    pub(super) channel: Arc<AmqpChannel>,
 }
 
 #[derive(Debug, Snafu)]
@@ -92,6 +94,11 @@ pub(super) enum AmqpError {
 
     #[snafu(display("Received Negative Acknowledgement from AMQP broker."))]
     Nack,
+
+    #[snafu(display("Failed to open AMQP channel: {}", error))]
+    ConnectFailed {
+        error: Box<dyn std::error::Error + Send + Sync>,
+    },
 }
 
 impl Service<AmqpRequest> for AmqpService {
@@ -109,6 +116,8 @@ impl Service<AmqpRequest> for AmqpService {
         let channel = Arc::clone(&self.channel);
 
         Box::pin(async move {
+            let channel = channel.channel().await?;
+
             let byte_size = req.body.len();
             let fut = channel
                 .basic_publish(
diff --git a/src/sinks/amqp/sink.rs b/src/sinks/amqp/sink.rs
index 1537713be206c..51528dc4489bf 100644
--- a/src/sinks/amqp/sink.rs
+++ b/src/sinks/amqp/sink.rs
@@ -1,10 +1,11 @@
 //! The sink for the `AMQP` sink that wires together the main stream that takes the
 //! event and sends it to `AMQP`.
 use crate::sinks::prelude::*;
-use lapin::{options::ConfirmSelectOptions, BasicProperties};
+use lapin::BasicProperties;
 use serde::Serialize;
 use std::sync::Arc;
 
+use super::channel::AmqpChannel;
 use super::{
     config::{AmqpPropertiesConfig, AmqpSinkConfig},
     encoder::AmqpEncoder,
@@ -27,7 +28,7 @@ pub(super) struct AmqpEvent {
 }
 
 pub(super) struct AmqpSink {
-    pub(super) channel: Arc<lapin::Channel>,
+    pub(super) channel: Arc<AmqpChannel>,
     exchange: Template,
     routing_key: Option<Template>,
     properties: Option<AmqpPropertiesConfig>,
@@ -37,19 +38,11 @@ pub(super) struct AmqpSink {
 
 impl AmqpSink {
     pub(super) async fn new(config: AmqpSinkConfig) -> crate::Result<Self> {
-        let (_, channel) = config
-            .connection
-            .connect()
-            .await
-            .map_err(|e| BuildError::AmqpCreateFailed { source: e })?;
-
-        // Enable confirmations on the channel.
-        channel
-            .confirm_select(ConfirmSelectOptions::default())
-            .await
-            .map_err(|e| BuildError::AmqpCreateFailed {
+        let channel = AmqpChannel::new(&config.connection).await.map_err(|e| {
+            BuildError::AmqpCreateFailed {
                 source: Box::new(e),
-            })?;
+            }
+        })?;
 
         let transformer = config.encoding.transformer();
         let serializer = config.encoding.build()?;

From 1295c1f293d8f99b8c6051b12e1e4e58440b3a4c Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Wed, 30 Apr 2025 22:57:35 -0400
Subject: [PATCH 02/12] add changelog.d

---
 changelog.d/22971_reconnect_broken_amqp_sink.md | 3 +++
 1 file changed, 3 insertions(+)
 create mode 100644 changelog.d/22971_reconnect_broken_amqp_sink.md

diff --git a/changelog.d/22971_reconnect_broken_amqp_sink.md b/changelog.d/22971_reconnect_broken_amqp_sink.md
new file mode 100644
index 0000000000000..31460649084ed
--- /dev/null
+++ b/changelog.d/22971_reconnect_broken_amqp_sink.md
@@ -0,0 +1,3 @@
+The `amqp` sink now attempts to re-connect to the AMQP broker when the channel has been disconnected.
+
+authors: aramperes

From a199e35bb4c12426803e427ca271fb81410c895c Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Thu, 1 May 2025 18:13:03 -0400
Subject: [PATCH 03/12] fix changelog.d

---
 ...roken_amqp_sink.md => 22971_reconnect_broken_amqp_sink.fix.md} | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename changelog.d/{22971_reconnect_broken_amqp_sink.md => 22971_reconnect_broken_amqp_sink.fix.md} (100%)

diff --git a/changelog.d/22971_reconnect_broken_amqp_sink.md b/changelog.d/22971_reconnect_broken_amqp_sink.fix.md
similarity index 100%
rename from changelog.d/22971_reconnect_broken_amqp_sink.md
rename to changelog.d/22971_reconnect_broken_amqp_sink.fix.md

From a0cde40e31b4a10fcb5db6170bba3cc05407fb63 Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Sat, 10 May 2025 13:13:12 -0400
Subject: [PATCH 04/12] Use vector_common::Error

---
 src/sinks/amqp/mod.rs     | 4 +---
 src/sinks/amqp/service.rs | 4 +---
 2 files changed, 2 insertions(+), 6 deletions(-)

diff --git a/src/sinks/amqp/mod.rs b/src/sinks/amqp/mod.rs
index ef53974c44c4f..6b478c4adc10d 100644
--- a/src/sinks/amqp/mod.rs
+++ b/src/sinks/amqp/mod.rs
@@ -16,7 +16,5 @@ use snafu::Snafu;
 #[derive(Debug, Snafu)]
 enum BuildError {
     #[snafu(display("creating amqp producer failed: {}", source))]
-    AmqpCreateFailed {
-        source: Box<dyn std::error::Error + Send + Sync>,
-    },
+    AmqpCreateFailed { source: vector_common::Error },
 }
diff --git a/src/sinks/amqp/service.rs b/src/sinks/amqp/service.rs
index 1e0b956a99913..f823f8df8a550 100644
--- a/src/sinks/amqp/service.rs
+++ b/src/sinks/amqp/service.rs
@@ -96,9 +96,7 @@ pub(super) enum AmqpError {
     Nack,
 
     #[snafu(display("Failed to open AMQP channel: {}", error))]
-    ConnectFailed {
-        error: Box<dyn std::error::Error + Send + Sync>,
-    },
+    ConnectFailed { error: vector_common::Error },
 }
 
 impl Service<AmqpRequest> for AmqpService {

From cdd660d8a3cfdf2672a941a7b4f47adccadde4ae Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Wed, 21 May 2025 22:37:17 -0400
Subject: [PATCH 05/12] use deadpool to manage channel recovery and multiple
 channels

---
 Cargo.lock                | 17 +++++++++-
 Cargo.toml                |  3 +-
 src/sinks/amqp/channel.rs | 65 +++++++++++++++++----------------------
 src/sinks/amqp/config.rs  |  9 +++---
 src/sinks/amqp/service.rs | 21 ++++++++-----
 src/sinks/amqp/sink.rs    | 18 ++++++-----
 6 files changed, 74 insertions(+), 59 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 3f09488ccebad..a246227aa7fd2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2975,11 +2975,25 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "deadpool"
+version = "0.12.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5ed5957ff93768adf7a65ab167a17835c3d2c3c50d084fe305174c112f468e2f"
+dependencies = [
+ "deadpool-runtime",
+ "num_cpus",
+ "tokio",
+]
+
 [[package]]
 name = "deadpool-runtime"
 version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "63dfa964fe2a66f3fde91fc70b267fe193d822c7e603e2a675a49a7f46ad3f49"
+dependencies = [
+ "tokio",
+]
 
 [[package]]
 name = "der"
@@ -11475,6 +11489,7 @@ dependencies = [
  "crossterm 0.29.0",
  "csv",
  "databend-client",
+ "deadpool 0.12.2",
  "derivative",
  "dirs-next",
  "dnsmsg-parser",
@@ -12771,7 +12786,7 @@ dependencies = [
  "assert-json-diff",
  "async-trait",
  "base64 0.22.1",
- "deadpool",
+ "deadpool 0.10.0",
  "futures 0.3.31",
  "http 1.1.0",
  "http-body-util",
diff --git a/Cargo.toml b/Cargo.toml
index 370158465977f..b07ddd4ae0c79 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -290,6 +290,7 @@ smpl_jwt = { version = "0.8.0", default-features = false, optional = true }
 
 # AMQP
 lapin = { version = "2.5.3", default-features = false, features = ["native-tls"], optional = true }
+deadpool = { version = "0.12.2", default-features = false, features = ["managed", "rt_tokio_1"], optional = true }
 
 # API
 async-graphql = { version = "7.0.17", default-features = false, optional = true, features = ["chrono", "playground"] }
@@ -784,7 +785,7 @@ sinks-metrics = [
   "sinks-splunk_hec"
 ]
 
-sinks-amqp = ["lapin"]
+sinks-amqp = ["deadpool", "lapin"]
 sinks-appsignal = []
 sinks-aws_cloudwatch_logs = ["aws-core", "dep:aws-sdk-cloudwatchlogs", "dep:aws-sdk-kms"]
 sinks-aws_cloudwatch_metrics = ["aws-core", "dep:aws-sdk-cloudwatch"]
diff --git a/src/sinks/amqp/channel.rs b/src/sinks/amqp/channel.rs
index 4ac1ecee72083..439b7cc4ff9c2 100644
--- a/src/sinks/amqp/channel.rs
+++ b/src/sinks/amqp/channel.rs
@@ -1,53 +1,44 @@
+use super::service::AmqpError;
 use crate::amqp::AmqpConfig;
 use lapin::options::ConfirmSelectOptions;
-use tokio::sync::{RwLock, RwLockReadGuard};
-
-use super::service::AmqpError;
 
 /// A wrapper around the AMQP channel that handles reconnections.
 pub(crate) struct AmqpChannel {
-    channel: RwLock<lapin::Channel>,
     config: AmqpConfig,
 }
 
-impl AmqpChannel {
-    /// Creates a new AMQP channel.
-    pub async fn new(config: &AmqpConfig) -> Result<Self, AmqpError> {
-        let channel = Self::new_channel(config).await?;
+impl deadpool::managed::Manager for AmqpChannel {
+    type Type = lapin::Channel;
+    type Error = AmqpError;
 
-        Ok(Self {
-            channel: RwLock::new(channel),
-            config: config.clone(),
-        })
+    async fn create(&self) -> Result<Self::Type, Self::Error> {
+        let channel = Self::new_channel(&self.config).await?;
+        info!(
+            message = "Created a new channel to the AMQP broker.",
+            id = channel.id()
+        );
+        Ok(channel)
     }
 
-    /// Returns a read lock to the AMQP channel. If the current channel is in an error state,
-    /// it will attempt to reconnect and create a new channel.
-    pub async fn channel(&self) -> Result<RwLockReadGuard<'_, lapin::Channel>, AmqpError> {
-        let need_reconnect =
-            { self.channel.read().await.status().state() == lapin::ChannelState::Error };
-
-        if need_reconnect {
-            let mut channel = self.channel.write().await;
-
-            // Check if we still need to reconnect after acquiring the write lock.
-            if channel.status().state() != lapin::ChannelState::Error {
-                return Ok(channel.downgrade());
-            }
-
-            info!(
-                message = "Recovering broken connection to the AMQP broker.",
-                internal_log_rate_limit = true,
-            );
-
-            *channel = Self::new_channel(&self.config).await?;
+    async fn recycle(
+        &self,
+        channel: &mut Self::Type,
+        _: &deadpool::managed::Metrics,
+    ) -> deadpool::managed::RecycleResult<Self::Error> {
+        if channel.status().state() == lapin::ChannelState::Connected {
+            Ok(())
+        } else {
+            Err((AmqpError::ChannelClosed {}).into())
+        }
+    }
+}
 
-            info!(
-                message = "Recovered connection to the AMQP broker.",
-                internal_log_rate_limit = true,
-            );
+impl AmqpChannel {
+    /// Creates a new AMQP channel.
+    pub fn new(config: &AmqpConfig) -> Self {
+        Self {
+            config: config.clone(),
         }
-        Ok(self.channel.read().await)
     }
 
     async fn new_channel(config: &AmqpConfig) -> Result<lapin::Channel, AmqpError> {
diff --git a/src/sinks/amqp/config.rs b/src/sinks/amqp/config.rs
index 78bccbda78feb..ac86e87be8b1f 100644
--- a/src/sinks/amqp/config.rs
+++ b/src/sinks/amqp/config.rs
@@ -2,7 +2,6 @@
 use super::channel::AmqpChannel;
 use crate::{amqp::AmqpConfig, sinks::prelude::*};
 use lapin::{types::ShortString, BasicProperties};
-use std::sync::Arc;
 use vector_lib::{
     codecs::TextSerializerConfig,
     internal_event::{error_stage, error_type},
@@ -123,7 +122,7 @@ impl GenerateConfig for AmqpSinkConfig {
 impl SinkConfig for AmqpSinkConfig {
     async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let sink = AmqpSink::new(self.clone()).await?;
-        let hc = healthcheck(Arc::clone(&sink.channel)).boxed();
+        let hc = healthcheck(sink.channel_pool.clone()).boxed();
         Ok((VectorSink::from_event_streamsink(sink), hc))
     }
 
@@ -136,10 +135,12 @@ impl SinkConfig for AmqpSinkConfig {
     }
 }
 
-pub(super) async fn healthcheck(channel: Arc<AmqpChannel>) -> crate::Result<()> {
+pub(super) async fn healthcheck(
+    channel: deadpool::managed::Pool<AmqpChannel>,
+) -> crate::Result<()> {
     trace!("Healthcheck started.");
 
-    let channel = channel.channel().await?;
+    let channel = channel.get().await?;
 
     if !channel.status().connected() {
         return Err(Box::new(std::io::Error::new(
diff --git a/src/sinks/amqp/service.rs b/src/sinks/amqp/service.rs
index f823f8df8a550..a2545ac9654dd 100644
--- a/src/sinks/amqp/service.rs
+++ b/src/sinks/amqp/service.rs
@@ -5,10 +5,7 @@ use bytes::Bytes;
 use futures::future::BoxFuture;
 use lapin::{options::BasicPublishOptions, BasicProperties};
 use snafu::Snafu;
-use std::{
-    sync::Arc,
-    task::{Context, Poll},
-};
+use std::task::{Context, Poll};
 
 use super::channel::AmqpChannel;
 
@@ -81,11 +78,11 @@ impl DriverResponse for AmqpResponse {
 
 /// The tower service that handles the actual sending of data to `AMQP`.
 pub(super) struct AmqpService {
-    pub(super) channel: Arc<AmqpChannel>,
+    pub(super) channel_pool: deadpool::managed::Pool<AmqpChannel>,
 }
 
 #[derive(Debug, Snafu)]
-pub(super) enum AmqpError {
+pub enum AmqpError {
     #[snafu(display("Failed retrieving Acknowledgement: {}", error))]
     AcknowledgementFailed { error: lapin::Error },
 
@@ -97,6 +94,12 @@ pub(super) enum AmqpError {
 
     #[snafu(display("Failed to open AMQP channel: {}", error))]
     ConnectFailed { error: vector_common::Error },
+
+    #[snafu(display("Channel is closed."))]
+    ChannelClosed,
+
+    #[snafu(display("Channel pool error: {}", error))]
+    PoolError { error: vector_common::Error },
 }
 
 impl Service<AmqpRequest> for AmqpService {
@@ -111,10 +114,12 @@ impl Service<AmqpRequest> for AmqpService {
     }
 
     fn call(&mut self, req: AmqpRequest) -> Self::Future {
-        let channel = Arc::clone(&self.channel);
+        let channel = self.channel_pool.clone();
 
         Box::pin(async move {
-            let channel = channel.channel().await?;
+            let channel = channel.get().await.map_err(|error| AmqpError::PoolError {
+                error: Box::new(error),
+            })?;
 
             let byte_size = req.body.len();
             let fut = channel
diff --git a/src/sinks/amqp/sink.rs b/src/sinks/amqp/sink.rs
index 51528dc4489bf..6e30e0e4a8a58 100644
--- a/src/sinks/amqp/sink.rs
+++ b/src/sinks/amqp/sink.rs
@@ -3,7 +3,6 @@
 use crate::sinks::prelude::*;
 use lapin::BasicProperties;
 use serde::Serialize;
-use std::sync::Arc;
 
 use super::channel::AmqpChannel;
 use super::{
@@ -28,7 +27,7 @@ pub(super) struct AmqpEvent {
 }
 
 pub(super) struct AmqpSink {
-    pub(super) channel: Arc<AmqpChannel>,
+    pub(super) channel_pool: deadpool::managed::Pool<AmqpChannel>,
     exchange: Template,
     routing_key: Option<Template>,
     properties: Option<AmqpPropertiesConfig>,
@@ -38,18 +37,21 @@ pub(super) struct AmqpSink {
 
 impl AmqpSink {
     pub(super) async fn new(config: AmqpSinkConfig) -> crate::Result<Self> {
-        let channel = AmqpChannel::new(&config.connection).await.map_err(|e| {
-            BuildError::AmqpCreateFailed {
+        let channel_manager = AmqpChannel::new(&config.connection);
+        let channel_pool = deadpool::managed::Pool::builder(channel_manager)
+            .max_size(4)
+            .runtime(deadpool::Runtime::Tokio1)
+            .build()
+            .map_err(|e| BuildError::AmqpCreateFailed {
                 source: Box::new(e),
-            }
-        })?;
+            })?;
 
         let transformer = config.encoding.transformer();
         let serializer = config.encoding.build()?;
         let encoder = crate::codecs::Encoder::<()>::new(serializer);
 
         Ok(AmqpSink {
-            channel: Arc::new(channel),
+            channel_pool,
             exchange: config.exchange,
             routing_key: config.routing_key,
             properties: config.properties,
@@ -108,7 +110,7 @@ impl AmqpSink {
             },
         };
         let service = ServiceBuilder::new().service(AmqpService {
-            channel: Arc::clone(&self.channel),
+            channel_pool: self.channel_pool.clone(),
         });
 
         input

From db7723f72db82f10eb1816bf1e4f42fb04c293df Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Wed, 21 May 2025 22:54:25 -0400
Subject: [PATCH 06/12] Rename AmqpSink

---
 src/sinks/amqp/channel.rs | 30 +++++++++++++++++++++++-------
 src/sinks/amqp/config.rs  | 10 ++++------
 src/sinks/amqp/service.rs | 10 +++++-----
 src/sinks/amqp/sink.rs    | 21 +++++++++------------
 4 files changed, 41 insertions(+), 30 deletions(-)

diff --git a/src/sinks/amqp/channel.rs b/src/sinks/amqp/channel.rs
index 439b7cc4ff9c2..c5ae1ff0a0c45 100644
--- a/src/sinks/amqp/channel.rs
+++ b/src/sinks/amqp/channel.rs
@@ -1,13 +1,27 @@
 use super::service::AmqpError;
 use crate::amqp::AmqpConfig;
+use deadpool::managed::Pool;
 use lapin::options::ConfirmSelectOptions;
 
-/// A wrapper around the AMQP channel that handles reconnections.
-pub(crate) struct AmqpChannel {
+pub type AmqpSinkChannels = Pool<AmqpSinkChannelManager>;
+
+pub(super) fn new_channel_pool(config: &AmqpConfig) -> crate::Result<AmqpSinkChannels> {
+    let channel_manager = AmqpSinkChannelManager::new(config);
+    let channels = Pool::builder(channel_manager)
+        .max_size(4)
+        .runtime(deadpool::Runtime::Tokio1)
+        .build()?;
+    Ok(channels)
+}
+
+/// A channel pool manager for the AMQP sink.
+/// This manager is responsible for creating and recycling AMQP channels.
+/// It uses the `deadpool` crate to manage the channels.
+pub(crate) struct AmqpSinkChannelManager {
     config: AmqpConfig,
 }
 
-impl deadpool::managed::Manager for AmqpChannel {
+impl deadpool::managed::Manager for AmqpSinkChannelManager {
     type Type = lapin::Channel;
     type Error = AmqpError;
 
@@ -25,22 +39,24 @@ impl deadpool::managed::Manager for AmqpChannel {
         channel: &mut Self::Type,
         _: &deadpool::managed::Metrics,
     ) -> deadpool::managed::RecycleResult<Self::Error> {
-        if channel.status().state() == lapin::ChannelState::Connected {
+        let state = channel.status().state();
+        if state == lapin::ChannelState::Connected {
             Ok(())
         } else {
-            Err((AmqpError::ChannelClosed {}).into())
+            Err((AmqpError::ChannelClosed { state }).into())
         }
     }
 }
 
-impl AmqpChannel {
-    /// Creates a new AMQP channel.
+impl AmqpSinkChannelManager {
+    /// Creates a new channel pool manager for the AMQP sink.
     pub fn new(config: &AmqpConfig) -> Self {
         Self {
             config: config.clone(),
         }
     }
 
+    /// Creates a new AMQP channel using the configuration of this sink.
     async fn new_channel(config: &AmqpConfig) -> Result<lapin::Channel, AmqpError> {
         let (_, channel) = config
             .connect()
diff --git a/src/sinks/amqp/config.rs b/src/sinks/amqp/config.rs
index ac86e87be8b1f..d29012b661a06 100644
--- a/src/sinks/amqp/config.rs
+++ b/src/sinks/amqp/config.rs
@@ -1,5 +1,5 @@
 //! Configuration functionality for the `AMQP` sink.
-use super::channel::AmqpChannel;
+use super::channel::AmqpSinkChannels;
 use crate::{amqp::AmqpConfig, sinks::prelude::*};
 use lapin::{types::ShortString, BasicProperties};
 use vector_lib::{
@@ -122,7 +122,7 @@ impl GenerateConfig for AmqpSinkConfig {
 impl SinkConfig for AmqpSinkConfig {
     async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let sink = AmqpSink::new(self.clone()).await?;
-        let hc = healthcheck(sink.channel_pool.clone()).boxed();
+        let hc = healthcheck(sink.channels.clone()).boxed();
         Ok((VectorSink::from_event_streamsink(sink), hc))
     }
 
@@ -135,12 +135,10 @@ impl SinkConfig for AmqpSinkConfig {
     }
 }
 
-pub(super) async fn healthcheck(
-    channel: deadpool::managed::Pool<AmqpChannel>,
-) -> crate::Result<()> {
+pub(super) async fn healthcheck(channels: AmqpSinkChannels) -> crate::Result<()> {
     trace!("Healthcheck started.");
 
-    let channel = channel.get().await?;
+    let channel = channels.get().await?;
 
     if !channel.status().connected() {
         return Err(Box::new(std::io::Error::new(
diff --git a/src/sinks/amqp/service.rs b/src/sinks/amqp/service.rs
index a2545ac9654dd..cee7d20344bd7 100644
--- a/src/sinks/amqp/service.rs
+++ b/src/sinks/amqp/service.rs
@@ -7,7 +7,7 @@ use lapin::{options::BasicPublishOptions, BasicProperties};
 use snafu::Snafu;
 use std::task::{Context, Poll};
 
-use super::channel::AmqpChannel;
+use super::channel::AmqpSinkChannels;
 
 /// The request contains the data to send to `AMQP` together
 /// with the information need to route the message.
@@ -78,7 +78,7 @@ impl DriverResponse for AmqpResponse {
 
 /// The tower service that handles the actual sending of data to `AMQP`.
 pub(super) struct AmqpService {
-    pub(super) channel_pool: deadpool::managed::Pool<AmqpChannel>,
+    pub(super) channels: AmqpSinkChannels,
 }
 
 #[derive(Debug, Snafu)]
@@ -95,8 +95,8 @@ pub enum AmqpError {
     #[snafu(display("Failed to open AMQP channel: {}", error))]
     ConnectFailed { error: vector_common::Error },
 
-    #[snafu(display("Channel is closed."))]
-    ChannelClosed,
+    #[snafu(display("Channel is not writeable: {:?}", state))]
+    ChannelClosed { state: lapin::ChannelState },
 
     #[snafu(display("Channel pool error: {}", error))]
     PoolError { error: vector_common::Error },
@@ -114,7 +114,7 @@ impl Service<AmqpRequest> for AmqpService {
     }
 
     fn call(&mut self, req: AmqpRequest) -> Self::Future {
-        let channel = self.channel_pool.clone();
+        let channel = self.channels.clone();
 
         Box::pin(async move {
             let channel = channel.get().await.map_err(|error| AmqpError::PoolError {
diff --git a/src/sinks/amqp/sink.rs b/src/sinks/amqp/sink.rs
index 6e30e0e4a8a58..a9cd60deee9c6 100644
--- a/src/sinks/amqp/sink.rs
+++ b/src/sinks/amqp/sink.rs
@@ -4,7 +4,7 @@ use crate::sinks::prelude::*;
 use lapin::BasicProperties;
 use serde::Serialize;
 
-use super::channel::AmqpChannel;
+use super::channel::AmqpSinkChannels;
 use super::{
     config::{AmqpPropertiesConfig, AmqpSinkConfig},
     encoder::AmqpEncoder,
@@ -27,7 +27,7 @@ pub(super) struct AmqpEvent {
 }
 
 pub(super) struct AmqpSink {
-    pub(super) channel_pool: deadpool::managed::Pool<AmqpChannel>,
+    pub(super) channels: AmqpSinkChannels,
     exchange: Template,
     routing_key: Option<Template>,
     properties: Option<AmqpPropertiesConfig>,
@@ -37,21 +37,18 @@ pub(super) struct AmqpSink {
 
 impl AmqpSink {
     pub(super) async fn new(config: AmqpSinkConfig) -> crate::Result<Self> {
-        let channel_manager = AmqpChannel::new(&config.connection);
-        let channel_pool = deadpool::managed::Pool::builder(channel_manager)
-            .max_size(4)
-            .runtime(deadpool::Runtime::Tokio1)
-            .build()
-            .map_err(|e| BuildError::AmqpCreateFailed {
-                source: Box::new(e),
-            })?;
+        let channels = super::channel::new_channel_pool(&config.connection).map_err(|e| {
+            BuildError::AmqpCreateFailed {
+                source: e,
+            }
+        })?;
 
         let transformer = config.encoding.transformer();
         let serializer = config.encoding.build()?;
         let encoder = crate::codecs::Encoder::<()>::new(serializer);
 
         Ok(AmqpSink {
-            channel_pool,
+            channels,
             exchange: config.exchange,
             routing_key: config.routing_key,
             properties: config.properties,
@@ -110,7 +107,7 @@ impl AmqpSink {
             },
         };
         let service = ServiceBuilder::new().service(AmqpService {
-            channel_pool: self.channel_pool.clone(),
+            channels: self.channels.clone(),
         });
 
         input

From c5d29f5c6b8bd44ed4c1dcc3245f04a922552879 Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Wed, 21 May 2025 23:05:49 -0400
Subject: [PATCH 07/12] Update integration tests

---
 src/sinks/amqp/channel.rs           | 5 +++--
 src/sinks/amqp/integration_tests.rs | 8 ++++----
 src/sinks/amqp/sink.rs              | 7 ++-----
 3 files changed, 9 insertions(+), 11 deletions(-)

diff --git a/src/sinks/amqp/channel.rs b/src/sinks/amqp/channel.rs
index c5ae1ff0a0c45..d80a469b511e5 100644
--- a/src/sinks/amqp/channel.rs
+++ b/src/sinks/amqp/channel.rs
@@ -1,3 +1,4 @@
+use super::config::AmqpSinkConfig;
 use super::service::AmqpError;
 use crate::amqp::AmqpConfig;
 use deadpool::managed::Pool;
@@ -5,8 +6,8 @@ use lapin::options::ConfirmSelectOptions;
 
 pub type AmqpSinkChannels = Pool<AmqpSinkChannelManager>;
 
-pub(super) fn new_channel_pool(config: &AmqpConfig) -> crate::Result<AmqpSinkChannels> {
-    let channel_manager = AmqpSinkChannelManager::new(config);
+pub(super) fn new_channel_pool(config: &AmqpSinkConfig) -> crate::Result<AmqpSinkChannels> {
+    let channel_manager = AmqpSinkChannelManager::new(&config.connection);
     let channels = Pool::builder(channel_manager)
         .max_size(4)
         .runtime(deadpool::Runtime::Tokio1)
diff --git a/src/sinks/amqp/integration_tests.rs b/src/sinks/amqp/integration_tests.rs
index 1e877ab0a3547..c974dbccb87cc 100644
--- a/src/sinks/amqp/integration_tests.rs
+++ b/src/sinks/amqp/integration_tests.rs
@@ -3,7 +3,7 @@ use crate::{
     amqp::await_connection,
     config::{SinkConfig, SinkContext},
     shutdown::ShutdownSignal,
-    sinks::amqp::channel::AmqpChannel,
+    sinks::amqp::channel::new_channel_pool,
     template::{Template, UnsignedIntTemplate},
     test_util::{
         components::{run_and_assert_sink_compliance, SINK_TAGS},
@@ -13,7 +13,7 @@ use crate::{
 };
 use config::AmqpPropertiesConfig;
 use futures::StreamExt;
-use std::{collections::HashSet, sync::Arc, time::Duration};
+use std::{collections::HashSet, time::Duration};
 use vector_lib::{config::LogNamespace, event::LogEvent};
 
 pub fn make_config() -> AmqpSinkConfig {
@@ -38,8 +38,8 @@ async fn healthcheck() {
     let mut config = make_config();
     config.exchange = Template::try_from(exchange.as_str()).unwrap();
     await_connection(&config.connection).await;
-    let channel = AmqpChannel::new(&config.connection).await.unwrap();
-    super::config::healthcheck(Arc::new(channel)).await.unwrap();
+    let channels = new_channel_pool(&config).unwrap();
+    super::config::healthcheck(channels).await.unwrap();
 }
 
 #[tokio::test]
diff --git a/src/sinks/amqp/sink.rs b/src/sinks/amqp/sink.rs
index a9cd60deee9c6..ca40fe46b21e5 100644
--- a/src/sinks/amqp/sink.rs
+++ b/src/sinks/amqp/sink.rs
@@ -37,11 +37,8 @@ pub(super) struct AmqpSink {
 
 impl AmqpSink {
     pub(super) async fn new(config: AmqpSinkConfig) -> crate::Result<Self> {
-        let channels = super::channel::new_channel_pool(&config.connection).map_err(|e| {
-            BuildError::AmqpCreateFailed {
-                source: e,
-            }
-        })?;
+        let channels = super::channel::new_channel_pool(&config)
+            .map_err(|e| BuildError::AmqpCreateFailed { source: e })?;
 
         let transformer = config.encoding.transformer();
         let serializer = config.encoding.build()?;

From d1c0726e82deecb21ae17840e4f1b2cfa8e98327 Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Wed, 11 Jun 2025 21:58:34 -0400
Subject: [PATCH 08/12] make amqp max_channels configurable; default to 4;
 added to docs and changelog

---
 changelog.d/22971_reconnect_broken_amqp_sink.fix.md |  2 +-
 src/sinks/amqp/channel.rs                           | 13 ++++++++++++-
 src/sinks/amqp/config.rs                            | 12 +++++++++++-
 .../cue/reference/components/sinks/base/amqp.cue    |  5 +++++
 4 files changed, 29 insertions(+), 3 deletions(-)

diff --git a/changelog.d/22971_reconnect_broken_amqp_sink.fix.md b/changelog.d/22971_reconnect_broken_amqp_sink.fix.md
index 31460649084ed..874a47eb9b9b3 100644
--- a/changelog.d/22971_reconnect_broken_amqp_sink.fix.md
+++ b/changelog.d/22971_reconnect_broken_amqp_sink.fix.md
@@ -1,3 +1,3 @@
-The `amqp` sink now attempts to re-connect to the AMQP broker when the channel has been disconnected.
+The `amqp` sink now attempts to re-connect to the AMQP broker when the channel has been disconnected. It will also create up to 4 channels in a pool (configurable with the `max_channels` configuration) to improve throughput.
 
 authors: aramperes
diff --git a/src/sinks/amqp/channel.rs b/src/sinks/amqp/channel.rs
index d80a469b511e5..862ff06e5d0dc 100644
--- a/src/sinks/amqp/channel.rs
+++ b/src/sinks/amqp/channel.rs
@@ -7,11 +7,22 @@ use lapin::options::ConfirmSelectOptions;
 pub type AmqpSinkChannels = Pool<AmqpSinkChannelManager>;
 
 pub(super) fn new_channel_pool(config: &AmqpSinkConfig) -> crate::Result<AmqpSinkChannels> {
+    let max_channels = config.max_channels.try_into().map_err(|_| {
+        Box::new(AmqpError::PoolError {
+            error: "max_channels must fit into usize".into(),
+        })
+    })?;
+    if max_channels == 0 {
+        return Err(Box::new(AmqpError::PoolError {
+            error: "max_channels must be positive".into(),
+        }));
+    }
     let channel_manager = AmqpSinkChannelManager::new(&config.connection);
     let channels = Pool::builder(channel_manager)
-        .max_size(4)
+        .max_size(max_channels)
         .runtime(deadpool::Runtime::Tokio1)
         .build()?;
+    debug!("AMQP channel pool created with max size: {}", max_channels);
     Ok(channels)
 }
 
diff --git a/src/sinks/amqp/config.rs b/src/sinks/amqp/config.rs
index d29012b661a06..afca8b47ccae5 100644
--- a/src/sinks/amqp/config.rs
+++ b/src/sinks/amqp/config.rs
@@ -90,6 +90,14 @@ pub struct AmqpSinkConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     pub(crate) acknowledgements: AcknowledgementsConfig,
+
+    /// Maximum number of AMQP channels to keep active (channels are created as needed).
+    #[serde(default = "default_max_channels")]
+    pub(crate) max_channels: u32,
+}
+
+const fn default_max_channels() -> u32 {
+    4
 }
 
 impl Default for AmqpSinkConfig {
@@ -101,6 +109,7 @@ impl Default for AmqpSinkConfig {
             encoding: TextSerializerConfig::default().into(),
             connection: AmqpConfig::default(),
             acknowledgements: AcknowledgementsConfig::default(),
+            max_channels: default_max_channels(),
         }
     }
 }
@@ -111,7 +120,8 @@ impl GenerateConfig for AmqpSinkConfig {
             r#"connection_string = "amqp://localhost:5672/%2f"
             routing_key = "user_id"
             exchange = "test"
-            encoding.codec = "json""#,
+            encoding.codec = "json"
+            max_channels = 4"#,
         )
         .unwrap()
     }
diff --git a/website/cue/reference/components/sinks/base/amqp.cue b/website/cue/reference/components/sinks/base/amqp.cue
index 9b2f6b79d710d..0915cc06b9c16 100644
--- a/website/cue/reference/components/sinks/base/amqp.cue
+++ b/website/cue/reference/components/sinks/base/amqp.cue
@@ -392,6 +392,11 @@ base: components: sinks: amqp: configuration: {
 		required:    true
 		type: string: syntax: "template"
 	}
+	max_channels: {
+		description: "Maximum number of AMQP channels to keep active (channels are created as needed)."
+		required: 	 false
+		type: uint: default: 4
+	}
 	properties: {
 		description: """
 			Configure the AMQP message properties.

From 6f464161b45dea04b285f2b2728e673272f3efbb Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Wed, 11 Jun 2025 23:24:27 -0400
Subject: [PATCH 09/12] dupe word

---
 changelog.d/22971_reconnect_broken_amqp_sink.fix.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/changelog.d/22971_reconnect_broken_amqp_sink.fix.md b/changelog.d/22971_reconnect_broken_amqp_sink.fix.md
index 874a47eb9b9b3..acd01e4280f13 100644
--- a/changelog.d/22971_reconnect_broken_amqp_sink.fix.md
+++ b/changelog.d/22971_reconnect_broken_amqp_sink.fix.md
@@ -1,3 +1,3 @@
-The `amqp` sink now attempts to re-connect to the AMQP broker when the channel has been disconnected. It will also create up to 4 channels in a pool (configurable with the `max_channels` configuration) to improve throughput.
+The `amqp` sink now attempts to re-connect to the AMQP broker when the channel has been disconnected. It will also create up to 4 channels in a pool (configurable with the `max_channels` setting) to improve throughput.
 
 authors: aramperes

From e19ab8bc4fd517c5ee790b01d7e62524df8ee3e8 Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Wed, 11 Jun 2025 23:31:39 -0400
Subject: [PATCH 10/12] Update 3rd party licenses for deadpool

---
 LICENSE-3rdparty.csv | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/LICENSE-3rdparty.csv b/LICENSE-3rdparty.csv
index e107729c35ddc..aaaaac24bed82 100644
--- a/LICENSE-3rdparty.csv
+++ b/LICENSE-3rdparty.csv
@@ -180,6 +180,7 @@ data-encoding,https://github.com/ia0/data-encoding,MIT,Julien Cretin <git@ia0.eu
 data-url,https://github.com/servo/rust-url,MIT OR Apache-2.0,Simon Sapin <simon.sapin@exyr.org>
 databend-client,https://github.com/databendlabs/bendsql,Apache-2.0,Databend Authors <opensource@databend.com>
 dbl,https://github.com/RustCrypto/utils,MIT OR Apache-2.0,RustCrypto Developers
+deadpool,https://github.com/bikeshedder/deadpool,MIT OR Apache-2.0,Michael P. Jung <michael.jung@terreon.de>
 der,https://github.com/RustCrypto/formats/tree/master/der,Apache-2.0 OR MIT,RustCrypto Developers
 deranged,https://github.com/jhpratt/deranged,MIT OR Apache-2.0,Jacob Pratt <jacob@jhpratt.dev>
 derivative,https://github.com/mcarton/rust-derivative,MIT OR Apache-2.0,mcarton <cartonmartin+git@gmail.com>
@@ -423,6 +424,7 @@ num-integer,https://github.com/rust-num/num-integer,MIT OR Apache-2.0,The Rust P
 num-iter,https://github.com/rust-num/num-iter,MIT OR Apache-2.0,The Rust Project Developers
 num-rational,https://github.com/rust-num/num-rational,MIT OR Apache-2.0,The Rust Project Developers
 num-traits,https://github.com/rust-num/num-traits,MIT OR Apache-2.0,The Rust Project Developers
+num_cpus,https://github.com/seanmonstar/num_cpus,MIT OR Apache-2.0,Sean McArthur <sean@seanmonstar.com>
 num_enum,https://github.com/illicitonion/num_enum,BSD-3-Clause OR MIT OR Apache-2.0,"Daniel Wagner-Hall <dawagner@gmail.com>, Daniel Henry-Mantilla <daniel.henry.mantilla@gmail.com>, Vincent Esche <regexident@gmail.com>"
 num_threads,https://github.com/jhpratt/num_threads,MIT OR Apache-2.0,Jacob Pratt <open-source@jhpratt.dev>
 number_prefix,https://github.com/ogham/rust-number-prefix,MIT,Benjamin Sago <ogham@bsago.me>

From 12b59e55fe4677c15fb7cbadff48853f09e2c227 Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Thu, 12 Jun 2025 11:20:44 -0400
Subject: [PATCH 11/12] Fix formatting in amqp.cue

---
 website/cue/reference/components/sinks/base/amqp.cue | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/website/cue/reference/components/sinks/base/amqp.cue b/website/cue/reference/components/sinks/base/amqp.cue
index 0915cc06b9c16..016cfa8404b0b 100644
--- a/website/cue/reference/components/sinks/base/amqp.cue
+++ b/website/cue/reference/components/sinks/base/amqp.cue
@@ -394,7 +394,7 @@ base: components: sinks: amqp: configuration: {
 	}
 	max_channels: {
 		description: "Maximum number of AMQP channels to keep active (channels are created as needed)."
-		required: 	 false
+		required:    false
 		type: uint: default: 4
 	}
 	properties: {

From b6180e5c3fa9dedc2f67bb7e2f71e987d96ca276 Mon Sep 17 00:00:00 2001
From: Aram Peres <6775216+aramperes@users.noreply.github.com>
Date: Thu, 12 Jun 2025 19:45:55 -0400
Subject: [PATCH 12/12] Update DEVELOPING.md

---
 docs/DEVELOPING.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/DEVELOPING.md b/docs/DEVELOPING.md
index 7b9116a2a806b..356f780e54cb6 100644
--- a/docs/DEVELOPING.md
+++ b/docs/DEVELOPING.md
@@ -126,7 +126,7 @@ Loosely, you'll need the following:
 - **To run integration tests:** Have `docker` available, or a real live version of that service. (Use `AUTOSPAWN=false`)
 - **To run `make check-component-features`:** Have `remarshal` installed.
 - **To run `make check-licenses` or `cargo vdev build licenses`:** Have `dd-rust-license-tool` [installed](https://github.com/DataDog/rust-license-tool).
-- **To run `cargo vdev build component-docs`:** Have `cue` [installed](https://cuelang.org/docs/install/).
+- **To run `make generate-component-docs`:** Have `cue` [installed](https://cuelang.org/docs/install/).
 
 If you find yourself needing to run something inside the Docker environment described above, that's totally fine, they won't collide or hurt each other. In this case, you'd just run `make environment-generate`.
 
@@ -161,7 +161,7 @@ cargo bench transforms::example
 make fmt
 cargo fmt
 # Build component documentation for the website
-cargo vdev build component-docs
+make generate-component-docs
 ```
 
 If you run `make` you'll see a full list of all our tasks. Some of these will start Docker containers, sign commits, or even make releases. These are not common development commands and your mileage may vary.