diff --git a/Cargo.lock b/Cargo.lock
index fcbddca1e6840..cc93cd3a6b656 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7776,9 +7776,9 @@ dependencies = [
 
 [[package]]
 name = "openssl"
-version = "0.10.80"
+version = "0.10.81"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a45fa2aa886c42762255da344f0a0d313e254066c46aad76f300c3d3da62d967"
+checksum = "77823a27f0babb03091cb9ed9ef80af3b39dbc82f97e8fa530374b7dafd87a45"
 dependencies = [
  "bitflags 2.10.0",
  "cfg-if",
@@ -7816,9 +7816,9 @@ dependencies = [
 
 [[package]]
 name = "openssl-sys"
-version = "0.9.116"
+version = "0.9.117"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f28a22dc7140cda5f096e5e7724a6962ca81a7f8bfd2979f9b18c11af56318c4"
+checksum = "b47e7e6bb2c38cd930d25a23b40fa52e068c10e85f3e03a7f5ba5aaca5713695"
 dependencies = [
  "cc",
  "libc",
@@ -13312,6 +13312,7 @@ dependencies = [
 name = "vector-core"
 version = "0.1.0"
 dependencies = [
+ "arc-swap",
  "async-trait",
  "base64 0.22.1",
  "bitmask-enum",
diff --git a/lib/vector-core/Cargo.toml b/lib/vector-core/Cargo.toml
index 7ed7c20aa0d6b..0cad6d4f39de7 100644
--- a/lib/vector-core/Cargo.toml
+++ b/lib/vector-core/Cargo.toml
@@ -9,6 +9,7 @@ publish = false
 workspace = true
 
 [dependencies]
+arc-swap = { workspace = true }
 async-trait.workspace = true
 bitmask-enum = { version = "2.2.5", default-features = false }
 bytes = { workspace = true, features = ["serde"] }
@@ -34,7 +35,7 @@ metrics-util.workspace = true
 mlua = { workspace = true, optional = true }
 no-proxy = { version = "0.3.6", default-features = false, features = ["serialize"] }
 ordered-float.workspace = true
-openssl = { version = "0.10.80", default-features = false, features = ["vendored"] }
+openssl = { version = "0.10.81", default-features = false, features = ["vendored"] }
 parking_lot = { version = "0.12.5", default-features = false }
 pin-project.workspace = true
 proptest = { workspace = true, optional = true }
diff --git a/lib/vector-core/src/tls/incoming.rs b/lib/vector-core/src/tls/incoming.rs
index 0d959bb170e1d..b22551b2d62da 100644
--- a/lib/vector-core/src/tls/incoming.rs
+++ b/lib/vector-core/src/tls/incoming.rs
@@ -7,6 +7,7 @@ use std::{
     task::{Context, Poll},
 };
 
+use arc_swap::ArcSwap;
 use futures::{FutureExt, Stream, future::BoxFuture, stream};
 use ipnet::IpNet;
 use openssl::{
@@ -24,7 +25,7 @@ use tonic::transport::{Certificate, server::Connected};
 
 use super::{
     CreateAcceptorSnafu, HandshakeSnafu, IncomingListenerSnafu, MaybeTlsSettings, MaybeTlsStream,
-    SslBuildSnafu, TcpBindSnafu, TlsError, TlsSettings,
+    SslBuildSnafu, TcpBindSnafu, TlsAcceptorReloader, TlsError, TlsSettings,
 };
 use crate::tcp::{self, TcpKeepaliveConfig};
 
@@ -43,43 +44,48 @@ impl TlsSettings {
 
 impl MaybeTlsSettings {
     pub async fn bind(&self, addr: &SocketAddr) -> crate::tls::Result<MaybeTlsListener> {
-        let listener = TcpListener::bind(addr).await.context(TcpBindSnafu)?;
-
-        let acceptor = match self {
-            Self::Tls(tls) => Some(tls.acceptor()?),
-            Self::Raw(()) => None,
-        };
-
-        Ok(MaybeTlsListener {
-            listener,
-            acceptor,
-            origin_filter: None,
-        })
+        self.bind_reloadable(addr, None).await
     }
 
     pub async fn bind_with_allowlist(
         &self,
         addr: &SocketAddr,
         allow_origin: Vec<IpNet>,
+    ) -> crate::tls::Result<MaybeTlsListener> {
+        Ok(self
+            .bind_reloadable(addr, None)
+            .await?
+            .with_allowlist(Some(allow_origin)))
+    }
+
+    /// Bind a listener, optionally sharing a [`TlsAcceptorReloader`] so its acceptor can be swapped
+    /// at runtime. When `reloader` is `Some` and TLS is enabled, connections accepted by the
+    /// returned listener use the acceptor the reloader currently holds; otherwise the acceptor is
+    /// fixed for the lifetime of the listener.
+    pub async fn bind_reloadable(
+        &self,
+        addr: &SocketAddr,
+        reloader: Option<TlsAcceptorReloader>,
     ) -> crate::tls::Result<MaybeTlsListener> {
         let listener = TcpListener::bind(addr).await.context(TcpBindSnafu)?;
 
-        let acceptor = match self {
-            Self::Tls(tls) => Some(tls.acceptor()?),
-            Self::Raw(()) => None,
+        let acceptor = match (self, reloader) {
+            (Self::Raw(()), _) => None,
+            (Self::Tls(_), Some(reloader)) => Some(reloader.shared()),
+            (Self::Tls(tls), None) => Some(Arc::new(ArcSwap::from_pointee(tls.acceptor()?))),
         };
 
         Ok(MaybeTlsListener {
             listener,
             acceptor,
-            origin_filter: Some(allow_origin),
+            origin_filter: None,
         })
     }
 }
 
 pub struct MaybeTlsListener {
     listener: TcpListener,
-    acceptor: Option<SslAcceptor>,
+    acceptor: Option<Arc<ArcSwap<SslAcceptor>>>,
     origin_filter: Option<Vec<IpNet>>,
 }
 
@@ -90,7 +96,11 @@ impl MaybeTlsListener {
             .accept()
             .await
             .map(|(stream, peer_addr)| {
-                MaybeTlsIncomingStream::new(stream, peer_addr, self.acceptor.clone())
+                let acceptor = self
+                    .acceptor
+                    .as_ref()
+                    .map(|accptr| accptr.load().as_ref().clone());
+                MaybeTlsIncomingStream::new(stream, peer_addr, acceptor)
             })
             .context(IncomingListenerSnafu)?;
 
@@ -421,10 +431,7 @@ impl From<X509> for CertificateMetadata {
     fn from(cert: X509) -> Self {
         let mut subject_metadata: HashMap<String, String> = HashMap::new();
         for entry in cert.subject_name().entries() {
-            let data_string = match entry.data().as_utf8() {
-                Ok(data) => data.to_string(),
-                Err(_) => String::new(),
-            };
+            let data_string = entry.data().to_string().unwrap_or_default();
             subject_metadata.insert(entry.object().to_string(), data_string);
         }
         Self {
diff --git a/lib/vector-core/src/tls/mod.rs b/lib/vector-core/src/tls/mod.rs
index 5e2e06aa4462b..02435e4da109b 100644
--- a/lib/vector-core/src/tls/mod.rs
+++ b/lib/vector-core/src/tls/mod.rs
@@ -15,10 +15,12 @@ use crate::tcp::{self, TcpKeepaliveConfig};
 mod incoming;
 mod maybe_tls;
 mod outgoing;
+mod reload;
 mod settings;
 
 pub use incoming::{CertificateMetadata, MaybeTlsIncomingStream, MaybeTlsListener};
 pub use maybe_tls::MaybeTls;
+pub use reload::{TlsAcceptorReloader, WeakTlsAcceptorReloader};
 pub use settings::{
     MaybeTlsSettings, PEM_START_MARKER, TEST_PEM_CA_PATH, TEST_PEM_CLIENT_CRT_PATH,
     TEST_PEM_CLIENT_KEY_PATH, TEST_PEM_CRT_PATH, TEST_PEM_INTERMEDIATE_CA_PATH, TEST_PEM_KEY_PATH,
diff --git a/lib/vector-core/src/tls/reload.rs b/lib/vector-core/src/tls/reload.rs
new file mode 100644
index 0000000000000..9309c87729b70
--- /dev/null
+++ b/lib/vector-core/src/tls/reload.rs
@@ -0,0 +1,278 @@
+//! A TLS acceptor that can be swapped at runtime.
+//!
+//! A [`TlsAcceptorReloader`] wraps the acceptor a bound [`MaybeTlsListener`](super::MaybeTlsListener)
+//! serves. Handing the same reloader to
+//! [`MaybeTlsSettings::bind_reloadable`](super::MaybeTlsSettings::bind_reloadable) lets a background
+//! task swap in freshly built material with [`reload`](TlsAcceptorReloader::reload); each new
+//! connection then handshakes with the latest acceptor while in-flight connections keep what they
+//! negotiated.
+
+use std::sync::{Arc, Weak};
+
+use arc_swap::ArcSwap;
+use openssl::ssl::SslAcceptor;
+
+use super::{MaybeTlsSettings, TlsSettings};
+
+/// A cloneable handle to a server TLS acceptor that can be swapped at runtime.
+///
+/// Connections accepted after [`reload`](Self::reload) use the new acceptor; connections already
+/// established keep whatever they negotiated at handshake time.
+#[derive(Clone)]
+pub struct TlsAcceptorReloader {
+    acceptor: Arc<ArcSwap<SslAcceptor>>,
+}
+
+impl TlsAcceptorReloader {
+    /// Wrap an initial acceptor in a swappable cell.
+    pub(super) fn new(acceptor: SslAcceptor) -> Self {
+        Self {
+            acceptor: Arc::new(ArcSwap::from_pointee(acceptor)),
+        }
+    }
+
+    /// The shared cell the bound listener reads from on each accept.
+    pub(super) fn shared(&self) -> Arc<ArcSwap<SslAcceptor>> {
+        Arc::clone(&self.acceptor)
+    }
+
+    /// Swap in a freshly built acceptor from `settings`. New connections pick it up
+    /// immediately; the previous acceptor is dropped once its last in-flight handshake completes.
+    pub fn reload(&self, settings: &TlsSettings) -> crate::tls::Result<()> {
+        self.acceptor.store(Arc::new(settings.acceptor()?));
+        Ok(())
+    }
+
+    /// Downgrade to a [`WeakTlsAcceptorReloader`] that does not keep the served acceptor alive.
+    ///
+    /// This is what a background reload task should hold: once the bound listener (the sole strong
+    /// owner) is dropped at source shutdown, the weak handle's
+    /// [`reload`](WeakTlsAcceptorReloader::reload) reports that it is gone, so the task can stop.
+    pub fn downgrade(&self) -> WeakTlsAcceptorReloader {
+        WeakTlsAcceptorReloader {
+            acceptor: Arc::downgrade(&self.acceptor),
+        }
+    }
+}
+
+/// A non-owning handle to a served TLS acceptor, obtained from [`TlsAcceptorReloader::downgrade`].
+#[derive(Clone)]
+pub struct WeakTlsAcceptorReloader {
+    acceptor: Weak<ArcSwap<SslAcceptor>>,
+}
+
+impl WeakTlsAcceptorReloader {
+    /// Swap in a freshly built acceptor if the listener is still alive.
+    ///
+    /// Returns `Ok(true)` on a successful swap, `Ok(false)` if the listener has been dropped (the
+    /// caller should stop reloading), and `Err` if the new settings failed to build an acceptor.
+    pub fn reload(&self, settings: &TlsSettings) -> crate::tls::Result<bool> {
+        match self.acceptor.upgrade() {
+            Some(acceptor) => {
+                acceptor.store(Arc::new(settings.acceptor()?));
+                Ok(true)
+            }
+            None => Ok(false),
+        }
+    }
+}
+
+impl MaybeTlsSettings {
+    /// Build a reloadable acceptor handle for server use, or `None` when TLS is disabled.
+    ///
+    /// Pass the returned handle to [`bind_reloadable`](Self::bind_reloadable) so the bound listener
+    /// serves it, and keep a clone to call [`reload`](TlsAcceptorReloader::reload) when the
+    /// certificate material rotates.
+    pub fn reloadable_acceptor(&self) -> crate::tls::Result<Option<TlsAcceptorReloader>> {
+        match self {
+            Self::Tls(tls) => Ok(Some(TlsAcceptorReloader::new(tls.acceptor()?))),
+            Self::Raw(()) => Ok(None),
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::{net::SocketAddr, pin::Pin};
+
+    use openssl::{
+        asn1::Asn1Time,
+        bn::{BigNum, MsbOption},
+        hash::MessageDigest,
+        nid::Nid,
+        pkey::PKey,
+        rsa::Rsa,
+        ssl::{SslConnector, SslMethod, SslVerifyMode},
+        x509::{X509, X509NameBuilder},
+    };
+
+    use crate::tls::{MaybeTls, MaybeTlsSettings, TlsConfig, TlsEnableableConfig};
+
+    #[test]
+    fn no_reloadable_acceptor_without_tls() {
+        assert!(
+            MaybeTlsSettings::Raw(())
+                .reloadable_acceptor()
+                .unwrap()
+                .is_none(),
+            "plaintext settings have no acceptor to reload"
+        );
+    }
+
+    #[tokio::test]
+    async fn reloadable_acceptor_swaps_and_detects_shutdown() {
+        let settings =
+            MaybeTlsSettings::from_config(Some(&TlsEnableableConfig::test_config()), true).unwrap();
+        let tls = match &settings {
+            MaybeTls::Tls(tls) => tls.clone(),
+            MaybeTls::Raw(()) => panic!("expected TLS to be enabled"),
+        };
+
+        let reloader = settings
+            .reloadable_acceptor()
+            .unwrap()
+            .expect("tls enabled, so an acceptor should exist");
+        let weak = reloader.downgrade();
+
+        // Binding takes over the reloader's (sole) strong reference to the served acceptor.
+        let addr = "127.0.0.1:0".parse().unwrap();
+        let listener = settings
+            .bind_reloadable(&addr, Some(reloader))
+            .await
+            .unwrap();
+
+        // While the listener is alive, a reload swaps in freshly built material.
+        assert!(
+            weak.reload(&tls).unwrap(),
+            "reload should apply while the listener is alive"
+        );
+
+        // Once the listener (the last strong owner) is dropped, the weak handle reports it is gone
+        // so the reload task can stop.
+        drop(listener);
+        assert!(
+            !weak.reload(&tls).unwrap(),
+            "reload should report the listener is gone after shutdown"
+        );
+    }
+
+    /// End-to-end: bind a reloadable TLS listener, complete a real handshake and confirm the served
+    /// leaf certificate, then reload with a different certificate and confirm a fresh connection is
+    /// served the new one.
+    #[tokio::test]
+    async fn served_certificate_changes_after_reload() {
+        let (crt_a, key_a) = self_signed("old.example");
+        let (crt_b, key_b) = self_signed("new.example");
+
+        let settings = server_settings(&crt_a, &key_a);
+        let reloader = settings
+            .reloadable_acceptor()
+            .unwrap()
+            .expect("tls enabled, so an acceptor should exist");
+
+        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
+        let mut listener = settings
+            .bind_reloadable(&addr, Some(reloader.clone()))
+            .await
+            .unwrap();
+        let local_addr = listener.local_addr().unwrap();
+
+        // Accept and complete the server side of each handshake until the test drops the listener.
+        let server = tokio::spawn(async move {
+            while let Ok(mut stream) = listener.accept().await {
+                // The client may drop as soon as it has the cert, so a handshake error is expected.
+                stream.handshake().await.ok();
+            }
+        });
+
+        // Before any reload, the original certificate is served.
+        assert_eq!(served_common_name(local_addr).await, "old.example");
+
+        // Reload with a different certificate...
+        let settings_b = server_settings(&crt_b, &key_b);
+        let tls_b = match &settings_b {
+            MaybeTls::Tls(tls) => tls.clone(),
+            MaybeTls::Raw(()) => unreachable!(),
+        };
+        reloader.reload(&tls_b).unwrap();
+
+        // ...and a new connection is served the rotated certificate.
+        assert_eq!(served_common_name(local_addr).await, "new.example");
+
+        server.abort();
+    }
+
+    /// Connect as a TLS client (trusting any server cert) and return the CN of the presented leaf.
+    async fn served_common_name(addr: SocketAddr) -> String {
+        let tcp = tokio::net::TcpStream::connect(addr).await.unwrap();
+
+        let mut builder = SslConnector::builder(SslMethod::tls()).unwrap();
+        builder.set_verify(SslVerifyMode::NONE);
+        let mut config = builder.build().configure().unwrap();
+        config.set_verify_hostname(false);
+        let ssl = config.into_ssl("localhost").unwrap();
+
+        let mut stream = tokio_openssl::SslStream::new(ssl, tcp).unwrap();
+        Pin::new(&mut stream).connect().await.unwrap();
+
+        let cert = stream
+            .ssl()
+            .peer_certificate()
+            .expect("server presents a certificate");
+        cert.subject_name()
+            .entries_by_nid(Nid::COMMONNAME)
+            .next()
+            .unwrap()
+            .data()
+            .to_string()
+            .unwrap()
+    }
+
+    fn server_settings(crt_pem: &str, key_pem: &str) -> MaybeTlsSettings {
+        // `crt_file`/`key_file` accept inline PEM (detected by the `-----BEGIN ` marker), so no
+        // temp files are needed.
+        let config = TlsEnableableConfig {
+            enabled: Some(true),
+            options: TlsConfig {
+                crt_file: Some(crt_pem.into()),
+                key_file: Some(key_pem.into()),
+                ..Default::default()
+            },
+        };
+        MaybeTlsSettings::from_config(Some(&config), true).unwrap()
+    }
+
+    /// Generate a self-signed certificate/key pair (PEM) with the given common name.
+    fn self_signed(common_name: &str) -> (String, String) {
+        let key = PKey::from_rsa(Rsa::generate(2048).unwrap()).unwrap();
+
+        let mut name = X509NameBuilder::new().unwrap();
+        name.append_entry_by_text("CN", common_name).unwrap();
+        let name = name.build();
+
+        let mut serial = BigNum::new().unwrap();
+        serial.rand(128, MsbOption::MAYBE_ZERO, false).unwrap();
+
+        let mut builder = X509::builder().unwrap();
+        builder.set_version(2).unwrap();
+        builder
+            .set_serial_number(&serial.to_asn1_integer().unwrap())
+            .unwrap();
+        builder.set_subject_name(&name).unwrap();
+        builder.set_issuer_name(&name).unwrap();
+        builder.set_pubkey(&key).unwrap();
+        builder
+            .set_not_before(&Asn1Time::days_from_now(0).unwrap())
+            .unwrap();
+        builder
+            .set_not_after(&Asn1Time::days_from_now(1).unwrap())
+            .unwrap();
+        builder.sign(&key, MessageDigest::sha256()).unwrap();
+        let cert = builder.build();
+
+        (
+            String::from_utf8(cert.to_pem().unwrap()).unwrap(),
+            String::from_utf8(key.private_key_to_pem_pkcs8().unwrap()).unwrap(),
+        )
+    }
+}
diff --git a/lib/vector-core/src/tls/settings.rs b/lib/vector-core/src/tls/settings.rs
index c301537082c6f..cf76a08ba0af4 100644
--- a/lib/vector-core/src/tls/settings.rs
+++ b/lib/vector-core/src/tls/settings.rs
@@ -1,8 +1,10 @@
 use std::{
+    collections::HashMap,
     fmt,
     fs::File,
     io::Read,
     path::{Path, PathBuf},
+    sync::{LazyLock, Mutex},
 };
 
 use super::{
@@ -324,9 +326,11 @@ impl TlsSettings {
 
         if let Some(alpn) = &self.alpn_protocols {
             if for_server {
-                let server_proto = alpn.clone();
-                // See https://github.com/sfackler/rust-openssl/pull/2360.
-                let server_proto_ref: &'static [u8] = Box::leak(server_proto.into_boxed_slice());
+                // The server ALPN select callback requires a `'static` protocol list (see
+                // https://github.com/sfackler/rust-openssl/pull/2360). Intern it so the intentional
+                // leak happens at most once per distinct list, rather than leaking a fresh copy
+                // every time the context is (re)built.
+                let server_proto_ref = intern_alpn_protocols(alpn);
                 context.set_alpn_select_callback(move |_, client_proto| {
                     select_next_proto(server_proto_ref, client_proto).ok_or(AlpnError::NOACK)
                 });
@@ -354,6 +358,26 @@ impl TlsSettings {
     }
 }
 
+/// Return a `'static` copy of a server ALPN protocol list, leaking each distinct list at most once.
+///
+/// `SslContextBuilder::set_alpn_select_callback` requires the protocol list to outlive the context
+/// with a `'static` lifetime, so the bytes must be leaked. Interning by content means rebuilding an
+/// acceptor with the same ALPN configuration (e.g. on every certificate reload) reuses the existing
+/// allocation instead of leaking a fresh copy each time, keeping the leak bounded and one-time.
+fn intern_alpn_protocols(protocols: &[u8]) -> &'static [u8] {
+    static INTERNED: LazyLock<Mutex<HashMap<Vec<u8>, &'static [u8]>>> =
+        LazyLock::new(|| Mutex::new(HashMap::new()));
+
+    let mut interned = INTERNED.lock().expect("mutex poisoned");
+
+    if let Some(existing) = interned.get(protocols).copied() {
+        return existing;
+    }
+    let leaked: &'static [u8] = Box::leak(protocols.to_vec().into_boxed_slice());
+    interned.insert(protocols.to_vec(), leaked);
+    leaked
+}
+
 impl TlsConfig {
     fn load_authorities(&self) -> Result<Vec<X509>> {
         match &self.ca_file {
diff --git a/src/sources/fluent/mod.rs b/src/sources/fluent/mod.rs
index d52af0d9f282c..30dcf8d4afee1 100644
--- a/src/sources/fluent/mod.rs
+++ b/src/sources/fluent/mod.rs
@@ -217,6 +217,7 @@ impl FluentTcpConfig {
             self.keepalive,
             shutdown_secs,
             tls,
+            None, // tls_reloader: not wired for this source
             tls_client_metadata_key,
             self.receive_buffer_bytes,
             None,
diff --git a/src/sources/logstash.rs b/src/sources/logstash.rs
index cc47259c02ddc..ac7481c053835 100644
--- a/src/sources/logstash.rs
+++ b/src/sources/logstash.rs
@@ -161,6 +161,7 @@ impl SourceConfig for LogstashConfig {
             self.keepalive,
             shutdown_secs,
             tls,
+            None, // tls_reloader: not wired for this source
             tls_client_metadata_key,
             self.receive_buffer_bytes,
             None,
diff --git a/src/sources/socket/mod.rs b/src/sources/socket/mod.rs
index a09887feead71..66149775d5aa8 100644
--- a/src/sources/socket/mod.rs
+++ b/src/sources/socket/mod.rs
@@ -139,6 +139,7 @@ impl SourceConfig for SocketConfig {
                     config.keepalive(),
                     config.shutdown_timeout_secs(),
                     tls,
+                    None, // tls_reloader: not wired for this source
                     tls_client_metadata_key,
                     config.receive_buffer_bytes(),
                     config.max_connection_duration_secs(),
diff --git a/src/sources/statsd/mod.rs b/src/sources/statsd/mod.rs
index 2b55b861012a0..5c9754e7fd0a5 100644
--- a/src/sources/statsd/mod.rs
+++ b/src/sources/statsd/mod.rs
@@ -219,6 +219,7 @@ impl SourceConfig for StatsdConfig {
                     config.keepalive,
                     config.shutdown_timeout_secs,
                     tls,
+                    None, // tls_reloader: not wired for this source
                     tls_client_metadata_key,
                     config.receive_buffer_bytes,
                     None,
diff --git a/src/sources/syslog.rs b/src/sources/syslog.rs
index b3c514d1aff0a..419247d7ec2ec 100644
--- a/src/sources/syslog.rs
+++ b/src/sources/syslog.rs
@@ -206,6 +206,7 @@ impl SourceConfig for SyslogConfig {
                     keepalive,
                     shutdown_secs,
                     tls,
+                    None, // tls_reloader: not wired for this source
                     tls_client_metadata_key,
                     receive_buffer_bytes,
                     None,
diff --git a/src/sources/util/framestream.rs b/src/sources/util/framestream.rs
index 02c1b5e4a9ed3..b7bb0a16d7fdb 100644
--- a/src/sources/util/framestream.rs
+++ b/src/sources/util/framestream.rs
@@ -419,6 +419,7 @@ pub fn build_framestream_tcp_source(
             addr,
             listenfd,
             &tls,
+            None, // tls_reloader: not wired for this source
             frame_handler
                 .allowed_origins()
                 .map(|origins| origins.to_vec()),
diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs
index c1bf2d8600879..aebce3dc9b6ee 100644
--- a/src/sources/util/net/tcp/mod.rs
+++ b/src/sources/util/net/tcp/mod.rs
@@ -26,7 +26,10 @@ use vector_lib::{
     shutdown::ShutdownSignal,
     source_sender::SourceSender,
     tcp::TcpKeepaliveConfig,
-    tls::{CertificateMetadata, MaybeTlsIncomingStream, MaybeTlsListener, MaybeTlsSettings},
+    tls::{
+        CertificateMetadata, MaybeTlsIncomingStream, MaybeTlsListener, MaybeTlsSettings,
+        TlsAcceptorReloader,
+    },
 };
 use vrl::value::ObjectMap;
 
@@ -49,10 +52,14 @@ pub async fn try_bind_tcp_listener(
     addr: SocketListenAddr,
     mut listenfd: ListenFd,
     tls: &MaybeTlsSettings,
+    tls_reloader: Option<TlsAcceptorReloader>,
     allowlist: Option<Vec<IpNet>>,
 ) -> crate::Result<MaybeTlsListener> {
     match addr {
-        SocketListenAddr::SocketAddr(addr) => tls.bind(&addr).await.map_err(Into::into),
+        SocketListenAddr::SocketAddr(addr) => tls
+            .bind_reloadable(&addr, tls_reloader)
+            .await
+            .map_err(Into::into),
         SocketListenAddr::SystemdFd(offset) => match listenfd.take_tcp_listener(offset)? {
             Some(listener) => TcpListener::from_std(listener)
                 .map(Into::into)
@@ -115,6 +122,7 @@ where
         keepalive: Option<TcpKeepaliveConfig>,
         shutdown_timeout_secs: Duration,
         tls: MaybeTlsSettings,
+        tls_reloader: Option<TlsAcceptorReloader>,
         tls_client_metadata_key: Option<OwnedValuePath>,
         receive_buffer_bytes: Option<usize>,
         max_connection_duration_secs: Option<u64>,
@@ -129,7 +137,7 @@ where
 
         Ok(Box::pin(async move {
             let listenfd = ListenFd::from_env();
-            let listener = try_bind_tcp_listener(addr, listenfd, &tls, allowlist)
+            let listener = try_bind_tcp_listener(addr, listenfd, &tls, tls_reloader, allowlist)
                 .await
                 .map_err(|error| {
                     emit!(SocketBindError {