diff --git a/x-pack/lib/config_management/elasticsearch_source.rb b/x-pack/lib/config_management/elasticsearch_source.rb
index 4f828bfefc..bda52d230d 100644
--- a/x-pack/lib/config_management/elasticsearch_source.rb
+++ b/x-pack/lib/config_management/elasticsearch_source.rb
@@ -9,7 +9,7 @@
 require 'helpers/elasticsearch_options'
 require 'helpers/loggable_try'
 require 'logstash/ssl_file_tracker'
-require 'helpers/ssl_rebuildable'
+require 'helpers/elasticsearch_client_holder'
 require "license_checker/licensed"
 
 module LogStash
@@ -51,7 +51,7 @@ def initialize(settings, ssl_file_tracker = nil)
           ssl_file_tracker.register_paths(SSL_TRACK_ID_LICENSE, paths)
         end
 
-        @rebuildable = LogStash::Helpers::SslRebuildable.new(ssl_file_tracker, SSL_TRACK_ID_CPM) { build_client }
+        @es_client_holder = LogStash::Helpers::ElasticsearchClientHolder.create(ssl_file_tracker, SSL_TRACK_ID_CPM) { build_client }
         setup_license_checker(FEATURE_INTERNAL,
                               ssl_file_tracker: ssl_file_tracker,
                               tracking_id: SSL_TRACK_ID_LICENSE)
@@ -74,8 +74,6 @@ def get_pipeline_fetcher(es_version)
       def pipeline_configs
         logger.trace("Fetch remote config pipeline", :pipeline_ids => pipeline_ids)
 
-        @rebuildable.maybe_rebuild
-
         license_check(true)
         es_version = get_es_version
         fetcher = get_pipeline_fetcher(es_version)
@@ -196,7 +194,7 @@ def pipeline_ids
       end
 
       def client
-        @rebuildable.client
+        @es_client_holder.get
       end
     end
 
diff --git a/x-pack/lib/helpers/elasticsearch_client_holder.rb b/x-pack/lib/helpers/elasticsearch_client_holder.rb
new file mode 100644
index 0000000000..24a6755883
--- /dev/null
+++ b/x-pack/lib/helpers/elasticsearch_client_holder.rb
@@ -0,0 +1,125 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License;
+# you may not use this file except in compliance with the Elastic License.
+
+require "logstash/logging/logger"
+
+module LogStash module Helpers
+  ##
+  # Holds an Elasticsearch client.
+  # @see ElasticsearchClientHolder#create
+  module ElasticsearchClientHolder
+
+    ##
+    # Creates an ElasticsearchClientHolder.
+    # If a `tracker` is specified, an SslRebuildable instance
+    # connected to the tracker with the provided id is returned;
+    # Otherwise, a Lazy instance is returned.
+    # @param tracker [#consume_stale, nil]
+    # @param id [#to_sym, nil]
+    # @yieldreturn [LogStash::Outputs::ElasticSearch::HttpClient]
+    # @return [ElasticsearchClientHolder]
+    def self.create(tracker=nil, id=nil, &client_factory)
+      return Lazy.new(&client_factory) if tracker.nil?
+
+      SslRebuildable.new(tracker, id, &client_factory)
+    end
+
+    ##
+    # Get a current client for immediate use.
+    # Consumers MUST NOT cache the returned client.
+    # @return [LogStash::Outputs::ElasticSearch::HttpClient]
+    def get
+      fail NotImplementedError
+    end
+
+    ##
+    # close the current client, if it exists and is connected
+    # @return [void]
+    def close
+      fail NotImplementedError
+    end
+
+    ##
+    # An ElasticsearchClientHolder that lazily creates the client when it is needed,
+    # caching the result indefinitely.
+    # @api internal (see ElasticsearchClientHolder::create)
+    class Lazy
+      include ElasticsearchClientHolder
+      include LogStash::Util::Loggable
+
+      ##
+      # @yieldreturn [LogStash::Outputs::ElasticSearch::HttpClient]
+      def initialize(&client_factory)
+        fail ArgumentError, "client_factory block is required" unless block_given?
+        @client_factory = client_factory
+      end
+
+      def get
+        @client || Util.synchronize(self) do
+          @client ||= begin
+                        logger.debug("initializing ES client")
+                        @client_factory.call
+                      end
+        end
+      end
+
+      def close
+        Util.synchronize(self) do
+          @client&.close
+        end
+      end
+    end
+
+    ##
+    # An ElasticsearchClientHolder that is connected to the provided tracker by the provided id.
+    # The client is created lazily, and is re-created when the tracker has marked the given id as stale.
+    # @api internal (see ElasticsearchClientHolder::create)
+    class SslRebuildable
+      include ElasticsearchClientHolder
+      include LogStash::Util::Loggable
+
+      attr_reader :tracker
+      attr_reader :id
+
+      ##
+      # @param tracker [#consume_stale]
+      # @param id [#to_sym]
+      # @yieldreturn [LogStash::Outputs::ElasticSearch::HttpClient]
+      def initialize(tracker, id, &client_factory)
+        @tracker = tracker or fail(ArgumentError, "tracker is required")
+        @id = id&.to_sym or fail(ArgumentError, "id is required")
+        @client_factory = client_factory or fail(ArgumentError, "client_factory block is required")
+      end
+
+      def get
+        Util.synchronize(self) do
+          @client ||= begin
+                        logger.debug("initializing rebuildable elasticsearch client `#{@id}`")
+                        @client_factory.call
+                      end
+
+          @tracker.consume_stale(@id) do
+            begin
+              old_client = @client
+              @client = @client_factory.call
+              logger.debug("rebuilt elasticsearch client `#{@id}` on certificate change")
+              old_client&.close rescue logger.warn("error closing stale elasticsearch client `#{@id}`", exception: $!.class, message: $!.message)
+            rescue => e
+              logger.warn("failed to rebuild elasticsearch client `#{@id}`", exception: e.class, message: e.message)
+              raise
+            end
+          end
+
+          @client
+        end
+      end
+
+      def close
+        Util.synchronize(self) do
+          @client&.close
+        end
+      end
+    end
+  end
+end end
diff --git a/x-pack/lib/helpers/ssl_rebuildable.rb b/x-pack/lib/helpers/ssl_rebuildable.rb
deleted file mode 100644
index 378d26b2e4..0000000000
--- a/x-pack/lib/helpers/ssl_rebuildable.rb
+++ /dev/null
@@ -1,57 +0,0 @@
-# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
-# or more contributor license agreements. Licensed under the Elastic License;
-# you may not use this file except in compliance with the Elastic License.
-
-require "logstash/logging/logger"
-
-module LogStash module Helpers
-  # Owns a lazily-built ES client and, when a tracker+id are provided,
-  # rebuilds it on certificate rotation.
-  #
-  # Without a tracker (e.g. ssl.reload.automatic=false), behaves as a plain
-  # lazy client cache with no rebuild logic.
-  class SslRebuildable
-    include LogStash::Util::Loggable
-
-    # @param tracker [LogStash::SslFileTracker, nil] may be nil when SSL auto-reload is disabled
-    # @param id [Symbol, String, nil] tracking id; ignored when tracker is nil
-    # @yield factory that builds a fresh client instance
-    def initialize(tracker, id, &client_factory)
-      raise ArgumentError, "client_factory block is required" unless block_given?
-      @tracker = tracker
-      @id = id&.to_sym
-      @client_factory = client_factory
-    end
-
-    # Lazily builds and returns the current client. Subsequent calls return
-    # the same instance until #invalidate or a #maybe_rebuild consumes stale.
-    def client
-      @client ||= @client_factory.call
-    end
-
-    # Asks the tracker whether the bound id is stale; if so, closes the current
-    # client and eagerly rebuilds a fresh one so the next #client call returns
-    # the new instance immediately. If the rebuild raises, the tracker
-    # re-asserts the stale flag so the next call retries.
-    # @return [Boolean] true if the rebuild path ran
-    def maybe_rebuild
-      return false unless @tracker
-      @tracker.consume_stale(@id) do
-        invalidate
-        client
-        logger.info("Rebuilt client on certificate change")
-      end
-    end
-
-    private
-
-    # Closes and clears the current client. Any close error is logged and swallowed.
-    def invalidate
-      @client&.close
-    rescue => e
-      logger.warn("Error closing stale ES client", :message => e.message)
-    ensure
-      @client = nil
-    end
-  end
-end end
diff --git a/x-pack/lib/license_checker/license_manager.rb b/x-pack/lib/license_checker/license_manager.rb
index 829376abf7..b97fdd6dcb 100644
--- a/x-pack/lib/license_checker/license_manager.rb
+++ b/x-pack/lib/license_checker/license_manager.rb
@@ -51,7 +51,6 @@ def serverless?
       end
 
       def fetch_license
-        @license_reader.maybe_rebuild_client
         fetch_cluster_info
         if serverless?
           update_xpack_info XPackInfo.serverless_response
diff --git a/x-pack/lib/license_checker/license_reader.rb b/x-pack/lib/license_checker/license_reader.rb
index 846edf8608..653846c500 100644
--- a/x-pack/lib/license_checker/license_reader.rb
+++ b/x-pack/lib/license_checker/license_reader.rb
@@ -4,7 +4,7 @@
 
 require 'logstash/logging/logger'
 require 'logstash/outputs/elasticsearch'
-require 'helpers/ssl_rebuildable'
+require 'helpers/elasticsearch_client_holder'
 
 module LogStash
   module LicenseChecker
@@ -20,7 +20,7 @@ def initialize(settings, feature, options, ssl_file_tracker: nil, tracking_id: n
         es_options = options.merge('resurrect_delay' => 30)
         @es_options = Helpers::ElasticsearchOptions::es_options_with_product_origin_header(es_options)
 
-        @rebuildable = LogStash::Helpers::SslRebuildable.new(ssl_file_tracker, tracking_id) { build_client }
+        @es_client_holder = LogStash::Helpers::ElasticsearchClientHolder.create(ssl_file_tracker, tracking_id) { build_client }
       end
 
       ##
@@ -68,11 +68,7 @@ def fetch_cluster_info
       ##
       # @api private
       def client
-        @rebuildable.client
-      end
-
-      def maybe_rebuild_client
-        @rebuildable.maybe_rebuild
+        @es_client_holder.get
       end
 
       private
diff --git a/x-pack/spec/config_management/elasticsearch_source_spec.rb b/x-pack/spec/config_management/elasticsearch_source_spec.rb
index 2f62c28e7c..c0e1490a85 100644
--- a/x-pack/spec/config_management/elasticsearch_source_spec.rb
+++ b/x-pack/spec/config_management/elasticsearch_source_spec.rb
@@ -873,7 +873,7 @@
 
       old_client = double("old_client")
       expect(old_client).to receive(:close)
-      instance.instance_variable_get(:@rebuildable).instance_variable_set(:@client, old_client)
+      instance.instance_variable_get(:@es_client_holder).instance_variable_set(:@client, old_client)
       allow(instance).to receive(:build_client).and_return(double("new_client"))
 
       instance.pipeline_configs
@@ -888,7 +888,7 @@
 
       old_client = double("old_client")
       expect(old_client).not_to receive(:close)
-      instance.instance_variable_get(:@rebuildable).instance_variable_set(:@client, old_client)
+      instance.instance_variable_get(:@es_client_holder).instance_variable_set(:@client, old_client)
 
       instance.pipeline_configs
     end
diff --git a/x-pack/spec/helpers/elasticsearch_client_holder_spec.rb b/x-pack/spec/helpers/elasticsearch_client_holder_spec.rb
new file mode 100644
index 0000000000..14b3b66c86
--- /dev/null
+++ b/x-pack/spec/helpers/elasticsearch_client_holder_spec.rb
@@ -0,0 +1,104 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License;
+# you may not use this file except in compliance with the Elastic License.
+
+require "spec_helper"
+require "helpers/elasticsearch_client_holder"
+require "logstash/ssl_file_tracker"
+
+describe LogStash::Helpers::ElasticsearchClientHolder do
+  let(:tracker) { instance_double(LogStash::SslFileTracker) }
+  let(:tracking_id) { :".cpm" }
+  let(:factory_calls) { [] }
+  let(:factory) { ->() { factory_calls << :built; double("client", close: nil) } }
+  subject(:es_client_holder) { described_class.create(tracker, tracking_id, &factory) }
+
+  describe "#initialize" do
+    it "raises when no factory block is given" do
+      expect { described_class.create(tracker, :".cpm") }
+        .to raise_error(ArgumentError, /client_factory block is required/)
+    end
+  end
+
+  describe "::create" do
+    shared_examples "without factory block" do
+      it "fails helpfully when no factory block is provided" do
+        expect { described_class.create(tracker, tracking_id) }
+          .to raise_error(ArgumentError, /client_factory block is required/)
+      end
+    end
+
+    context "when no tracker is given" do
+      let(:tracker) { nil }
+      it "returns a Lazy instance" do
+        client_holder = described_class.create(tracker, tracking_id, &factory)
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder::Lazy
+      end
+      include_examples "without factory block"
+    end
+
+    context "when a tracker and id are both given" do
+      it "returns an SslRebuildable instance with the provided tracker and id" do
+        client_holder = described_class.create(tracker, tracking_id, &factory)
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder::SslRebuildable
+        expect(client_holder.tracker).to eq(tracker)
+        expect(client_holder.id).to eq(tracking_id)
+      end
+
+      include_examples "without factory block"
+    end
+
+    context "when `tracker` is provided and `id` is not" do
+      let(:tracking_id) { nil }
+      it "fails helpfully" do
+        expect { described_class.create(tracker, tracking_id) }.to raise_error(ArgumentError, /id is required/)
+      end
+    end
+  end
+
+  shared_examples "#get memoization" do
+    it "lazily calls the factory once and caches the result" do
+      es_client_holder
+      expect(factory_calls.size).to eq(0)
+      first = es_client_holder.get
+      second = es_client_holder.get
+      expect(factory_calls.size).to eq(1)
+      expect(second).to be(first)
+    end
+  end
+
+  describe "Lazy implementation" do
+    let(:es_client_holder) { described_class::Lazy.new(&factory) }
+
+    describe "#get" do
+      include_examples "#get memoization"
+    end
+  end
+
+  describe "SslRebuildable implementation" do
+    let(:es_client_holder) { described_class::SslRebuildable.new(tracker, tracking_id, &factory) }
+
+    describe "#get" do
+      before(:each) do
+        # mimic tracking behaviour when the id has NOT been marked stale
+        allow(tracker).to receive(:consume_stale).with(tracking_id).and_return(nil)
+      end
+      include_examples "#get memoization"
+
+      it "reloads the client when it has been invalidated" do
+        existing = es_client_holder.get
+        expect(existing).to receive(:close)
+
+        # mimic tracker behaviour when the id has been marked stale
+        allow(tracker).to receive(:consume_stale).with(tracking_id).and_yield
+
+        expect do
+          fresh = es_client_holder.get
+          expect(fresh).to_not be(existing)
+        end.to change { factory_calls.size }.by(1)
+      end
+    end
+  end
+end
diff --git a/x-pack/spec/helpers/ssl_rebuildable_spec.rb b/x-pack/spec/helpers/ssl_rebuildable_spec.rb
deleted file mode 100644
index 79a69145d4..0000000000
--- a/x-pack/spec/helpers/ssl_rebuildable_spec.rb
+++ /dev/null
@@ -1,80 +0,0 @@
-# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
-# or more contributor license agreements. Licensed under the Elastic License;
-# you may not use this file except in compliance with the Elastic License.
-
-require "spec_helper"
-require "helpers/ssl_rebuildable"
-require "logstash/ssl_file_tracker"
-
-describe LogStash::Helpers::SslRebuildable do
-  let(:tracker) { instance_double(LogStash::SslFileTracker) }
-  let(:factory_calls) { [] }
-  let(:factory) { ->() { factory_calls << :built; double("client", close: nil) } }
-  subject(:rebuildable) { described_class.new(tracker, :".cpm", &factory) }
-
-  describe "#initialize" do
-    it "raises when no factory block is given" do
-      expect { described_class.new(tracker, :".cpm") }
-        .to raise_error(ArgumentError, /client_factory block is required/)
-    end
-  end
-
-  describe "#client" do
-    it "lazily calls the factory once and caches the result" do
-      first = rebuildable.client
-      second = rebuildable.client
-      expect(factory_calls.size).to eq(1)
-      expect(second).to be(first)
-    end
-  end
-
-  describe "#maybe_rebuild" do
-    it "delegates to tracker.consume_stale with the configured id" do
-      expect(tracker).to receive(:consume_stale).with(:".cpm").and_return(false)
-      rebuildable.maybe_rebuild
-    end
-
-    it "closes the existing client and eagerly builds a fresh one when stale" do
-      allow(tracker).to receive(:consume_stale).with(:".cpm").and_yield
-      existing = rebuildable.client
-      expect(existing).to receive(:close)
-      rebuildable.maybe_rebuild
-      expect(factory_calls.size).to eq(2)
-      fresh = rebuildable.client
-      expect(fresh).not_to be(existing)
-    end
-
-    it "is a no-op when the tracker is nil" do
-      untracked = described_class.new(nil, nil, &factory)
-      expect(untracked.maybe_rebuild).to eq(false)
-    end
-
-    it "coerces string ids to symbol" do
-      r = described_class.new(tracker, ".cpm", &factory)
-      expect(tracker).to receive(:consume_stale).with(:".cpm")
-      r.maybe_rebuild
-    end
-  end
-
-  describe "#invalidate" do
-    it "closes and clears the cached client" do
-      existing = rebuildable.client
-      expect(existing).to receive(:close)
-      rebuildable.send(:invalidate)
-      expect(rebuildable.client).not_to be(existing)
-    end
-
-    it "logs and clears when close raises" do
-      failing = double("client")
-      allow(failing).to receive(:close).and_raise(StandardError, "boom")
-      rebuildable.instance_variable_set(:@client, failing)
-      expect(rebuildable.logger).to receive(:warn).with(/Error closing stale ES client/, hash_including(:message))
-      rebuildable.send(:invalidate)
-      expect(rebuildable.instance_variable_get(:@client)).to be_nil
-    end
-
-    it "is safe when no client has been built yet" do
-      expect { rebuildable.send(:invalidate) }.not_to raise_error
-    end
-  end
-end
diff --git a/x-pack/spec/license_checker/license_manager_spec.rb b/x-pack/spec/license_checker/license_manager_spec.rb
index d463639c62..af41a73565 100644
--- a/x-pack/spec/license_checker/license_manager_spec.rb
+++ b/x-pack/spec/license_checker/license_manager_spec.rb
@@ -26,10 +26,6 @@ def update(xpack_info, is_serverless)
 describe LogStash::LicenseChecker::LicenseManager do
   let(:subject) { described_class.new(license_reader, 'monitoring') }
 
-  before do
-    allow(license_reader).to receive(:maybe_rebuild_client).and_return(nil)
-  end
-
   let(:status) { "active"}
   let(:type) { 'trial' }
 
@@ -192,20 +188,4 @@ def update(xpack_info, is_serverless)
       end
     end
   end
-
-  describe "ssl tracking delegation" do
-    before do
-      allow(license_reader).to receive(:fetch_cluster_info).and_return({})
-      allow(license_reader).to receive(:fetch_xpack_info).and_return(LogStash::LicenseChecker::XPackInfo.failed_to_fetch)
-    end
-
-    it "fetch_license asks the reader to maybe_rebuild_client before any HTTP call" do
-      subject
-      expect(license_reader).to receive(:maybe_rebuild_client).ordered
-      expect(license_reader).to receive(:fetch_cluster_info).ordered.and_return({})
-      expect(license_reader).to receive(:fetch_xpack_info).ordered
-        .and_return(LogStash::LicenseChecker::XPackInfo.failed_to_fetch)
-      subject.fetch_license
-    end
-  end
 end
diff --git a/x-pack/spec/license_checker/license_reader_spec.rb b/x-pack/spec/license_checker/license_reader_spec.rb
index 0ffdff45e2..f691d38ee9 100644
--- a/x-pack/spec/license_checker/license_reader_spec.rb
+++ b/x-pack/spec/license_checker/license_reader_spec.rb
@@ -6,6 +6,7 @@
 require 'support/helpers'
 require "license_checker/license_reader"
 require "helpers/elasticsearch_options"
+require "helpers/elasticsearch_client_holder"
 require "monitoring/monitoring"
 require "logstash/runner"
 require "logstash/ssl_file_tracker"
@@ -218,31 +219,38 @@
 
   describe "SSL tracker wiring" do
     let(:tracker) { instance_double(LogStash::SslFileTracker) }
-    let(:license_reader) do
+    subject(:license_reader) do
       described_class.new(system_settings, 'monitoring', elasticsearch_options,
                           ssl_file_tracker: tracker, tracking_id: :".cpm_license")
     end
 
     it "passes the tracker and tracking_id to its SslRebuildable" do
-      rebuildable = license_reader.instance_variable_get(:@rebuildable)
-      expect(rebuildable.instance_variable_get(:@tracker)).to be(tracker)
-      expect(rebuildable.instance_variable_get(:@id)).to eq(:".cpm_license")
+      client_holder = license_reader.instance_variable_get(:@es_client_holder)
+      expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder
+      expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder::SslRebuildable
+      expect(client_holder.id).to eq(:".cpm_license")
+      expect(client_holder.tracker).to eq(tracker)
     end
 
-    it "maybe_rebuild_client delegates to the rebuildable" do
-      rebuildable = subject.instance_variable_get(:@rebuildable)
-      expect(rebuildable).to receive(:maybe_rebuild)
-      subject.maybe_rebuild_client
-    end
+    it "client delegates to the es_client_holder" do
+      client_holder = license_reader.instance_variable_get(:@es_client_holder)
+      expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder
+      expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder::SslRebuildable
+      expect(client_holder).to receive(:get)
 
-    it "client delegates to the rebuildable" do
-      rebuildable = subject.instance_variable_get(:@rebuildable)
-      expect(rebuildable).to receive(:client)
       subject.client
     end
 
-    it "maybe_rebuild_client is safe when no tracker was injected" do
-      expect { subject.maybe_rebuild_client }.not_to raise_error
+    context "when tracker is not provided" do
+      let(:tracker) { nil }
+      it "uses a non-rebuildable elasticsearch client holder" do
+        client_holder = license_reader.instance_variable_get(:@es_client_holder)
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder
+        expect(client_holder).to be_a_kind_of LogStash::Helpers::ElasticsearchClientHolder::Lazy
+        expect(client_holder).to receive(:get)
+
+        subject.client
+      end
     end
   end
 end