From b83d3c7691e5987d6313eb95be6ac33c0f0558c8 Mon Sep 17 00:00:00 2001
From: Orestis Floros <orestis.floros@elastic.co>
Date: Mon, 6 Jul 2026 16:24:02 +0200
Subject: [PATCH 1/5] [libbeat/processing] add behavioral contract tests for
 processor lifecycle

Black-box tests written against current main by an author with no
knowledge of the upcoming processor-sharing implementation, so they pin
the observable contract rather than implementation details:
construction through the registry, lifecycle (double-Close tolerance,
run after close), SetPaths semantics, cross-pipeline independence
(closing one pipeline's processors must not affect another pipeline
built from an identical config) and concurrent construct/run/close
churn, at both the libbeat/processors and filebeat/channel level.

They pass unmodified on this commit (pre-change) and must keep passing
after the sharing change lands; any commit that needs to modify them is
changing the contract and must show that in its diff.

Also extend the channel runner mock to close the per-client processor
groups, pinning that repeated Close of a group is tolerated and that
owners release their processors on shutdown.

Assisted-By: GitHub Copilot CLI
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 filebeat/channel/contract_test.go    | 513 +++++++++++++++++
 filebeat/channel/runner_mock_test.go |   9 +
 libbeat/processors/contract_test.go  | 830 +++++++++++++++++++++++++++
 3 files changed, 1352 insertions(+)
 create mode 100644 filebeat/channel/contract_test.go
 create mode 100644 libbeat/processors/contract_test.go

diff --git a/filebeat/channel/contract_test.go b/filebeat/channel/contract_test.go
new file mode 100644
index 000000000000..92df1aa774a4
--- /dev/null
+++ b/filebeat/channel/contract_test.go
@@ -0,0 +1,513 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Black-box contracts for per-client processors constructed through
+// RunnerFactoryWithCommonInputSettings.
+
+package channel_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/v7/filebeat/channel"
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/cfgfile"
+	"github.com/elastic/beats/v7/libbeat/processors"
+	_ "github.com/elastic/beats/v7/libbeat/processors/actions"
+	conf "github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
+	"github.com/elastic/elastic-agent-libs/mapstr"
+)
+
+const chanCloserName = "contract_test_channel_closer"
+
+var chanIDCounter atomic.Int64
+
+func chanUniqueID(name string) string {
+	return fmt.Sprintf("%s-%d", name, chanIDCounter.Add(1))
+}
+
+// chanCloserState records calls received by an underlying mock processor.
+type chanCloserState struct {
+	id    string
+	field string
+	value string
+
+	mu             sync.Mutex
+	closed         bool
+	closeCount     int
+	runsAfterClose int
+}
+
+func (p *chanCloserState) Run(event *beat.Event) (*beat.Event, error) {
+	p.mu.Lock()
+	if p.closed {
+		p.runsAfterClose++
+		p.mu.Unlock()
+		return nil, fmt.Errorf("contract mock processor %q ran after Close", p.id)
+	}
+	p.mu.Unlock()
+
+	if _, err := event.PutValue(p.field, p.value); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func (p *chanCloserState) Close() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.closeCount++
+	p.closed = true
+	return nil
+}
+
+func (p *chanCloserState) String() string {
+	return fmt.Sprintf("contract_test_channel_mock(id=%s)", p.id)
+}
+
+type chanTracker struct {
+	mu        sync.Mutex
+	instances map[string][]*chanCloserState
+}
+
+var chanInstances = &chanTracker{instances: map[string][]*chanCloserState{}}
+
+func (tr *chanTracker) add(st *chanCloserState) {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	tr.instances[st.id] = append(tr.instances[st.id], st)
+}
+
+func (tr *chanTracker) get(id string) []*chanCloserState {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	return append([]*chanCloserState(nil), tr.instances[id]...)
+}
+
+func init() {
+	processors.RegisterPlugin(chanCloserName,
+		func(cfg *conf.C, _ *logp.Logger) (beat.Processor, error) {
+			c := struct {
+				ID    string `config:"id"`
+				Field string `config:"field"`
+				Value string `config:"value"`
+			}{}
+			if cfg != nil {
+				if err := cfg.Unpack(&c); err != nil {
+					return nil, err
+				}
+			}
+			if c.ID == "" || c.Field == "" {
+				return nil, errors.New("contract_test_channel mock: 'id' and 'field' are required")
+			}
+			st := &chanCloserState{id: c.ID, field: c.Field, value: c.Value}
+			chanInstances.add(st)
+			return st, nil
+		})
+}
+
+type nopRunner struct{}
+
+func (nopRunner) String() string { return "contract-test-runner" }
+func (nopRunner) Start()         {}
+func (nopRunner) Stop()          {}
+
+// connectorCapturingFactory exposes the connector passed to a runner.
+type connectorCapturingFactory struct {
+	mu         sync.Mutex
+	connectors []beat.PipelineConnector
+}
+
+func (f *connectorCapturingFactory) Create(p beat.PipelineConnector, _ *conf.C) (cfgfile.Runner, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.connectors = append(f.connectors, p)
+	return nopRunner{}, nil
+}
+
+func (f *connectorCapturingFactory) CheckConfig(*conf.C) error { return nil }
+
+func (f *connectorCapturingFactory) last(t testing.TB) beat.PipelineConnector {
+	t.Helper()
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	require.NotEmpty(t, f.connectors)
+	return f.connectors[len(f.connectors)-1]
+}
+
+type recordedClient struct{ cfg beat.ClientConfig }
+
+func (recordedClient) Publish(beat.Event)      {}
+func (recordedClient) PublishAll([]beat.Event) {}
+func (recordedClient) Close() error            { return nil }
+
+type stubPipeline struct{}
+
+func (stubPipeline) ConnectWith(cfg beat.ClientConfig) (beat.Client, error) {
+	return &recordedClient{cfg: cfg}, nil
+}
+
+func (p stubPipeline) Connect() (beat.Client, error) {
+	return p.ConnectWith(beat.ClientConfig{})
+}
+
+func (stubPipeline) Disconnect(context.Context) error { return nil }
+
+func chanBeatInfo(t testing.TB) beat.Info {
+	return beat.Info{
+		Beat:    "contractbeat",
+		Version: "9.9.9",
+		Logger:  logptest.NewTestingLogger(t, ""),
+	}
+}
+
+func chanInputCfg(id, value string) mapstr.M {
+	return mapstr.M{
+		"type":  "filestream",
+		"index": "contract-index",
+		"processors": []mapstr.M{
+			{chanCloserName: mapstr.M{"id": id, "field": "marker", "value": value}},
+			{"add_fields": mapstr.M{"target": "", "fields": mapstr.M{"static": "static-value"}}},
+		},
+	}
+}
+
+func createInputConnector(t testing.TB, raw mapstr.M) beat.PipelineConnector {
+	t.Helper()
+	cfg, err := conf.NewConfigFrom(raw)
+	require.NoError(t, err)
+
+	factory := &connectorCapturingFactory{}
+	rfwc := channel.RunnerFactoryWithCommonInputSettings(chanBeatInfo(t), factory)
+	_, err = rfwc.Create(stubPipeline{}, cfg)
+	require.NoError(t, err)
+	return factory.last(t)
+}
+
+func connectProcessors(t testing.TB, connector beat.PipelineConnector) beat.ProcessorList {
+	t.Helper()
+	client, err := connector.ConnectWith(beat.ClientConfig{})
+	require.NoError(t, err)
+	rc, ok := client.(*recordedClient)
+	require.True(t, ok)
+	require.NotNil(t, rc.cfg.Processing.Processor)
+	return rc.cfg.Processing.Processor
+}
+
+func chanEvent() *beat.Event {
+	return &beat.Event{
+		Timestamp: time.Now(),
+		Fields:    mapstr.M{"message": "contract-test"},
+	}
+}
+
+func chanRequireMarker(t *testing.T, evt *beat.Event, err error, field, want string) {
+	t.Helper()
+	require.NoError(t, err)
+	require.NotNil(t, evt)
+	got, gerr := evt.GetValue(field)
+	require.NoError(t, gerr, "field %q missing from event", field)
+	require.Equal(t, want, got)
+}
+
+// A Run racing with or after Close may fail or return a correctly processed
+// event, but it must not silently drop or misprocess the event.
+func chanCheckErrOrCorrect(evt *beat.Event, err error, field, want string) error {
+	if err != nil {
+		return nil //nolint:nilerr // an error result is an acceptable outcome here
+	}
+	if evt == nil {
+		return errors.New("event was dropped without an error")
+	}
+	got, gerr := evt.GetValue(field)
+	if gerr != nil {
+		return fmt.Errorf("run succeeded but marker field %q is missing: %w", field, gerr)
+	}
+	if got != want {
+		return fmt.Errorf("marker field %q has value %v, want %q", field, got, want)
+	}
+	return nil
+}
+
+func chanAssertLifecycleInvariants(t *testing.T, id string) {
+	t.Helper()
+	for i, st := range chanInstances.get(id) {
+		st.mu.Lock()
+		closeCount, runsAfterClose := st.closeCount, st.runsAfterClose
+		st.mu.Unlock()
+		assert.LessOrEqualf(t, closeCount, 1,
+			"instance %d of %q: underlying Close must be called at most once per constructed instance", i, id)
+		assert.Zerof(t, runsAfterClose,
+			"instance %d of %q: underlying Run must never be called after Close", i, id)
+	}
+}
+
+func chanAssertAllClosed(t *testing.T, id string) {
+	t.Helper()
+	insts := chanInstances.get(id)
+	require.NotEmptyf(t, insts, "no mock instances were constructed for id %q", id)
+	for i, st := range insts {
+		st.mu.Lock()
+		closed := st.closed
+		st.mu.Unlock()
+		assert.Truef(t, closed,
+			"instance %d of %q: closing all owning clients' processors must close the processor", i, id)
+	}
+}
+
+func TestContractInputProcessorConstruction(t *testing.T) {
+	id := chanUniqueID("input-basic")
+	connector := createInputConnector(t, chanInputCfg(id, "input-value"))
+	procs := connectProcessors(t, connector)
+
+	evt, err := procs.Run(chanEvent())
+	chanRequireMarker(t, evt, err, "marker", "input-value")
+	chanRequireMarker(t, evt, err, "static", "static-value")
+	chanRequireMarker(t, evt, err, "@metadata.raw_index", "contract-index")
+
+	assert.NoError(t, procs.Close())
+	chanAssertAllClosed(t, id)
+	assert.NoError(t, procs.Close())
+	chanAssertLifecycleInvariants(t, id)
+
+	revt, rerr := procs.Run(chanEvent())
+	assert.NoError(t, chanCheckErrOrCorrect(revt, rerr, "marker", "input-value"))
+}
+
+func TestContractInputConstructionErrorPropagation(t *testing.T) {
+	t.Run("unknown processor", func(t *testing.T) {
+		raw := mapstr.M{
+			"processors": []mapstr.M{
+				{"contract_test_channel_does_not_exist": mapstr.M{}},
+			},
+		}
+		cfg, err := conf.NewConfigFrom(raw)
+		require.NoError(t, err)
+
+		factory := &connectorCapturingFactory{}
+		rfwc := channel.RunnerFactoryWithCommonInputSettings(chanBeatInfo(t), factory)
+		_, err = rfwc.Create(stubPipeline{}, cfg)
+		if err == nil {
+			// The error may surface when the client connects instead.
+			_, err = factory.last(t).ConnectWith(beat.ClientConfig{})
+		}
+		assert.ErrorContains(t, err, "contract_test_channel_does_not_exist")
+	})
+
+	t.Run("failing processor configuration", func(t *testing.T) {
+		raw := mapstr.M{
+			"processors": []mapstr.M{
+				{chanCloserName: mapstr.M{"value": "no-id-no-field"}},
+			},
+		}
+		cfg, err := conf.NewConfigFrom(raw)
+		require.NoError(t, err)
+
+		factory := &connectorCapturingFactory{}
+		rfwc := channel.RunnerFactoryWithCommonInputSettings(chanBeatInfo(t), factory)
+		_, err = rfwc.Create(stubPipeline{}, cfg)
+		if err == nil {
+			_, err = factory.last(t).ConnectWith(beat.ClientConfig{})
+		}
+		assert.ErrorContains(t, err, "'id' and 'field' are required")
+	})
+}
+
+func TestContractInputClientIndependence(t *testing.T) {
+	t.Run("clients of the same input", func(t *testing.T) {
+		id := chanUniqueID("same-input")
+		connector := createInputConnector(t, chanInputCfg(id, "same-input-value"))
+
+		procsA := connectProcessors(t, connector)
+		procsB := connectProcessors(t, connector)
+
+		evt, err := procsA.Run(chanEvent())
+		chanRequireMarker(t, evt, err, "marker", "same-input-value")
+		evt, err = procsB.Run(chanEvent())
+		chanRequireMarker(t, evt, err, "marker", "same-input-value")
+
+		assert.NoError(t, procsA.Close())
+
+		for range 3 {
+			evt, err = procsB.Run(chanEvent())
+			chanRequireMarker(t, evt, err, "marker", "same-input-value")
+			chanRequireMarker(t, evt, err, "static", "static-value")
+		}
+
+		assert.NoError(t, procsB.Close())
+		chanAssertAllClosed(t, id)
+		chanAssertLifecycleInvariants(t, id)
+	})
+
+	t.Run("separate inputs with identical configuration", func(t *testing.T) {
+		id := chanUniqueID("identical-inputs")
+		connectorA := createInputConnector(t, chanInputCfg(id, "identical-value"))
+		connectorB := createInputConnector(t, chanInputCfg(id, "identical-value"))
+
+		procsA := connectProcessors(t, connectorA)
+		procsB := connectProcessors(t, connectorB)
+
+		evt, err := procsA.Run(chanEvent())
+		chanRequireMarker(t, evt, err, "marker", "identical-value")
+
+		assert.NoError(t, procsA.Close())
+
+		for range 3 {
+			evt, err = procsB.Run(chanEvent())
+			chanRequireMarker(t, evt, err, "marker", "identical-value")
+		}
+
+		assert.NoError(t, procsB.Close())
+		chanAssertAllClosed(t, id)
+		chanAssertLifecycleInvariants(t, id)
+	})
+
+	t.Run("separate inputs with different configurations", func(t *testing.T) {
+		idA := chanUniqueID("input-a")
+		idB := chanUniqueID("input-b")
+		procsA := connectProcessors(t, createInputConnector(t, chanInputCfg(idA, "value-a")))
+		procsB := connectProcessors(t, createInputConnector(t, chanInputCfg(idB, "value-b")))
+
+		evt, err := procsA.Run(chanEvent())
+		chanRequireMarker(t, evt, err, "marker", "value-a")
+		evt, err = procsB.Run(chanEvent())
+		chanRequireMarker(t, evt, err, "marker", "value-b")
+
+		assert.NoError(t, procsA.Close())
+
+		for range 3 {
+			evt, err = procsB.Run(chanEvent())
+			chanRequireMarker(t, evt, err, "marker", "value-b")
+		}
+
+		assert.NoError(t, procsB.Close())
+		for _, id := range []string{idA, idB} {
+			chanAssertAllClosed(t, id)
+			chanAssertLifecycleInvariants(t, id)
+		}
+	})
+}
+
+func TestContractInputConcurrentHarvesters(t *testing.T) {
+	const (
+		workers      = 10
+		harvesters   = 3
+		eventsPerRun = 8
+	)
+
+	info := chanBeatInfo(t)
+
+	sharedID := chanUniqueID("conc-shared-input")
+	const sharedValue = "conc-shared-value"
+
+	ids := []string{sharedID}
+	workerID := make([]string, workers)
+	workerValue := make([]string, workers)
+	for w := range workers {
+		if w%2 == 0 {
+			workerID[w], workerValue[w] = sharedID, sharedValue
+		} else {
+			workerID[w] = chanUniqueID(fmt.Sprintf("conc-input-%d", w))
+			workerValue[w] = fmt.Sprintf("conc-value-%d", w)
+			ids = append(ids, workerID[w])
+		}
+	}
+
+	var wg sync.WaitGroup
+	for w := range workers {
+		id, value := workerID[w], workerValue[w]
+		wg.Go(func() {
+			cfg, err := conf.NewConfigFrom(chanInputCfg(id, value))
+			if err != nil {
+				t.Errorf("worker %d: config: %v", w, err)
+				return
+			}
+			factory := &connectorCapturingFactory{}
+			rfwc := channel.RunnerFactoryWithCommonInputSettings(info, factory)
+			if _, err := rfwc.Create(stubPipeline{}, cfg); err != nil {
+				t.Errorf("worker %d: Create: %v", w, err)
+				return
+			}
+			factory.mu.Lock()
+			connector := factory.connectors[0]
+			factory.mu.Unlock()
+
+			for range harvesters {
+				client, err := connector.ConnectWith(beat.ClientConfig{})
+				if err != nil {
+					t.Errorf("worker %d: ConnectWith: %v", w, err)
+					return
+				}
+				rc, ok := client.(*recordedClient)
+				if !ok || rc.cfg.Processing.Processor == nil {
+					t.Errorf("worker %d: client has no processors", w)
+					return
+				}
+				procs := rc.cfg.Processing.Processor
+
+				for range eventsPerRun {
+					evt, err := procs.Run(chanEvent())
+					if err != nil {
+						t.Errorf("worker %d: Run: %v", w, err)
+						continue
+					}
+					if evt == nil {
+						t.Errorf("worker %d: Run dropped the event", w)
+						continue
+					}
+					got, gerr := evt.GetValue("marker")
+					if gerr != nil {
+						t.Errorf("worker %d: marker missing: %v", w, gerr)
+						continue
+					}
+					if got != value {
+						t.Errorf("worker %d: marker = %v, want %q (event processed with another input's config)",
+							w, got, value)
+					}
+				}
+
+				if err := procs.Close(); err != nil {
+					t.Errorf("worker %d: Close: %v", w, err)
+				}
+				if err := procs.Close(); err != nil {
+					t.Errorf("worker %d: double Close: %v", w, err)
+				}
+				evt, rerr := procs.Run(chanEvent())
+				if err := chanCheckErrOrCorrect(evt, rerr, "marker", value); err != nil {
+					t.Errorf("worker %d: Run after Close: %v", w, err)
+				}
+			}
+		})
+	}
+	wg.Wait()
+
+	for _, id := range ids {
+		chanAssertAllClosed(t, id)
+		chanAssertLifecycleInvariants(t, id)
+	}
+}
diff --git a/filebeat/channel/runner_mock_test.go b/filebeat/channel/runner_mock_test.go
index 0ba1d6fd7bfb..72824742a612 100644
--- a/filebeat/channel/runner_mock_test.go
+++ b/filebeat/channel/runner_mock_test.go
@@ -107,6 +107,15 @@ func (r runnerFactoryMock) Assert(t *testing.T) {
 			}
 		}
 	})
+
+	t.Run("close processor groups", func(t *testing.T) {
+		// Every client must close its processor group: repeated Close must be
+		// tolerated, and unreleased references would keep the processors (and
+		// their background goroutines) alive past the owners' shutdown.
+		for _, c := range r.cfgs {
+			require.NoError(t, c.Processing.Processor.Close())
+		}
+	})
 }
 
 type clientMock struct {
diff --git a/libbeat/processors/contract_test.go b/libbeat/processors/contract_test.go
new file mode 100644
index 000000000000..721f4abcbb89
--- /dev/null
+++ b/libbeat/processors/contract_test.go
@@ -0,0 +1,830 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Black-box contracts for public processor construction and lifecycle behavior.
+
+package processors_test
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/processors"
+	_ "github.com/elastic/beats/v7/libbeat/processors/actions"
+	"github.com/elastic/beats/v7/libbeat/processors/actions/addfields"
+	conf "github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
+	"github.com/elastic/elastic-agent-libs/mapstr"
+	"github.com/elastic/elastic-agent-libs/paths"
+)
+
+const (
+	contractCloserName     = "contract_test_closer"
+	contractPathSetterName = "contract_test_pathsetter"
+	contractFailingName    = "contract_test_failing"
+)
+
+const contractConstructorFailure = "intentional contract_test constructor failure"
+
+var contractIDCounter atomic.Int64
+
+func contractUniqueID(name string) string {
+	return fmt.Sprintf("%s-%d", name, contractIDCounter.Add(1))
+}
+
+type contractProcSnapshot struct {
+	closed             bool
+	closeCount         int
+	runCount           int
+	runsAfterClose     int
+	runsBeforeSetPaths int
+	setPathsCount      int
+	setPathsAfterClose int
+	pathsSeen          []paths.Path
+}
+
+// contractProcState records calls received by an underlying mock processor.
+type contractProcState struct {
+	id        string
+	field     string
+	value     string
+	pathAware bool
+
+	mu                 sync.Mutex
+	closed             bool
+	closeCount         int
+	runCount           int
+	runsAfterClose     int
+	runsBeforeSetPaths int
+	setPathsCount      int
+	setPathsAfterClose int
+	pathsSeen          []paths.Path
+}
+
+func (p *contractProcState) Run(event *beat.Event) (*beat.Event, error) {
+	p.mu.Lock()
+	p.runCount++
+	if p.closed {
+		p.runsAfterClose++
+		p.mu.Unlock()
+		return nil, fmt.Errorf("contract mock processor %q ran after Close", p.id)
+	}
+	if p.pathAware && len(p.pathsSeen) == 0 {
+		p.runsBeforeSetPaths++
+		p.mu.Unlock()
+		return nil, fmt.Errorf("contract mock processor %q ran before SetPaths", p.id)
+	}
+	p.mu.Unlock()
+
+	if _, err := event.PutValue(p.field, p.value); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func (p *contractProcState) Close() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.closeCount++
+	p.closed = true
+	return nil
+}
+
+func (p *contractProcState) String() string {
+	return fmt.Sprintf("contract_test_mock(id=%s)", p.id)
+}
+
+func (p *contractProcState) snapshot() contractProcSnapshot {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return contractProcSnapshot{
+		closed:             p.closed,
+		closeCount:         p.closeCount,
+		runCount:           p.runCount,
+		runsAfterClose:     p.runsAfterClose,
+		runsBeforeSetPaths: p.runsBeforeSetPaths,
+		setPathsCount:      p.setPathsCount,
+		setPathsAfterClose: p.setPathsAfterClose,
+		pathsSeen:          append([]paths.Path(nil), p.pathsSeen...),
+	}
+}
+
+type contractCloserProcessor struct{ *contractProcState }
+
+type contractPathSetterProcessor struct{ *contractProcState }
+
+func (p *contractPathSetterProcessor) SetPaths(pp *paths.Path) error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.closed {
+		p.setPathsAfterClose++
+		return fmt.Errorf("contract mock processor %q: SetPaths after Close", p.id)
+	}
+	p.setPathsCount++
+	if pp != nil {
+		p.pathsSeen = append(p.pathsSeen, *pp)
+	}
+	return nil
+}
+
+type contractTracker struct {
+	mu        sync.Mutex
+	instances map[string][]*contractProcState
+}
+
+var contractInstances = &contractTracker{instances: map[string][]*contractProcState{}}
+
+func (tr *contractTracker) add(st *contractProcState) {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	tr.instances[st.id] = append(tr.instances[st.id], st)
+}
+
+func (tr *contractTracker) get(id string) []*contractProcState {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	return append([]*contractProcState(nil), tr.instances[id]...)
+}
+
+type contractMockConfig struct {
+	ID    string `config:"id"`
+	Field string `config:"field"`
+	Value string `config:"value"`
+}
+
+func newContractProcState(cfg *conf.C, pathAware bool) (*contractProcState, error) {
+	c := contractMockConfig{}
+	if cfg != nil {
+		if err := cfg.Unpack(&c); err != nil {
+			return nil, err
+		}
+	}
+	if c.ID == "" || c.Field == "" {
+		return nil, errors.New("contract_test mock: 'id' and 'field' settings are required")
+	}
+	st := &contractProcState{id: c.ID, field: c.Field, value: c.Value, pathAware: pathAware}
+	contractInstances.add(st)
+	return st, nil
+}
+
+func init() {
+	processors.RegisterPlugin(contractCloserName,
+		func(cfg *conf.C, _ *logp.Logger) (beat.Processor, error) {
+			st, err := newContractProcState(cfg, false)
+			if err != nil {
+				return nil, err
+			}
+			return &contractCloserProcessor{st}, nil
+		})
+	processors.RegisterPlugin(contractPathSetterName,
+		func(cfg *conf.C, _ *logp.Logger) (beat.Processor, error) {
+			st, err := newContractProcState(cfg, true)
+			if err != nil {
+				return nil, err
+			}
+			return &contractPathSetterProcessor{st}, nil
+		})
+	processors.RegisterPlugin(contractFailingName,
+		func(cfg *conf.C, _ *logp.Logger) (beat.Processor, error) {
+			return nil, errors.New(contractConstructorFailure)
+		})
+}
+
+func closerCfg(id, field, value string) mapstr.M {
+	return mapstr.M{
+		contractCloserName: mapstr.M{"id": id, "field": field, "value": value},
+	}
+}
+
+func buildContractProcessors(t testing.TB, raw ...mapstr.M) *processors.Processors {
+	t.Helper()
+	cfg, err := processors.NewPluginConfigFromList(raw)
+	require.NoError(t, err)
+	procs, err := processors.New(cfg, logptest.NewTestingLogger(t, ""))
+	require.NoError(t, err)
+	require.NotNil(t, procs)
+	return procs
+}
+
+func contractEvent() *beat.Event {
+	return &beat.Event{
+		Timestamp: time.Now(),
+		Fields:    mapstr.M{"type": "contract-test"},
+	}
+}
+
+func requireMarker(t *testing.T, evt *beat.Event, err error, field, want string) {
+	t.Helper()
+	require.NoError(t, err)
+	require.NotNil(t, evt)
+	got, gerr := evt.GetValue(field)
+	require.NoError(t, gerr, "field %q missing from event", field)
+	require.Equal(t, want, got)
+}
+
+// A Run racing with or after Close may fail or return a correctly processed
+// event, but it must not silently drop or misprocess the event.
+func checkErrOrCorrect(evt *beat.Event, err error, field, want string) error {
+	if err != nil {
+		return nil //nolint:nilerr // an error result is an acceptable outcome here
+	}
+	if evt == nil {
+		return errors.New("event was dropped without an error")
+	}
+	got, gerr := evt.GetValue(field)
+	if gerr != nil {
+		return fmt.Errorf("run succeeded but marker field %q is missing: %w", field, gerr)
+	}
+	if got != want {
+		return fmt.Errorf("marker field %q has value %v, want %q", field, got, want)
+	}
+	return nil
+}
+
+func assertLifecycleInvariants(t *testing.T, id string) {
+	t.Helper()
+	for i, st := range contractInstances.get(id) {
+		snap := st.snapshot()
+		assert.LessOrEqualf(t, snap.closeCount, 1,
+			"instance %d of %q: underlying Close must be called at most once per constructed instance", i, id)
+		assert.Zerof(t, snap.runsAfterClose,
+			"instance %d of %q: underlying Run must never be called after Close", i, id)
+		assert.Zerof(t, snap.setPathsAfterClose,
+			"instance %d of %q: underlying SetPaths must never be called after Close", i, id)
+	}
+}
+
+func assertAllClosed(t *testing.T, id string) {
+	t.Helper()
+	insts := contractInstances.get(id)
+	require.NotEmptyf(t, insts, "no mock instances were constructed for id %q", id)
+	for i, st := range insts {
+		snap := st.snapshot()
+		assert.Truef(t, snap.closed,
+			"instance %d of %q: closing all owning processor groups must close the processor", i, id)
+	}
+}
+
+func TestContractRegistryConstruction(t *testing.T) {
+	logger := logptest.NewTestingLogger(t, "")
+
+	t.Run("GetConstructor returns a working constructor", func(t *testing.T) {
+		id := contractUniqueID("get-constructor")
+
+		constructor, err := processors.GetConstructor(contractCloserName)
+		require.NoError(t, err)
+		require.NotNil(t, constructor)
+
+		cfg, err := conf.NewConfigFrom(mapstr.M{"id": id, "field": "marker", "value": "via-get-constructor"})
+		require.NoError(t, err)
+
+		p, err := constructor(cfg, logger)
+		require.NoError(t, err)
+		require.NotNil(t, p)
+
+		evt, err := p.Run(contractEvent())
+		requireMarker(t, evt, err, "marker", "via-get-constructor")
+
+		assert.NoError(t, processors.Close(p))
+		assertAllClosed(t, id)
+		assert.NoError(t, processors.Close(p))
+		assertLifecycleInvariants(t, id)
+	})
+
+	t.Run("GetConstructor fails for unknown processor", func(t *testing.T) {
+		_, err := processors.GetConstructor("contract_test_does_not_exist")
+		assert.Error(t, err)
+	})
+
+	t.Run("New fails for unknown processor and names it", func(t *testing.T) {
+		cfg, err := processors.NewPluginConfigFromList([]mapstr.M{
+			{"contract_test_does_not_exist": mapstr.M{}},
+		})
+		require.NoError(t, err)
+
+		procs, err := processors.New(cfg, logger)
+		assert.ErrorContains(t, err, "contract_test_does_not_exist")
+		assert.Nil(t, procs)
+	})
+
+	t.Run("New rejects an entry with multiple actions", func(t *testing.T) {
+		cfg, err := processors.NewPluginConfigFromList([]mapstr.M{
+			{
+				"add_fields":  mapstr.M{"target": "", "fields": mapstr.M{"a": "b"}},
+				"drop_fields": mapstr.M{"fields": []string{"a"}},
+			},
+		})
+		require.NoError(t, err)
+
+		_, err = processors.New(cfg, logger)
+		assert.Error(t, err)
+	})
+
+	t.Run("empty configuration yields a pass-through group", func(t *testing.T) {
+		procs, err := processors.New(nil, logger)
+		require.NoError(t, err)
+		require.NotNil(t, procs)
+
+		evt, err := procs.Run(contractEvent())
+		require.NoError(t, err)
+		require.NotNil(t, evt)
+		typ, err := evt.GetValue("type")
+		require.NoError(t, err)
+		require.Equal(t, "contract-test", typ)
+
+		assert.NoError(t, procs.Close())
+	})
+
+	t.Run("NewList yields an empty runnable group", func(t *testing.T) {
+		procs := processors.NewList(logger)
+		require.NotNil(t, procs)
+
+		evt, err := procs.Run(contractEvent())
+		require.NoError(t, err)
+		require.NotNil(t, evt)
+		assert.NoError(t, procs.Close())
+	})
+}
+
+func TestContractConstructorErrorPropagation(t *testing.T) {
+	logger := logptest.NewTestingLogger(t, "")
+
+	t.Run("direct construction via GetConstructor", func(t *testing.T) {
+		constructor, err := processors.GetConstructor(contractFailingName)
+		require.NoError(t, err)
+
+		cfg, err := conf.NewConfigFrom(mapstr.M{})
+		require.NoError(t, err)
+
+		p, err := constructor(cfg, logger)
+		assert.ErrorContains(t, err, contractConstructorFailure)
+		assert.Nil(t, p)
+	})
+
+	t.Run("construction via New", func(t *testing.T) {
+		cfg, err := processors.NewPluginConfigFromList([]mapstr.M{
+			{contractFailingName: mapstr.M{}},
+		})
+		require.NoError(t, err)
+
+		procs, err := processors.New(cfg, logger)
+		assert.ErrorContains(t, err, contractConstructorFailure)
+		assert.Nil(t, procs)
+	})
+
+	t.Run("one failing constructor fails the whole list", func(t *testing.T) {
+		id := contractUniqueID("fail-mixed")
+		cfg, err := processors.NewPluginConfigFromList([]mapstr.M{
+			closerCfg(id, "marker", "ok"),
+			{contractFailingName: mapstr.M{}},
+		})
+		require.NoError(t, err)
+
+		procs, err := processors.New(cfg, logger)
+		assert.ErrorContains(t, err, contractConstructorFailure)
+		assert.Nil(t, procs)
+	})
+}
+
+func TestContractRunOrderAndConditions(t *testing.T) {
+	t.Run("processors run in configuration order", func(t *testing.T) {
+		procs := buildContractProcessors(t,
+			mapstr.M{"add_fields": mapstr.M{"target": "", "fields": mapstr.M{"a": "first"}}},
+			mapstr.M{"rename": mapstr.M{
+				"fields":        []mapstr.M{{"from": "a", "to": "b"}},
+				"fail_on_error": true,
+			}},
+		)
+		defer procs.Close()
+
+		evt, err := procs.Run(contractEvent())
+		requireMarker(t, evt, err, "b", "first")
+		_, err = evt.GetValue("a")
+		assert.Error(t, err, "field 'a' should have been renamed away")
+	})
+
+	t.Run("when condition gates a registered processor", func(t *testing.T) {
+		id := contractUniqueID("conditional")
+		procs := buildContractProcessors(t, mapstr.M{
+			contractCloserName: mapstr.M{
+				"id": id, "field": "marker", "value": "matched",
+				"when": mapstr.M{"equals": mapstr.M{"flag": "yes"}},
+			},
+		})
+
+		matching := contractEvent()
+		matching.Fields["flag"] = "yes"
+		evt, err := procs.Run(matching)
+		requireMarker(t, evt, err, "marker", "matched")
+
+		other := contractEvent()
+		other.Fields["flag"] = "no"
+		evt, err = procs.Run(other)
+		require.NoError(t, err)
+		require.NotNil(t, evt)
+		_, gerr := evt.GetValue("marker")
+		assert.Error(t, gerr, "processor must not run when the condition does not match")
+
+		assert.NoError(t, procs.Close())
+		assertAllClosed(t, id)
+		assertLifecycleInvariants(t, id)
+	})
+
+	t.Run("drop is propagated", func(t *testing.T) {
+		procs := buildContractProcessors(t, mapstr.M{"drop_event": mapstr.M{}})
+		defer procs.Close()
+
+		evt, err := procs.Run(contractEvent())
+		assert.NoError(t, err)
+		assert.Nil(t, evt, "dropped events must yield a nil event and nil error")
+	})
+}
+
+func TestContractGroupCloseLifecycle(t *testing.T) {
+	id := contractUniqueID("group-close")
+	group := buildContractProcessors(t,
+		closerCfg(id, "marker", "one"),
+		closerCfg(id, "marker2", "two"),
+	)
+
+	evt, err := group.Run(contractEvent())
+	requireMarker(t, evt, err, "marker", "one")
+	requireMarker(t, evt, err, "marker2", "two")
+
+	assert.NoError(t, group.Close())
+	assertAllClosed(t, id)
+	assertLifecycleInvariants(t, id)
+
+	assert.NoError(t, group.Close())
+	assertLifecycleInvariants(t, id)
+
+	for _, p := range group.All() {
+		assert.NoError(t, processors.Close(p))
+	}
+	assertLifecycleInvariants(t, id)
+}
+
+func TestContractRunAfterClose(t *testing.T) {
+	id := contractUniqueID("run-after-close")
+	group := buildContractProcessors(t,
+		closerCfg(id, "marker", "wanted"),
+		mapstr.M{"add_fields": mapstr.M{"target": "", "fields": mapstr.M{"extra": "yes"}}},
+	)
+
+	evt, err := group.Run(contractEvent())
+	requireMarker(t, evt, err, "marker", "wanted")
+
+	assert.NoError(t, group.Close())
+
+	for range 3 {
+		var rerr error
+		var revt *beat.Event
+		assert.NotPanics(t, func() {
+			revt, rerr = group.Run(contractEvent())
+		})
+		assert.NoError(t, checkErrOrCorrect(revt, rerr, "marker", "wanted"))
+	}
+	assertLifecycleInvariants(t, id)
+}
+
+func TestContractSetPathsLifecycle(t *testing.T) {
+	id := contractUniqueID("set-paths")
+	group := buildContractProcessors(t, mapstr.M{
+		contractPathSetterName: mapstr.M{"id": id, "field": "marker", "value": "with-paths"},
+	})
+
+	insts := contractInstances.get(id)
+	require.NotEmpty(t, insts)
+
+	// Pipelines discover PathSetter through group-entry type assertions.
+	var setters []processors.PathSetter
+	for _, p := range group.All() {
+		if ps, ok := p.(processors.PathSetter); ok {
+			setters = append(setters, ps)
+		}
+	}
+	require.NotEmpty(t, setters, "a PathSetter processor must be reachable via the group entries")
+
+	_, err := group.Run(contractEvent())
+	assert.Error(t, err)
+	for i, st := range insts {
+		assert.Zerof(t, st.snapshot().runsBeforeSetPaths,
+			"instance %d: underlying Run must not be called before SetPaths", i)
+	}
+
+	testPaths := &paths.Path{Home: "/contract/home-a"}
+	for _, s := range setters {
+		assert.NoError(t, s.SetPaths(testPaths))
+	}
+	for _, s := range setters {
+		assert.NoError(t, s.SetPaths(testPaths))
+	}
+	for i, st := range insts {
+		snap := st.snapshot()
+		assert.LessOrEqualf(t, snap.setPathsCount, 1,
+			"instance %d: underlying SetPaths must be called at most once", i)
+		if snap.setPathsCount == 1 {
+			assert.Equalf(t, "/contract/home-a", snap.pathsSeen[0].Home,
+				"instance %d: underlying processor must receive the configured paths", i)
+		}
+	}
+
+	conflicting := &paths.Path{Home: "/contract/home-b"}
+	for _, s := range setters {
+		assert.Error(t, s.SetPaths(conflicting))
+	}
+	for i, st := range insts {
+		snap := st.snapshot()
+		for _, seen := range snap.pathsSeen {
+			assert.Equalf(t, "/contract/home-a", seen.Home,
+				"instance %d: underlying processor must never observe conflicting paths", i)
+		}
+	}
+
+	evt, err := group.Run(contractEvent())
+	requireMarker(t, evt, err, "marker", "with-paths")
+
+	assert.NoError(t, group.Close())
+	assertAllClosed(t, id)
+
+	for _, s := range setters {
+		_ = s.SetPaths(testPaths) // returned error is allowed but not required
+	}
+	assertLifecycleInvariants(t, id)
+}
+
+func TestContractCrossPipelineIndependence(t *testing.T) {
+	const numPipelines = 4
+
+	cases := []struct {
+		name         string
+		sharedConfig bool
+	}{
+		{name: "identical config", sharedConfig: true},
+		{name: "different configs", sharedConfig: false},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ids := make([]string, numPipelines)
+			values := make([]string, numPipelines)
+			sharedID := contractUniqueID("xpipe-shared")
+			for i := range numPipelines {
+				if tc.sharedConfig {
+					ids[i] = sharedID
+					values[i] = "shared-value"
+				} else {
+					ids[i] = contractUniqueID(fmt.Sprintf("xpipe-%d", i))
+					values[i] = fmt.Sprintf("value-%d", i)
+				}
+			}
+
+			pipelines := make([]*processors.Processors, numPipelines)
+			for i := range pipelines {
+				pipelines[i] = buildContractProcessors(t,
+					closerCfg(ids[i], "marker", values[i]),
+					mapstr.M{"add_fields": mapstr.M{"target": "", "fields": mapstr.M{"static": "static-value"}}},
+				)
+			}
+
+			for i, pl := range pipelines {
+				evt, err := pl.Run(contractEvent())
+				requireMarker(t, evt, err, "marker", values[i])
+				requireMarker(t, evt, err, "static", "static-value")
+			}
+
+			for i := range pipelines {
+				assert.NoError(t, pipelines[i].Close())
+
+				for j := i + 1; j < numPipelines; j++ {
+					for range 3 {
+						evt, err := pipelines[j].Run(contractEvent())
+						requireMarker(t, evt, err, "marker", values[j])
+						requireMarker(t, evt, err, "static", "static-value")
+					}
+				}
+
+				evt, err := pipelines[i].Run(contractEvent())
+				assert.NoError(t, checkErrOrCorrect(evt, err, "marker", values[i]))
+			}
+
+			for _, id := range ids {
+				assertAllClosed(t, id)
+				assertLifecycleInvariants(t, id)
+			}
+		})
+	}
+}
+
+func TestContractProcessorsSharedAcrossGroups(t *testing.T) {
+	id := contractUniqueID("multi-group")
+	inner := buildContractProcessors(t, closerCfg(id, "marker", "inner-value"))
+
+	outer := processors.NewList(logptest.NewTestingLogger(t, ""))
+	outer.AddProcessors(*inner)
+	outer.AddProcessor(addfields.NewAddFields(mapstr.M{"outer_marker": "yes"}, false, true))
+
+	evt, err := outer.Run(contractEvent())
+	requireMarker(t, evt, err, "marker", "inner-value")
+	requireMarker(t, evt, err, "outer_marker", "yes")
+
+	assert.NoError(t, outer.Close())
+	assert.NoError(t, inner.Close())
+
+	assertAllClosed(t, id)
+	assertLifecycleInvariants(t, id)
+
+	for _, group := range []*processors.Processors{outer, inner} {
+		revt, rerr := group.Run(contractEvent())
+		assert.NoError(t, checkErrOrCorrect(revt, rerr, "marker", "inner-value"))
+	}
+}
+
+func TestContractConcurrentPipelines(t *testing.T) {
+	logger := logptest.NewTestingLogger(t, "")
+
+	const (
+		workers     = 12
+		iterations  = 4
+		eventsPerIt = 8
+	)
+
+	sharedID := contractUniqueID("conc-shared")
+	const sharedValue = "conc-shared-value"
+
+	ids := []string{sharedID}
+	workerID := make([]string, workers)
+	workerValue := make([]string, workers)
+	for w := range workers {
+		if w%2 == 0 {
+			workerID[w], workerValue[w] = sharedID, sharedValue
+		} else {
+			workerID[w] = contractUniqueID(fmt.Sprintf("conc-worker-%d", w))
+			workerValue[w] = fmt.Sprintf("conc-value-%d", w)
+			ids = append(ids, workerID[w])
+		}
+	}
+
+	var wg sync.WaitGroup
+	for w := range workers {
+		id, value := workerID[w], workerValue[w]
+		wg.Go(func() {
+			for range iterations {
+				if _, err := processors.GetConstructor(contractCloserName); err != nil {
+					t.Errorf("worker %d: GetConstructor: %v", w, err)
+					return
+				}
+
+				failCfg, err := processors.NewPluginConfigFromList([]mapstr.M{
+					{contractFailingName: mapstr.M{}},
+				})
+				if err != nil {
+					t.Errorf("worker %d: building failing config: %v", w, err)
+					return
+				}
+				if _, err := processors.New(failCfg, logger); err == nil {
+					t.Errorf("worker %d: expected constructor error, got none", w)
+				}
+
+				cfg, err := processors.NewPluginConfigFromList([]mapstr.M{
+					closerCfg(id, "marker", value),
+					{"add_fields": mapstr.M{"target": "", "fields": mapstr.M{"static": "static-value"}}},
+				})
+				if err != nil {
+					t.Errorf("worker %d: building config: %v", w, err)
+					return
+				}
+				group, err := processors.New(cfg, logger)
+				if err != nil {
+					t.Errorf("worker %d: processors.New: %v", w, err)
+					return
+				}
+
+				for range eventsPerIt {
+					evt, err := group.Run(contractEvent())
+					if err != nil {
+						t.Errorf("worker %d: Run: %v", w, err)
+						continue
+					}
+					if evt == nil {
+						t.Errorf("worker %d: Run dropped the event", w)
+						continue
+					}
+					got, gerr := evt.GetValue("marker")
+					if gerr != nil {
+						t.Errorf("worker %d: marker missing: %v", w, gerr)
+						continue
+					}
+					if got != value {
+						t.Errorf("worker %d: marker = %v, want %q (event processed with another pipeline's config)",
+							w, got, value)
+					}
+				}
+
+				if err := group.Close(); err != nil {
+					t.Errorf("worker %d: Close: %v", w, err)
+				}
+				if err := group.Close(); err != nil {
+					t.Errorf("worker %d: double Close: %v", w, err)
+				}
+				evt, rerr := group.Run(contractEvent())
+				if err := checkErrOrCorrect(evt, rerr, "marker", value); err != nil {
+					t.Errorf("worker %d: Run after Close: %v", w, err)
+				}
+			}
+		})
+	}
+	wg.Wait()
+
+	for _, id := range ids {
+		assertAllClosed(t, id)
+		assertLifecycleInvariants(t, id)
+	}
+}
+
+func TestContractConcurrentRunAndClose(t *testing.T) {
+	id := contractUniqueID("run-vs-close")
+	const value = "run-vs-close-value"
+	group := buildContractProcessors(t, closerCfg(id, "marker", value))
+
+	evt, err := group.Run(contractEvent())
+	requireMarker(t, evt, err, "marker", value)
+
+	const (
+		runners    = 8
+		preRuns    = 20
+		racingRuns = 50
+		postRuns   = 10
+	)
+
+	var preClose sync.WaitGroup
+	preClose.Add(runners)
+	closed := make(chan struct{})
+
+	var wg sync.WaitGroup
+	for r := range runners {
+		wg.Go(func() {
+			// Phase 1: strictly before Close, every run must succeed.
+			for range preRuns {
+				evt, err := group.Run(contractEvent())
+				if err != nil {
+					t.Errorf("runner %d: unexpected error before Close: %v", r, err)
+					continue
+				}
+				if cerr := checkErrOrCorrect(evt, err, "marker", value); cerr != nil {
+					t.Errorf("runner %d: before Close: %v", r, cerr)
+				}
+			}
+			preClose.Done()
+
+			// Phase 2: racing with Close.
+			for range racingRuns {
+				evt, err := group.Run(contractEvent())
+				if cerr := checkErrOrCorrect(evt, err, "marker", value); cerr != nil {
+					t.Errorf("runner %d: racing with Close: %v", r, cerr)
+				}
+			}
+
+			// Phase 3: strictly after Close.
+			<-closed
+			for range postRuns {
+				evt, err := group.Run(contractEvent())
+				if cerr := checkErrOrCorrect(evt, err, "marker", value); cerr != nil {
+					t.Errorf("runner %d: after Close: %v", r, cerr)
+				}
+			}
+		})
+	}
+
+	preClose.Wait()
+	assert.NoError(t, group.Close())
+	assert.NoError(t, group.Close())
+	close(closed)
+	wg.Wait()
+
+	assertAllClosed(t, id)
+	assertLifecycleInvariants(t, id)
+}

From 9aa52416a488b66bcd6897547a1f1103a168fb2e Mon Sep 17 00:00:00 2001
From: Orestis Floros <orestis.floros@elastic.co>
Date: Thu, 2 Jul 2026 09:54:11 +0200
Subject: [PATCH 2/5] Revert "[libbeat/processing] add a new shared processor
 wrapper (#50713)"

This reverts commit 54f7600480f2fe82f991b6bd1db97e61e6d07bd4.
---
 .../add_kubernetes_metadata/kubernetes.go     |   3 +-
 .../kubernetes_concurrent_test.go             |  69 ----
 libbeat/processors/shared/shared.go           | 112 -------
 libbeat/processors/shared/shared_test.go      | 316 ------------------
 4 files changed, 1 insertion(+), 499 deletions(-)
 delete mode 100644 libbeat/processors/shared/shared.go
 delete mode 100644 libbeat/processors/shared/shared_test.go

diff --git a/libbeat/processors/add_kubernetes_metadata/kubernetes.go b/libbeat/processors/add_kubernetes_metadata/kubernetes.go
index c65dc30db5c7..6ffee5cb0a59 100644
--- a/libbeat/processors/add_kubernetes_metadata/kubernetes.go
+++ b/libbeat/processors/add_kubernetes_metadata/kubernetes.go
@@ -41,7 +41,6 @@ import (
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/otel/otelmap"
 	"github.com/elastic/beats/v7/libbeat/processors"
-	"github.com/elastic/beats/v7/libbeat/processors/shared"
 	"github.com/elastic/beats/v7/pkg/autodiscover/kubernetes"
 	"github.com/elastic/beats/v7/pkg/autodiscover/kubernetes/metadata"
 )
@@ -71,7 +70,7 @@ type kubernetesAnnotator struct {
 }
 
 func init() {
-	processors.RegisterPlugin("add_kubernetes_metadata", shared.New(New))
+	processors.RegisterPlugin("add_kubernetes_metadata", New)
 
 	// Register default indexers
 	Indexing.AddIndexer(PodNameIndexerName, NewPodNameIndexer)
diff --git a/libbeat/processors/add_kubernetes_metadata/kubernetes_concurrent_test.go b/libbeat/processors/add_kubernetes_metadata/kubernetes_concurrent_test.go
index d6c2008ba64c..f50bc096a088 100644
--- a/libbeat/processors/add_kubernetes_metadata/kubernetes_concurrent_test.go
+++ b/libbeat/processors/add_kubernetes_metadata/kubernetes_concurrent_test.go
@@ -29,9 +29,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
-	"github.com/elastic/beats/v7/libbeat/processors/shared"
 	"github.com/elastic/elastic-agent-libs/config"
-	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
@@ -318,70 +316,3 @@ func TestAnnotatorRun_CacheMutationDoesNotAffectInFlightEvents(t *testing.T) {
 			"event %d: pod.name must be one of the two valid values", i)
 	}
 }
-
-func TestAnnotatorRun_SharedWrapper_EventIndependenceUnderConcurrency(t *testing.T) {
-	const goroutines = 60
-
-	cfg := config.MustNewConfigFrom(map[string]any{
-		"lookup_fields": []string{"container.id"},
-	})
-	matcher, err := NewFieldMatcher(*cfg, logptest.NewTestingLogger(t, ""))
-	require.NoError(t, err)
-
-	cache := newCache(10 * time.Second)
-	annotator := &kubernetesAnnotator{
-		log:   logptest.NewTestingLogger(t, selector),
-		cache: cache,
-	}
-	setReady(annotator, &Matchers{matchers: []Matcher{matcher}})
-
-	// each goroutine has its own container ID in the cache.
-	for i := range goroutines {
-		cid := fmt.Sprintf("ev-%d", i)
-		cache.set(cid, metaMap(cid))
-	}
-
-	sharedConstructor := shared.New(func(_ *config.C, _ *logp.Logger) (beat.Processor, error) {
-		return annotator, nil
-	})
-	proc, err := sharedConstructor(nil, nil)
-	require.NoError(t, err)
-
-	type result struct {
-		event *beat.Event
-		cid   string
-		err   error
-	}
-	results := make([]result, goroutines)
-	var wg sync.WaitGroup
-	wg.Add(goroutines)
-	for i := range goroutines {
-		cid := fmt.Sprintf("ev-%d", i)
-		go func() {
-			defer wg.Done()
-			out, runErr := proc.Run(containerEvent(cid))
-			results[i] = result{event: out, cid: cid, err: runErr}
-		}()
-	}
-	wg.Wait()
-
-	for i, r := range results {
-		require.NoError(t, r.err, "goroutine %d must not return an error", i)
-		require.NotNil(t, r.event, "goroutine %d must receive a non-nil event", i)
-
-		// container.id must match this goroutine's own cache key.
-		containerRaw, getErr := r.event.Fields.GetValue("container")
-		require.NoError(t, getErr, "goroutine %d: event must have container field", i)
-		container := containerRaw.(mapstr.M) //nolint:errcheck // it's a test
-		assert.Equal(t, r.cid, container["id"],
-			"goroutine %d: container.id must equal its own cache key, not another goroutine's", i)
-
-		// kubernetes.pod.uid must also be specific to this goroutine's entry.
-		k8sRaw, getErr := r.event.Fields.GetValue("kubernetes")
-		require.NoError(t, getErr, "goroutine %d: event must have kubernetes field", i)
-		k8s := k8sRaw.(mapstr.M)     //nolint:errcheck // it's a test
-		pod := k8s["pod"].(mapstr.M) //nolint:errcheck // it's a test
-		assert.Equal(t, "uid-"+r.cid, pod["uid"],
-			"goroutine %d: kubernetes.pod.uid must belong to its own cache entry", i)
-	}
-}
diff --git a/libbeat/processors/shared/shared.go b/libbeat/processors/shared/shared.go
deleted file mode 100644
index a31e4698a056..000000000000
--- a/libbeat/processors/shared/shared.go
+++ /dev/null
@@ -1,112 +0,0 @@
-// Licensed to Elasticsearch B.V. under one or more contributor
-// license agreements. See the NOTICE file distributed with
-// this work for additional information regarding copyright
-// ownership. Elasticsearch B.V. licenses this file to you under
-// the Apache License, Version 2.0 (the "License"); you may
-// not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied.  See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-package shared
-
-import (
-	"sync"
-
-	"go.opentelemetry.io/collector/pdata/pcommon"
-
-	"github.com/elastic/beats/v7/libbeat/beat"
-	"github.com/elastic/beats/v7/libbeat/cfgfile"
-	"github.com/elastic/beats/v7/libbeat/processors"
-	"github.com/elastic/elastic-agent-libs/config"
-	"github.com/elastic/elastic-agent-libs/logp"
-)
-
-type sharedProcessorWithClose struct {
-	beat.Processor
-	hash     uint64
-	refCount int
-
-	sharedProcessors  map[uint64]*sharedProcessorWithClose
-	sharedProcessorMu *sync.Mutex
-}
-
-var _ processors.PdataProcessor = (*sharedPdataProcessorWithClose)(nil)
-
-// sharedPdataProcessorWithClose extends sharedProcessorWithClose with the
-// pdata fast path. It is only created by New when the inner processor
-// implements PdataProcessor.
-type sharedPdataProcessorWithClose struct {
-	*sharedProcessorWithClose
-	pdataProc processors.PdataProcessor
-}
-
-func (p *sharedPdataProcessorWithClose) RunPdata(body pcommon.Map) (bool, error) {
-	return p.pdataProc.RunPdata(body)
-}
-
-// New wraps a processor constructor to return a shared processor.
-// The shared processor will be shared across all processors with the same configuration.
-// The shared processor will be closed when the last processor using it is closed.
-// Warning: the processor is built only once and then reused. Subsequent calls ignore the provided logger.
-// Warning: To be used in conjunction with SafeProcessor. Ref: https://github.com/elastic/beats/blob/5586a1dcc31a748de8805e68c07094d08291fd7c/libbeat/processors/safe_processor.go#L133
-func New(constructor processors.Constructor) processors.Constructor {
-	sharedProcessors := make(map[uint64]*sharedProcessorWithClose)
-	sharedProcessorMu := &sync.Mutex{}
-
-	return func(cfg *config.C, logger *logp.Logger) (beat.Processor, error) {
-		hash := uint64(0)
-		if cfg != nil {
-			var err error
-			hash, err = cfgfile.HashConfig(cfg)
-			if err != nil {
-				return nil, err
-			}
-		}
-
-		sharedProcessorMu.Lock()
-		defer sharedProcessorMu.Unlock()
-		if p, ok := sharedProcessors[hash]; ok {
-			p.refCount++
-			if pp, ok := p.Processor.(processors.PdataProcessor); ok {
-				return &sharedPdataProcessorWithClose{p, pp}, nil
-			}
-			return p, nil
-		}
-
-		proc, err := constructor(cfg, logger)
-		if err != nil {
-			return nil, err
-		}
-		// if the processor does not implement `Closer` it does not need a wrap.
-		// We can extend this in future, if needed.
-		if _, ok := proc.(processors.Closer); !ok {
-			return proc, nil
-		}
-
-		sw := &sharedProcessorWithClose{Processor: proc, hash: hash, sharedProcessors: sharedProcessors, sharedProcessorMu: sharedProcessorMu, refCount: 1}
-		sharedProcessors[hash] = sw
-		if pp, ok := proc.(processors.PdataProcessor); ok {
-			return &sharedPdataProcessorWithClose{sw, pp}, nil
-		}
-		return sw, nil
-	}
-}
-
-func (p *sharedProcessorWithClose) Close() error {
-	p.sharedProcessorMu.Lock()
-	defer p.sharedProcessorMu.Unlock()
-	p.refCount--
-	if p.refCount == 0 {
-		delete(p.sharedProcessors, p.hash)
-		return processors.Close(p.Processor)
-	}
-	return nil
-}
diff --git a/libbeat/processors/shared/shared_test.go b/libbeat/processors/shared/shared_test.go
deleted file mode 100644
index e66da9812e7c..000000000000
--- a/libbeat/processors/shared/shared_test.go
+++ /dev/null
@@ -1,316 +0,0 @@
-// Licensed to Elasticsearch B.V. under one or more contributor
-// license agreements. See the NOTICE file distributed with
-// this work for additional information regarding copyright
-// ownership. Elasticsearch B.V. licenses this file to you under
-// the Apache License, Version 2.0 (the "License"); you may
-// not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied.  See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-package shared_test
-
-import (
-	"fmt"
-	"sync"
-	"sync/atomic"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"github.com/elastic/beats/v7/libbeat/beat"
-	"github.com/elastic/beats/v7/libbeat/processors"
-	"github.com/elastic/beats/v7/libbeat/processors/shared"
-	"github.com/elastic/elastic-agent-libs/config"
-	"github.com/elastic/elastic-agent-libs/logp"
-	"github.com/elastic/elastic-agent-libs/mapstr"
-)
-
-// fakeProcessor records Run/Close call counts.
-type fakeProcessor struct {
-	runCalls   atomic.Int64
-	closeCalls atomic.Int64
-}
-
-func (f *fakeProcessor) Run(event *beat.Event) (*beat.Event, error) {
-	f.runCalls.Add(1)
-	return event, nil
-}
-
-func (f *fakeProcessor) Close() error   { f.closeCalls.Add(1); return nil }
-func (f *fakeProcessor) String() string { return "fake" }
-
-func newFakeConstructor(fp *fakeProcessor) processors.Constructor {
-	return func(_ *config.C, _ *logp.Logger) (beat.Processor, error) { return fp, nil }
-}
-
-func newEvent(id int) *beat.Event {
-	return &beat.Event{Fields: mapstr.M{"id": id}}
-}
-
-func TestNew_SameConfigReturnsSameInstance(t *testing.T) {
-	var created atomic.Int32
-	constructor := shared.New(func(_ *config.C, _ *logp.Logger) (beat.Processor, error) {
-		created.Add(1)
-		return &fakeProcessor{}, nil
-	})
-	cfg := config.MustNewConfigFrom(map[string]any{"key": "value"})
-
-	p1, err := constructor(cfg, nil)
-	require.NoError(t, err)
-	p2, err := constructor(cfg, nil)
-	require.NoError(t, err)
-
-	assert.Equal(t, int32(1), created.Load())
-	assert.Equal(t, p1, p2)
-}
-
-func TestNew_DifferentConfigsReturnDifferentInstances(t *testing.T) {
-	var created atomic.Int32
-	constructor := shared.New(func(_ *config.C, _ *logp.Logger) (beat.Processor, error) {
-		created.Add(1)
-		return &fakeProcessor{}, nil
-	})
-
-	pA, err := constructor(config.MustNewConfigFrom(map[string]any{"key": "A"}), nil)
-	require.NoError(t, err)
-	pB, err := constructor(config.MustNewConfigFrom(map[string]any{"key": "B"}), nil)
-	require.NoError(t, err)
-
-	assert.Equal(t, int32(2), created.Load())
-	assert.NotEqual(t, pA, pB)
-}
-
-func TestNew_NilConfigReturnsSameInstance(t *testing.T) {
-	var created atomic.Int32
-	constructor := shared.New(func(_ *config.C, _ *logp.Logger) (beat.Processor, error) {
-		created.Add(1)
-		return &fakeProcessor{}, nil
-	})
-
-	p1, err := constructor(nil, nil)
-	require.NoError(t, err)
-	p2, err := constructor(nil, nil)
-	require.NoError(t, err)
-
-	assert.Equal(t, int32(1), created.Load())
-	assert.Equal(t, p1, p2)
-}
-
-func TestNew_NonCloserIsNotWrapped(t *testing.T) {
-	type simpleProc struct{ beat.Processor }
-	inner := &simpleProc{}
-	constructor := shared.New(func(_ *config.C, _ *logp.Logger) (beat.Processor, error) { return inner, nil })
-
-	p, err := constructor(nil, nil)
-	require.NoError(t, err)
-	assert.Equal(t, p, inner)
-}
-
-func TestSharedProcessor_CloseUnderlyingWhenLastUserGone(t *testing.T) {
-	tests := []struct {
-		users, closesBeforeStop int
-		expectClosed            bool
-	}{
-		{2, 2, true},
-		{3, 2, false},
-		{3, 3, true},
-	}
-	for _, tc := range tests {
-		t.Run(fmt.Sprintf("users=%d closes=%d", tc.users, tc.closesBeforeStop), func(t *testing.T) {
-			fp := &fakeProcessor{}
-			constructor := shared.New(newFakeConstructor(fp))
-			cfg := config.MustNewConfigFrom(map[string]any{"k": "v"})
-
-			procs := make([]beat.Processor, tc.users)
-			for i := range procs {
-				p, err := constructor(cfg, nil)
-				require.NoError(t, err)
-				procs[i] = p
-			}
-			for i := 0; i < tc.closesBeforeStop; i++ {
-				require.NoError(t, processors.Close(procs[i]))
-			}
-
-			if tc.expectClosed {
-				assert.Equal(t, int64(1), fp.closeCalls.Load(), "expected processor to be closed once")
-			} else {
-				assert.Equal(t, int64(0), fp.closeCalls.Load(), "didn't expect processor to be closed")
-			}
-		})
-	}
-}
-
-func TestNew_ConcurrentSameConfig_NoRace(t *testing.T) {
-	const goroutines = 50
-	var created atomic.Int32
-	constructor := shared.New(func(_ *config.C, _ *logp.Logger) (beat.Processor, error) {
-		created.Add(1)
-		return &fakeProcessor{}, nil
-	})
-	cfg := config.MustNewConfigFrom(map[string]any{"concurrent": true})
-
-	results := make([]beat.Processor, goroutines)
-	var wg sync.WaitGroup
-	wg.Add(goroutines)
-	for i := range goroutines {
-		go func() {
-			defer wg.Done()
-			p, err := constructor(cfg, nil)
-			if err != nil {
-				t.Errorf("goroutine %d: %v", i, err)
-				return
-			}
-			results[i] = p
-		}()
-	}
-	wg.Wait()
-
-	assert.Equal(t, int32(1), created.Load())
-	for i, p := range results {
-		assert.Equalf(t, results[0], p, "goroutine %d got different instance", i)
-	}
-}
-
-func TestSharedProcessor_ConcurrentRun_NoRace(t *testing.T) {
-	const goroutines = 100
-	fp := &fakeProcessor{}
-	proc, err := shared.New(newFakeConstructor(fp))(nil, nil)
-	require.NoError(t, err)
-
-	var wg sync.WaitGroup
-	wg.Add(goroutines)
-	for i := range goroutines {
-		go func() {
-			defer wg.Done()
-			out, err := proc.Run(newEvent(i))
-			if err != nil {
-				t.Errorf("goroutine %d: %v", i, err)
-				return
-			}
-			if out.Fields["id"] != i {
-				t.Errorf("goroutine %d: id mismatch: got %v", i, out.Fields["id"])
-			}
-		}()
-	}
-	wg.Wait()
-	assert.Equal(t, int64(goroutines), fp.runCalls.Load())
-}
-
-func TestSharedProcessor_ConcurrentClose_NoRace(t *testing.T) {
-	const users = 20
-	fp := &fakeProcessor{}
-	constructor := shared.New(newFakeConstructor(fp))
-	cfg := config.MustNewConfigFrom(map[string]any{"cc": true})
-
-	procs := make([]beat.Processor, users)
-	for i := range procs {
-		p, err := constructor(cfg, nil)
-		require.NoError(t, err)
-		procs[i] = p
-	}
-
-	var wg sync.WaitGroup
-	wg.Add(users)
-	for i := range users {
-		go func() {
-			defer wg.Done()
-			if err := processors.Close(procs[i]); err != nil {
-				t.Errorf("goroutine %d: %v", i, err)
-			}
-		}()
-	}
-	wg.Wait()
-	assert.Equal(t, int64(1), fp.closeCalls.Load())
-}
-
-func TestSharedProcessor_ConcurrentRunAndClose_NoRace(t *testing.T) {
-	const runners, closers = 40, 10
-	fp := &fakeProcessor{}
-	constructor := shared.New(newFakeConstructor(fp))
-
-	procs := make([]beat.Processor, closers+1)
-	for i := range procs {
-		p, err := constructor(nil, nil)
-		require.NoError(t, err)
-		procs[i] = p
-	}
-
-	stop := make(chan struct{})
-
-	var runWg sync.WaitGroup
-	runWg.Add(runners)
-	for i := range runners {
-		go func() {
-			defer runWg.Done()
-			for {
-				select {
-				case <-stop:
-					return
-				default:
-					_, _ = procs[0].Run(newEvent(i))
-				}
-			}
-		}()
-	}
-
-	var closeWg sync.WaitGroup
-	closeWg.Add(closers)
-	for i := range closers {
-		go func() {
-			defer closeWg.Done()
-			if err := processors.Close(procs[i+1]); err != nil {
-				t.Errorf("closer %d: %v", i, err)
-			}
-		}()
-	}
-	closeWg.Wait()
-	close(stop)
-	runWg.Wait()
-}
-
-func TestSharedProcessor_RunEventIsolation(t *testing.T) {
-	proc, err := shared.New(func(_ *config.C, _ *logp.Logger) (beat.Processor, error) {
-		return &fakeProcessor{}, nil
-	})(nil, nil)
-	require.NoError(t, err)
-
-	e1, err := proc.Run(newEvent(1))
-	require.NoError(t, err)
-	e2, err := proc.Run(newEvent(2))
-	require.NoError(t, err)
-
-	e1.Fields["injected"] = "mutation"
-	assert.NotContains(t, e2.Fields, "injected")
-	assert.Equal(t, 2, e2.Fields["id"])
-}
-
-func TestSafeProcessorWithSafe(t *testing.T) {
-	fp := &fakeProcessor{}
-	safeProcConstructor := processors.SafeWrap(shared.New(newFakeConstructor(fp)))
-	proc1, err := safeProcConstructor(nil, nil)
-	require.NoError(t, err)
-	proc2, err := safeProcConstructor(nil, nil)
-	require.NoError(t, err)
-
-	require.NoError(t, processors.Close(proc1))
-	// processors should not be closed
-	require.Zero(t, fp.closeCalls.Load())
-
-	// double-close on first instance
-	require.NoError(t, processors.Close(proc1))
-	// processors should not be closed, yet
-	require.Zero(t, fp.closeCalls.Load())
-
-	require.NoError(t, processors.Close(proc2))
-	// processors should be closed now.
-	require.Equal(t, int64(1), fp.closeCalls.Load())
-}

From 79437189d55b425dde214df5699f01b3a836d3f7 Mon Sep 17 00:00:00 2001
From: Vihas Makwana <vihas.makwana@elastic.co>
Date: Thu, 2 Jul 2026 09:56:54 +0200
Subject: [PATCH 3/5] [libbeat/processing] create a unique processor for a
 given config

Cherry-pick of #50586 (squashed). Replaces the approach from #50713
(reverted in the previous commit): instead of a dedicated shared
processor wrapper opted into per-processor, SafeWrap now dedupes
processor instances globally by (name, config hash) with reference
counting, so per-input processors are no longer instantiated per
harvester.

Fixes memory/goroutine blow-up with many harvesters (#50376).

Note: the lifecycle contract tests introduced in the first commit of
this PR fail at this commit (TestContractConcurrentPipelines,
TestContractInputConcurrentHarvesters): closing a group twice releases
another owner's reference and the shared instance is closed under a
live holder. The next commit restores the contract.
---
 filebeat/channel/runner_mock_test.go      |  20 ----
 libbeat/processors/conditionals_test.go   |   2 +-
 libbeat/processors/registry.go            |   2 +-
 libbeat/processors/safe_processor.go      | 114 ++++++++++++++++------
 libbeat/processors/safe_processor_test.go |  97 +++++++++++++++++-
 metricbeat/mb/module/runner_test.go       |   2 +-
 6 files changed, 181 insertions(+), 56 deletions(-)

diff --git a/filebeat/channel/runner_mock_test.go b/filebeat/channel/runner_mock_test.go
index 72824742a612..aa8c27b510cc 100644
--- a/filebeat/channel/runner_mock_test.go
+++ b/filebeat/channel/runner_mock_test.go
@@ -88,26 +88,6 @@ func (r runnerFactoryMock) Assert(t *testing.T) {
 		}
 	})
 
-	t.Run("new processors each time", func(t *testing.T) {
-		var processors []beat.Processor
-		for _, c := range r.cfgs {
-			processors = append(processors, c.Processing.Processor.All()...)
-			defer c.Processing.Processor.Close()
-		}
-
-		require.NotEmptyf(t, processors, "for this test the list of processors cannot be empty")
-
-		for i, p1 := range processors {
-			for j, p2 := range processors {
-				if i == j {
-					continue
-				}
-
-				require.NotSamef(t, p1, p2, "processors must not be re-used")
-			}
-		}
-	})
-
 	t.Run("close processor groups", func(t *testing.T) {
 		// Every client must close its processor group: repeated Close must be
 		// tolerated, and unreleased references would keep the processors (and
diff --git a/libbeat/processors/conditionals_test.go b/libbeat/processors/conditionals_test.go
index f85cf9a6ec00..a881d89cbe1e 100644
--- a/libbeat/processors/conditionals_test.go
+++ b/libbeat/processors/conditionals_test.go
@@ -399,7 +399,7 @@ func requireImplements[T any](t *testing.T, v any) T {
 
 func TestConditionErrorClosesProcessor(t *testing.T) {
 	cons, p := newMockCloserConstructor()
-	wrapped := NewConditional(SafeWrap(cons))
+	wrapped := NewConditional(SafeWrap("test-cond-close-on-error", cons))
 
 	cfg, err := conf.NewConfigFrom(map[string]any{"when": map[string]any{}})
 	require.NoError(t, err)
diff --git a/libbeat/processors/registry.go b/libbeat/processors/registry.go
index 074022c61b7e..f872f4ddd305 100644
--- a/libbeat/processors/registry.go
+++ b/libbeat/processors/registry.go
@@ -28,7 +28,7 @@ type Constructor func(config *config.C, logger *logp.Logger) (beat.Processor, er
 var registry = NewNamespace()
 
 func RegisterPlugin(name string, constructor Constructor) {
-	err := registry.Register(name, SafeWrap(constructor))
+	err := registry.Register(name, SafeWrap(name, constructor))
 	if err != nil {
 		panic(err)
 	}
diff --git a/libbeat/processors/safe_processor.go b/libbeat/processors/safe_processor.go
index 8b45a32f49df..88131f134784 100644
--- a/libbeat/processors/safe_processor.go
+++ b/libbeat/processors/safe_processor.go
@@ -25,6 +25,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/cfgfile"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/paths"
@@ -47,6 +48,9 @@ const (
 	stateClosed
 )
 
+var sharedProcessorMu sync.Mutex
+var sharedProcessors map[string]map[uint64]beat.Processor = make(map[string]map[uint64]beat.Processor)
+
 // SafeProcessor wraps a beat.Processor to provide thread-safe state management.
 // It ensures SetPaths is called only once and prevents Run after Close.
 // Use safeProcessorWithClose for processors that also implement Closer.
@@ -56,6 +60,10 @@ type SafeProcessor struct {
 	mu    sync.RWMutex
 	state state
 	paths *paths.Path
+
+	refCount int
+	hash     uint64
+	name     string
 }
 
 // safeProcessorWithClose extends SafeProcessor to also handle Close.
@@ -110,16 +118,31 @@ func (p *safePdataProcessorWithClose) RunPdata(body pcommon.Map) (bool, error) {
 
 // Close makes sure the underlying `Close` function is called only once.
 func (p *safeProcessorWithClose) Close() (err error) {
+	sharedProcessorMu.Lock()
+	defer sharedProcessorMu.Unlock()
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	if p.state != stateClosed {
+	if p.state == stateClosed {
+		logp.L().Warnf("tried to close already closed %q processor", p.String())
+		return
+	}
+	p.refCount--
+	if p.refCount == 0 {
+		p.deleteFromSharedMap()
 		p.state = stateClosed
 		return Close(p.Processor)
 	}
-	logp.L().Warnf("tried to close already closed %q processor", p.String())
 	return nil
 }
 
+// NOTE: To be called while holding the sharedProcessorMu lock to ensure.
+func (p *SafeProcessor) deleteFromSharedMap() {
+	if _, ok := sharedProcessors[p.name]; !ok {
+		return
+	}
+	delete(sharedProcessors[p.name], p.hash)
+}
+
 // SetPaths delegates to the underlying processor if it implements PathSetter.
 // Returns ErrPathsAlreadySet if called more than once, or ErrSetPathsOnClosed
 // if the processor has been closed.
@@ -160,38 +183,73 @@ func (p *SafeProcessor) SetPaths(paths *paths.Path) error {
 //
 // Without SafeWrap, processors must handle these cases manually using sync.Once
 // or similar mechanisms. SafeWrap is automatically applied by RegisterPlugin.
-//
-// When the inner processor implements both Closer and PdataProcessor, SafeWrap
-// returns a safePdataProcessorWithClose so that the pdata fast path in
-// buildPdataProcs can detect the capability without a fallback round-trip.
-func SafeWrap(constructor Constructor) Constructor {
-	return func(config *config.C, log *logp.Logger) (beat.Processor, error) {
-		processor, err := constructor(config, log)
+func SafeWrap(name string, constructor Constructor) Constructor {
+	return func(cfg *config.C, log *logp.Logger) (beat.Processor, error) {
+		sharedProcessorMu.Lock()
+		defer sharedProcessorMu.Unlock()
+		hash, err := cfgfile.HashConfig(cfg)
+		if cfg == nil {
+			err = nil
+			hash = 0
+		}
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to hash processor config: %w", err)
 		}
-		_, isCloser := processor.(Closer)
-		pdataProc, isPdata := processor.(PdataProcessor)
-
-		// if the processor does not implement `Closer` it does not need a wrap
-		if !isCloser {
-			// if SetPaths is implemented, ensure single call of SetPaths
-			if _, ok := processor.(PathSetter); ok {
-				return &SafeProcessor{Processor: processor}, nil
+		if p, ok := sharedProcessors[name][hash]; ok {
+			switch proc := p.(type) {
+			case *safePdataProcessorWithClose:
+				proc.mu.Lock()
+				defer proc.mu.Unlock()
+				proc.refCount++
+				return proc, nil
+			case *safeProcessorWithClose:
+				proc.mu.Lock()
+				defer proc.mu.Unlock()
+				proc.refCount++
+				return proc, nil
+			case *SafeProcessor:
+				proc.mu.Lock()
+				defer proc.mu.Unlock()
+				proc.refCount++
+				return proc, nil
 			}
-			return processor, nil
+			return p, nil
 		}
+		safeProcessor, err := newSafeProcessor(log, constructor, cfg, hash, name)
+		if err != nil {
+			return nil, err
+		}
+		if sharedProcessors[name] == nil {
+			sharedProcessors[name] = make(map[uint64]beat.Processor)
+		}
+		sharedProcessors[name][hash] = safeProcessor
+		return safeProcessor, nil
+	}
+}
 
-		if isPdata {
-			return &safePdataProcessorWithClose{
-				safeProcessorWithClose: safeProcessorWithClose{
-					SafeProcessor: SafeProcessor{Processor: processor},
-				},
-				pdataProc: pdataProc,
-			}, nil
+func newSafeProcessor(log *logp.Logger, constructor Constructor, config *config.C, hash uint64, name string) (beat.Processor, error) {
+	processor, err := constructor(config, log)
+	if err != nil {
+		return nil, err
+	}
+	// if the processor does not implement `Closer` it does not need a wrap
+	if _, ok := processor.(Closer); !ok {
+		// if SetPaths is implemented, ensure single call of SetPaths
+		if _, ok = processor.(PathSetter); ok {
+			return &SafeProcessor{Processor: processor, hash: hash, name: name, refCount: 1}, nil
 		}
-		return &safeProcessorWithClose{
-			SafeProcessor: SafeProcessor{Processor: processor},
+		return processor, nil
+	}
+
+	if pdataProc, ok := processor.(PdataProcessor); ok {
+		return &safePdataProcessorWithClose{
+			safeProcessorWithClose: safeProcessorWithClose{
+				SafeProcessor: SafeProcessor{Processor: processor, hash: hash, name: name, refCount: 1},
+			},
+			pdataProc: pdataProc,
 		}, nil
 	}
+	return &safeProcessorWithClose{
+		SafeProcessor: SafeProcessor{Processor: processor, hash: hash, name: name, refCount: 1},
+	}, nil
 }
diff --git a/libbeat/processors/safe_processor_test.go b/libbeat/processors/safe_processor_test.go
index 04a286df2dd3..f5d02fba4f79 100644
--- a/libbeat/processors/safe_processor_test.go
+++ b/libbeat/processors/safe_processor_test.go
@@ -107,7 +107,7 @@ func mockCloserConstructor(config *config.C, log *logp.Logger) (beat.Processor,
 func TestSafeWrap(t *testing.T) {
 	t.Run("does not wrap a non-closer processor", func(t *testing.T) {
 		nonCloser := mockConstructor
-		wrappedNonCloser := SafeWrap(nonCloser)
+		wrappedNonCloser := SafeWrap("non-closer processor", nonCloser)
 		wp, err := wrappedNonCloser(nil, nil)
 		require.NoError(t, err)
 		assert.IsType(t, &mockProcessor{}, wp)
@@ -116,7 +116,7 @@ func TestSafeWrap(t *testing.T) {
 
 	t.Run("wraps a closer processor", func(t *testing.T) {
 		closer := mockCloserConstructor
-		wrappedCloser := SafeWrap(closer)
+		wrappedCloser := SafeWrap("closer processor", closer)
 		wcp, err := wrappedCloser(nil, nil)
 		require.NoError(t, err)
 		assert.IsType(t, &safeProcessorWithClose{}, wcp)
@@ -131,7 +131,7 @@ func TestSafeProcessor(t *testing.T) {
 		err error
 	)
 	t.Run("creates a wrapped processor", func(t *testing.T) {
-		sw := SafeWrap(cons)
+		sw := SafeWrap("", cons)
 		sp, err = sw(nil, nil)
 		require.NoError(t, err)
 	})
@@ -177,7 +177,7 @@ func TestSafeProcessorSetPathsClose(t *testing.T) {
 		err error
 	)
 	t.Run("creates a wrapped processor", func(t *testing.T) {
-		sw := SafeWrap(cons)
+		sw := SafeWrap("", cons)
 		bp, err = sw(nil, nil)
 		require.NoError(t, err)
 		assert.Equal(t, 0, p.setPathsCount)
@@ -253,6 +253,93 @@ func TestSafeProcessorSetPathsClose(t *testing.T) {
 	})
 }
 
+func TestSafeWrapSharedInstanceByNameAndHash(t *testing.T) {
+	cons, p := newMockCloserConstructor()
+	sw := SafeWrap("test-shared-instance", cons)
+
+	proc1, err := sw(nil, nil)
+	require.NoError(t, err, "first SafeWrap call should succeed")
+
+	proc2, err := sw(nil, nil)
+	require.NoError(t, err, "second SafeWrap call should succeed")
+
+	assert.Same(t, proc1, proc2, "same name+config should return the same processor pointer")
+
+	_, err = proc1.Run(nil)
+	require.NoError(t, err, "Run via proc1 should succeed")
+	assert.Equal(t, 1, p.runCount, "run should be reflected in the underlying mock")
+
+	require.NoError(t, Close(proc1), "first Close should not error")
+	assert.Equal(t, 0, p.closeCount, "underlying processor should not be closed while a ref remains")
+
+	require.NoError(t, Close(proc2), "second Close should not error")
+	assert.Equal(t, 1, p.closeCount, "underlying processor should be closed once all refs are released")
+}
+
+func TestSafeWrapDifferentNamesNotShared(t *testing.T) {
+	cons1, p1 := newMockCloserConstructor()
+	cons2, p2 := newMockCloserConstructor()
+
+	proc1, err := SafeWrap("test-name-a", cons1)(nil, nil)
+	require.NoError(t, err, "SafeWrap for name-a should succeed")
+
+	proc2, err := SafeWrap("test-name-b", cons2)(nil, nil)
+	require.NoError(t, err, "SafeWrap for name-b should succeed")
+
+	assert.NotSame(t, proc1, proc2, "different names must produce separate processor instances")
+
+	_, err = proc1.Run(nil)
+	require.NoError(t, err, "Run on proc1 should succeed")
+	assert.Equal(t, 1, p1.runCount, "run should only increment p1.runCount")
+	assert.Equal(t, 0, p2.runCount, "p2.runCount must remain 0")
+
+	require.NoError(t, Close(proc1))
+	require.NoError(t, Close(proc2))
+	assert.Equal(t, 1, p1.closeCount, "p1 should be closed exactly once")
+	assert.Equal(t, 1, p2.closeCount, "p2 should be closed exactly once")
+}
+
+func TestSafeWrapRefCountingPreventsEarlyClose(t *testing.T) {
+	cons, p := newMockCloserConstructor()
+	sw := SafeWrap("test-refcount", cons)
+
+	proc1, err := sw(nil, nil)
+	require.NoError(t, err)
+	proc2, err := sw(nil, nil)
+	require.NoError(t, err)
+	proc3, err := sw(nil, nil)
+	require.NoError(t, err)
+
+	require.NoError(t, Close(proc1))
+	assert.Equal(t, 0, p.closeCount, "should not close after first of three Close calls")
+
+	require.NoError(t, Close(proc2))
+	assert.Equal(t, 0, p.closeCount, "should not close after second of three Close calls")
+
+	require.NoError(t, Close(proc3))
+	assert.Equal(t, 1, p.closeCount, "should close exactly once after last ref is released")
+}
+
+func TestSafeWrapNewInstanceAfterAllRefsClosed(t *testing.T) {
+	sw := SafeWrap("test-recreate-after-close", mockCloserConstructor)
+
+	proc1, err := sw(nil, nil)
+	require.NoError(t, err, "initial SafeWrap call should succeed")
+
+	require.NoError(t, Close(proc1), "closing the only reference should succeed")
+
+	// Entry is removed from sharedProcessors; next call must build a fresh instance.
+	proc2, err := sw(nil, nil)
+	require.NoError(t, err, "SafeWrap after full close should succeed")
+
+	assert.NotSame(t, proc1, proc2, "a new instance must be created after all refs are closed")
+
+	_, err = proc2.Run(nil)
+	assert.NoError(t, err, "newly created processor must be runnable")
+
+	require.NoError(t, Close(proc2))
+}
+
 func TestSafeProcessorSetPaths(t *testing.T) {
 	cons, p := newMockPathSetterProcessor()
 	var (
@@ -261,7 +348,7 @@ func TestSafeProcessorSetPaths(t *testing.T) {
 		err error
 	)
 	t.Run("creates a wrapped processor", func(t *testing.T) {
-		sw := SafeWrap(cons)
+		sw := SafeWrap("", cons)
 		bp, err = sw(nil, nil)
 		require.NoError(t, err)
 		assert.Equal(t, 0, p.setPathsCount)
diff --git a/metricbeat/mb/module/runner_test.go b/metricbeat/mb/module/runner_test.go
index 8d620a3f0196..f0ded3436de2 100644
--- a/metricbeat/mb/module/runner_test.go
+++ b/metricbeat/mb/module/runner_test.go
@@ -138,7 +138,7 @@ func newPubClientFactory() (*pubtest.ChanClient, func() beat.Client) {
 
 func TestRunnerStop_ClientClosedAfterPublishGoroutine(t *testing.T) {
 	// Wrap a no-op processor with SafeWrap so it becomes a safeProcessorWithClose.
-	proc, err := processors.SafeWrap(func(_ *conf.C, _ *logp.Logger) (beat.Processor, error) {
+	proc, err := processors.SafeWrap("runner-test-noop-closer", func(_ *conf.C, _ *logp.Logger) (beat.Processor, error) {
 		return noopCloser{}, nil
 	})(nil, nil)
 	require.NoError(t, err)

From 9ccee8ece22ce5062ac9bd2ce35ff1a408b55c27 Mon Sep 17 00:00:00 2001
From: Orestis Floros <orestis.floros@elastic.co>
Date: Fri, 24 Jul 2026 11:32:47 +0200
Subject: [PATCH 4/5] [libbeat/processing] harden processor sharing: per-owner
 handles, opt-out, singleflight construction

Rework of the sharing mechanism from the previous commit, addressing
findings from adversarial review, contract tests and benchmarks:

- Each constructor call returns its own handle to the shared instance.
  Close is idempotent per handle (the documented SafeWrap contract: one
  instance can be reachable from multiple processor groups that each
  close it): a duplicate Close can no longer release another owner's
  reference and close the shared instance under a live holder.
- Processors are shared by default with two exceptions: PathSetter
  implementations (paths are owner-specific; in otel mode each beat
  receiver sets its own paths and pointer-compared paths would fail the
  second receiver) and Unshareable implementations, a new marker
  interface for processors whose semantics change when shared
  (rate_limit: one shared token bucket would divide the configured
  per-input limit by the number of owners).
- The real constructor runs outside the global lock via per-key
  singleflight: a blocking constructor (add_kubernetes_metadata with
  wait_for_metadata waits for the API server, potentially forever with
  wait_for_metadata_timeout: 0) no longer stalls all processor
  construction and teardown process-wide, and a construction panic
  cannot leave concurrent waiters blocked. The underlying Close (which
  drains in-flight Runs) also runs outside the global lock.
- Shared handles preserve the pdata fast path: a shared processor that
  implements PdataProcessor is returned as a handle that forwards
  RunPdata, and unshared Closer+PdataProcessor instances keep the
  safePdataProcessorWithClose wrapping.
- The eviction of the last reference removes the cache entry, bounding
  the shared map under autodiscover config churn.

Includes a concurrency churn stress test (same-config construct/Run/
Close churn racing a steady different-config holder of the same
processor name) designed for script/stresstest.sh; 1000+ runs with 0
failures under -race.
---
 ...y-configured-processors-across-inputs.yaml |  46 +++
 filebeat/channel/contract_test.go             |   3 -
 filebeat/channel/runner_mock_test.go          |   4 +-
 libbeat/processors/contract_test.go           |   6 -
 libbeat/processors/processor.go               |   9 +-
 libbeat/processors/ratelimit/rate_limit.go    |   5 +
 libbeat/processors/safe_processor.go          | 339 +++++++++++------
 libbeat/processors/safe_processor_test.go     | 354 +++++++++++++++++-
 .../processors/shared_churn_stress_test.go    | 151 ++++++++
 9 files changed, 789 insertions(+), 128 deletions(-)
 create mode 100644 changelog/fragments/1783069200-share-identically-configured-processors-across-inputs.yaml
 create mode 100644 libbeat/processors/shared_churn_stress_test.go

diff --git a/changelog/fragments/1783069200-share-identically-configured-processors-across-inputs.yaml b/changelog/fragments/1783069200-share-identically-configured-processors-across-inputs.yaml
new file mode 100644
index 000000000000..c206f387eca3
--- /dev/null
+++ b/changelog/fragments/1783069200-share-identically-configured-processors-across-inputs.yaml
@@ -0,0 +1,46 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: fix excessive memory and goroutine usage when many inputs use the same processor configuration
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+description: >
+  Memory usage and goroutine count no longer grow with the number of log files
+  when inputs use processors such as add_kubernetes_metadata, for example
+  filestream collecting container logs. Previously each harvester ran its own
+  copy of the configured processors, which on hosts with many log files could
+  consume hundreds of megabytes and cause OOM restarts.
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: all
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/beats/issues/50376
diff --git a/filebeat/channel/contract_test.go b/filebeat/channel/contract_test.go
index 92df1aa774a4..f24a5ad8a98b 100644
--- a/filebeat/channel/contract_test.go
+++ b/filebeat/channel/contract_test.go
@@ -15,9 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-// Black-box contracts for per-client processors constructed through
-// RunnerFactoryWithCommonInputSettings.
-
 package channel_test
 
 import (
diff --git a/filebeat/channel/runner_mock_test.go b/filebeat/channel/runner_mock_test.go
index aa8c27b510cc..bc225cbcfc85 100644
--- a/filebeat/channel/runner_mock_test.go
+++ b/filebeat/channel/runner_mock_test.go
@@ -89,9 +89,7 @@ func (r runnerFactoryMock) Assert(t *testing.T) {
 	})
 
 	t.Run("close processor groups", func(t *testing.T) {
-		// Every client must close its processor group: repeated Close must be
-		// tolerated, and unreleased references would keep the processors (and
-		// their background goroutines) alive past the owners' shutdown.
+		// Close every processor group to release shared references.
 		for _, c := range r.cfgs {
 			require.NoError(t, c.Processing.Processor.Close())
 		}
diff --git a/libbeat/processors/contract_test.go b/libbeat/processors/contract_test.go
index 721f4abcbb89..0151427e1158 100644
--- a/libbeat/processors/contract_test.go
+++ b/libbeat/processors/contract_test.go
@@ -15,8 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-// Black-box contracts for public processor construction and lifecycle behavior.
-
 package processors_test
 
 import (
@@ -58,7 +56,6 @@ func contractUniqueID(name string) string {
 type contractProcSnapshot struct {
 	closed             bool
 	closeCount         int
-	runCount           int
 	runsAfterClose     int
 	runsBeforeSetPaths int
 	setPathsCount      int
@@ -76,7 +73,6 @@ type contractProcState struct {
 	mu                 sync.Mutex
 	closed             bool
 	closeCount         int
-	runCount           int
 	runsAfterClose     int
 	runsBeforeSetPaths int
 	setPathsCount      int
@@ -86,7 +82,6 @@ type contractProcState struct {
 
 func (p *contractProcState) Run(event *beat.Event) (*beat.Event, error) {
 	p.mu.Lock()
-	p.runCount++
 	if p.closed {
 		p.runsAfterClose++
 		p.mu.Unlock()
@@ -123,7 +118,6 @@ func (p *contractProcState) snapshot() contractProcSnapshot {
 	return contractProcSnapshot{
 		closed:             p.closed,
 		closeCount:         p.closeCount,
-		runCount:           p.runCount,
 		runsAfterClose:     p.runsAfterClose,
 		runsBeforeSetPaths: p.runsBeforeSetPaths,
 		setPathsCount:      p.setPathsCount,
diff --git a/libbeat/processors/processor.go b/libbeat/processors/processor.go
index 0341fecb56ad..a88e437b698a 100644
--- a/libbeat/processors/processor.go
+++ b/libbeat/processors/processor.go
@@ -30,8 +30,6 @@ import (
 	"github.com/elastic/elastic-agent-libs/paths"
 )
 
-const logName = "processors"
-
 // Processors is
 type Processors struct {
 	List []beat.Processor
@@ -53,6 +51,13 @@ type PathSetter interface {
 	SetPaths(*paths.Path) error
 }
 
+// Unshareable opts a processor out of sharing with other owners using the same
+// configuration. Implement it when sharing would change per-owner semantics.
+// Its marker method is never called.
+type Unshareable interface {
+	Unshareable()
+}
+
 // PdataProcessor is an optional interface that beat processors can implement to
 // operate directly on a pcommon.Map, avoiding the round-trip conversion to/from
 // mapstr.M. When all processors in a chain implement this interface, the
diff --git a/libbeat/processors/ratelimit/rate_limit.go b/libbeat/processors/ratelimit/rate_limit.go
index fa7a792abfd5..d96b1cb58506 100644
--- a/libbeat/processors/ratelimit/rate_limit.go
+++ b/libbeat/processors/ratelimit/rate_limit.go
@@ -117,6 +117,11 @@ func (p *rateLimit) Run(event *beat.Event) (*beat.Event, error) {
 	return nil, nil
 }
 
+// Unshareable opts rate_limit out of process-wide processor sharing: the
+// algorithm keeps a token bucket, so sharing one instance across owners would
+// enforce a single global limit instead of an independent per-owner limit.
+func (p *rateLimit) Unshareable() {}
+
 func (p *rateLimit) String() string {
 	return fmt.Sprintf(
 		"%v=[limit=[%v],fields=[%v],algorithm=[%v]]",
diff --git a/libbeat/processors/safe_processor.go b/libbeat/processors/safe_processor.go
index 88131f134784..f9491c52b451 100644
--- a/libbeat/processors/safe_processor.go
+++ b/libbeat/processors/safe_processor.go
@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
 
@@ -48,9 +49,6 @@ const (
 	stateClosed
 )
 
-var sharedProcessorMu sync.Mutex
-var sharedProcessors map[string]map[uint64]beat.Processor = make(map[string]map[uint64]beat.Processor)
-
 // SafeProcessor wraps a beat.Processor to provide thread-safe state management.
 // It ensures SetPaths is called only once and prevents Run after Close.
 // Use safeProcessorWithClose for processors that also implement Closer.
@@ -60,10 +58,6 @@ type SafeProcessor struct {
 	mu    sync.RWMutex
 	state state
 	paths *paths.Path
-
-	refCount int
-	hash     uint64
-	name     string
 }
 
 // safeProcessorWithClose extends SafeProcessor to also handle Close.
@@ -98,51 +92,6 @@ func (p *SafeProcessor) Run(event *beat.Event) (*beat.Event, error) {
 	return p.Processor.Run(event)
 }
 
-// RunPdata delegates to the inner processor's RunPdata. The inner is guaranteed
-// to implement PdataProcessor because SafeWrap only creates this type when the
-// inner satisfies both Closer and PdataProcessor.
-func (p *safePdataProcessorWithClose) RunPdata(body pcommon.Map) (bool, error) {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	switch p.state {
-	case stateClosed:
-		return false, ErrClosed
-	case stateInit:
-		if _, ok := p.Processor.(PathSetter); ok {
-			return false, ErrPathsNotSet
-		}
-	default: // proceed
-	}
-	return p.pdataProc.RunPdata(body)
-}
-
-// Close makes sure the underlying `Close` function is called only once.
-func (p *safeProcessorWithClose) Close() (err error) {
-	sharedProcessorMu.Lock()
-	defer sharedProcessorMu.Unlock()
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	if p.state == stateClosed {
-		logp.L().Warnf("tried to close already closed %q processor", p.String())
-		return
-	}
-	p.refCount--
-	if p.refCount == 0 {
-		p.deleteFromSharedMap()
-		p.state = stateClosed
-		return Close(p.Processor)
-	}
-	return nil
-}
-
-// NOTE: To be called while holding the sharedProcessorMu lock to ensure.
-func (p *SafeProcessor) deleteFromSharedMap() {
-	if _, ok := sharedProcessors[p.name]; !ok {
-		return
-	}
-	delete(sharedProcessors[p.name], p.hash)
-}
-
 // SetPaths delegates to the underlying processor if it implements PathSetter.
 // Returns ErrPathsAlreadySet if called more than once, or ErrSetPathsOnClosed
 // if the processor has been closed.
@@ -170,86 +119,264 @@ func (p *SafeProcessor) SetPaths(paths *paths.Path) error {
 	return fmt.Errorf("unknown state: %d", p.state)
 }
 
+// Close makes sure the underlying `Close` function is called only once.
+func (p *safeProcessorWithClose) Close() (err error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.state != stateClosed {
+		p.state = stateClosed
+		return Close(p.Processor)
+	}
+	return nil
+}
+
+// RunPdata delegates to the inner processor's RunPdata. The inner is guaranteed
+// to implement PdataProcessor because SafeWrap only creates this type when the
+// inner satisfies both Closer and PdataProcessor.
+func (p *safePdataProcessorWithClose) RunPdata(body pcommon.Map) (bool, error) {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	switch p.state {
+	case stateClosed:
+		return false, ErrClosed
+	case stateInit:
+		if _, ok := p.Processor.(PathSetter); ok {
+			return false, ErrPathsNotSet
+		}
+	default: // proceed
+	}
+	return p.pdataProc.RunPdata(body)
+}
+
+// wrapUnshared preserves SafeWrap's Close and SetPaths guarantees for processor
+// instances that cannot be shared.
+func wrapUnshared(processor beat.Processor) beat.Processor {
+	if _, ok := processor.(Closer); ok {
+		if pdataProc, ok := processor.(PdataProcessor); ok {
+			return &safePdataProcessorWithClose{
+				safeProcessorWithClose: safeProcessorWithClose{
+					SafeProcessor: SafeProcessor{Processor: processor},
+				},
+				pdataProc: pdataProc,
+			}
+		}
+		return &safeProcessorWithClose{
+			SafeProcessor: SafeProcessor{Processor: processor},
+		}
+	}
+	if _, ok := processor.(PathSetter); ok {
+		return &SafeProcessor{Processor: processor}
+	}
+	return processor
+}
+
+type sharedProcessorKey struct {
+	name       string
+	configHash uint64
+}
+
+// sharedCore is the process-wide reference-counted instance for one processor key.
+type sharedCore struct {
+	safeProcessorWithClose
+
+	key sharedProcessorKey
+
+	// refCount includes live and pending handles; guarded by sharedProcessorMu.
+	refCount int
+	// constructionErr is published before ready closes.
+	constructionErr error
+	// ready is closed when construction has finished, successfully or not.
+	ready chan struct{}
+}
+
+// sharedHandle releases one core reference at most once.
+type sharedHandle struct {
+	*sharedCore
+	closed atomic.Bool
+}
+
+// sharedPdataHandle preserves PdataProcessor on a shared handle.
+type sharedPdataHandle struct {
+	*sharedHandle
+	pdata PdataProcessor
+}
+
+var sharedProcessorMu sync.Mutex
+var sharedProcessors = make(map[sharedProcessorKey]*sharedCore)
+
 // SafeWrap wraps a processor constructor to handle common edge cases:
 //
 //   - Multiple Close calls: Each processor might end up in multiple processor
 //     groups. Every group has its own Close that calls Close on each processor,
-//     leading to multiple Close calls on the same processor.
+//     leading to multiple Close calls on the same processor. Close is
+//     idempotent per constructed handle.
 //
 //   - Multiple SetPaths calls: The wrapper ensures SetPaths is called at most once.
 //
 //   - Close before/during SetPaths: Prevents initialization after shutdown and
 //     protects against race conditions between SetPaths and Close.
 //
+// Processors with the same name and configuration share one process-wide
+// instance unless they implement PathSetter or Unshareable. Handles reference
+// count the shared instance, which is closed after its last handle.
+//
 // Without SafeWrap, processors must handle these cases manually using sync.Once
 // or similar mechanisms. SafeWrap is automatically applied by RegisterPlugin.
 func SafeWrap(name string, constructor Constructor) Constructor {
 	return func(cfg *config.C, log *logp.Logger) (beat.Processor, error) {
-		sharedProcessorMu.Lock()
-		defer sharedProcessorMu.Unlock()
-		hash, err := cfgfile.HashConfig(cfg)
-		if cfg == nil {
-			err = nil
-			hash = 0
-		}
+		configHash, err := hashProcessorConfig(cfg)
 		if err != nil {
 			return nil, fmt.Errorf("failed to hash processor config: %w", err)
 		}
-		if p, ok := sharedProcessors[name][hash]; ok {
-			switch proc := p.(type) {
-			case *safePdataProcessorWithClose:
-				proc.mu.Lock()
-				defer proc.mu.Unlock()
-				proc.refCount++
-				return proc, nil
-			case *safeProcessorWithClose:
-				proc.mu.Lock()
-				defer proc.mu.Unlock()
-				proc.refCount++
-				return proc, nil
-			case *SafeProcessor:
-				proc.mu.Lock()
-				defer proc.mu.Unlock()
-				proc.refCount++
-				return proc, nil
-			}
-			return p, nil
-		}
-		safeProcessor, err := newSafeProcessor(log, constructor, cfg, hash, name)
-		if err != nil {
-			return nil, err
+
+		core, needsConstruction := reserveSharedCore(sharedProcessorKey{name: name, configHash: configHash})
+		if needsConstruction {
+			return core.constructOnce(constructor, cfg, log)
 		}
-		if sharedProcessors[name] == nil {
-			sharedProcessors[name] = make(map[uint64]beat.Processor)
+		return core.waitForProcessor(constructor, cfg, log)
+	}
+}
+
+func hashProcessorConfig(cfg *config.C) (uint64, error) {
+	if cfg == nil {
+		return 0, nil
+	}
+	return cfgfile.HashConfig(cfg)
+}
+
+// reserveSharedCore reserves one reference and reports whether the caller must construct it.
+func reserveSharedCore(key sharedProcessorKey) (*sharedCore, bool) {
+	sharedProcessorMu.Lock()
+	defer sharedProcessorMu.Unlock()
+
+	if core, ok := sharedProcessors[key]; ok {
+		core.refCount++
+		return core, false
+	}
+
+	core := &sharedCore{key: key, refCount: 1, ready: make(chan struct{})}
+	sharedProcessors[key] = core
+	return core, true
+}
+
+// constructOnce constructs and publishes a reserved core.
+func (c *sharedCore) constructOnce(constructor Constructor, cfg *config.C, log *logp.Logger) (processor beat.Processor, err error) {
+	defer func() {
+		if panicValue := recover(); panicValue != nil {
+			err = fmt.Errorf("%q processor constructor panicked: %v", c.key.name, panicValue)
+			sharedProcessorMu.Lock()
+			c.constructionErr = err
+			evictLocked(c)
+			sharedProcessorMu.Unlock()
 		}
-		sharedProcessors[name][hash] = safeProcessor
-		return safeProcessor, nil
+		close(c.ready)
+	}()
+
+	processor, err = constructor(cfg, log)
+	if err == nil && processor == nil {
+		err = fmt.Errorf("%q processor constructor returned a nil processor", c.key.name)
 	}
+
+	sharedProcessorMu.Lock()
+	defer sharedProcessorMu.Unlock()
+	if err != nil {
+		c.constructionErr = err
+		// Evict failed attempts so future callers can retry construction.
+		evictLocked(c)
+		return nil, err
+	}
+	c.Processor = processor
+	if !shareable(processor) {
+		evictLocked(c)
+		return wrapUnshared(processor), nil
+	}
+	return newSharedHandle(c), nil
 }
 
-func newSafeProcessor(log *logp.Logger, constructor Constructor, config *config.C, hash uint64, name string) (beat.Processor, error) {
-	processor, err := constructor(config, log)
+func (c *sharedCore) waitForProcessor(constructor Constructor, cfg *config.C, log *logp.Logger) (beat.Processor, error) {
+	<-c.ready
+	if c.constructionErr != nil {
+		// Construction failed, constructOnce has already evicted the core.
+		return nil, c.constructionErr
+	}
+
+	if shareable(c.Processor) {
+		return newSharedHandle(c), nil
+	}
+
+	// The unshareable core was evicted, so this caller needs its own instance.
+	processor, err := constructor(cfg, log)
 	if err != nil {
 		return nil, err
 	}
-	// if the processor does not implement `Closer` it does not need a wrap
-	if _, ok := processor.(Closer); !ok {
-		// if SetPaths is implemented, ensure single call of SetPaths
-		if _, ok = processor.(PathSetter); ok {
-			return &SafeProcessor{Processor: processor, hash: hash, name: name, refCount: 1}, nil
-		}
-		return processor, nil
+	return wrapUnshared(processor), nil
+}
+
+// evictLocked removes core from sharedProcessors. Callers must hold
+// sharedProcessorMu.
+func evictLocked(core *sharedCore) {
+	delete(sharedProcessors, core.key)
+}
+
+// shareable reports whether p can be reused by independent owners.
+func shareable(p beat.Processor) bool {
+	switch p.(type) {
+	case PathSetter, Unshareable:
+		return false
+	default:
+		return true
 	}
+}
 
-	if pdataProc, ok := processor.(PdataProcessor); ok {
-		return &safePdataProcessorWithClose{
-			safeProcessorWithClose: safeProcessorWithClose{
-				SafeProcessor: SafeProcessor{Processor: processor, hash: hash, name: name, refCount: 1},
-			},
-			pdataProc: pdataProc,
-		}, nil
+// newSharedHandle returns a handle owning one reference to core. When the shared
+// processor implements PdataProcessor the returned handle also forwards
+// RunPdata so the pdata fast path survives sharing.
+func newSharedHandle(core *sharedCore) beat.Processor {
+	h := &sharedHandle{sharedCore: core}
+	if pp, ok := core.Processor.(PdataProcessor); ok {
+		return &sharedPdataHandle{sharedHandle: h, pdata: pp}
+	}
+	return h
+}
+
+func (h *sharedHandle) Run(event *beat.Event) (*beat.Event, error) {
+	if h.closed.Load() {
+		return nil, ErrClosed
+	}
+	return h.sharedCore.Run(event)
+}
+
+func (h *sharedHandle) Close() error {
+	if h.closed.Swap(true) {
+		return nil
+	}
+
+	sharedProcessorMu.Lock()
+	h.refCount--
+	if h.refCount > 0 {
+		sharedProcessorMu.Unlock()
+		return nil
+	}
+	evictLocked(h.sharedCore)
+	sharedProcessorMu.Unlock()
+
+	// Close outside the global lock so a slow processor cannot stall unrelated
+	// construction or teardown.
+	return h.sharedCore.Close()
+}
+
+func (h *sharedPdataHandle) RunPdata(body pcommon.Map) (bool, error) {
+	if h.closed.Load() {
+		return false, ErrClosed
+	}
+	// A live handle holds a reference, so the core cannot be released and its
+	// underlying processor cannot be closed while this runs. The RLock mirrors
+	// SafeProcessor.Run and guards against a concurrent Close on the core.
+	c := h.sharedCore
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	if c.state == stateClosed {
+		return false, ErrClosed
 	}
-	return &safeProcessorWithClose{
-		SafeProcessor: SafeProcessor{Processor: processor, hash: hash, name: name, refCount: 1},
-	}, nil
+	return h.pdata.RunPdata(body)
 }
diff --git a/libbeat/processors/safe_processor_test.go b/libbeat/processors/safe_processor_test.go
index f5d02fba4f79..da2703147bfd 100644
--- a/libbeat/processors/safe_processor_test.go
+++ b/libbeat/processors/safe_processor_test.go
@@ -18,14 +18,20 @@
 package processors
 
 import (
+	"errors"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/paths"
 )
 
@@ -54,6 +60,32 @@ func (p *mockCloserProcessor) Close() error {
 	return nil
 }
 
+type concurrentMockCloserProcessor struct {
+	closed atomic.Bool
+	runs   atomic.Int32
+	closes atomic.Int32
+}
+
+func (p *concurrentMockCloserProcessor) Run(event *beat.Event) (*beat.Event, error) {
+	if p.closed.Load() {
+		return nil, errors.New("processor is closed")
+	}
+	p.runs.Add(1)
+	return event, nil
+}
+
+func (p *concurrentMockCloserProcessor) Close() error {
+	p.closes.Add(1)
+	if p.closed.Swap(true) {
+		return errors.New("processor closed twice")
+	}
+	return nil
+}
+
+func (p *concurrentMockCloserProcessor) String() string {
+	return "concurrent-mock-closer"
+}
+
 func newMockCloserConstructor() (Constructor, *mockCloserProcessor) {
 	p := mockCloserProcessor{}
 	constructor := func(config *config.C, _ *logp.Logger) (beat.Processor, error) {
@@ -96,6 +128,22 @@ func newMockPathSetterProcessor() (Constructor, *mockPathSetterProcessor) {
 	return constructor, &p
 }
 
+type mockUnshareableProcessor struct {
+	mockProcessor
+}
+
+func (p *mockUnshareableProcessor) Unshareable() {}
+
+type mockPdataProcessor struct {
+	mockProcessor
+	pdataCount int
+}
+
+func (p *mockPdataProcessor) RunPdata(pcommon.Map) (bool, error) {
+	p.pdataCount++
+	return false, nil
+}
+
 func mockConstructor(config *config.C, log *logp.Logger) (beat.Processor, error) {
 	return &mockProcessor{}, nil
 }
@@ -105,13 +153,14 @@ func mockCloserConstructor(config *config.C, log *logp.Logger) (beat.Processor,
 }
 
 func TestSafeWrap(t *testing.T) {
-	t.Run("does not wrap a non-closer processor", func(t *testing.T) {
+	t.Run("shares a non-closer processor", func(t *testing.T) {
 		nonCloser := mockConstructor
-		wrappedNonCloser := SafeWrap("non-closer processor", nonCloser)
+		wrappedNonCloser := SafeWrap("shared non-closer processor", nonCloser)
 		wp, err := wrappedNonCloser(nil, nil)
 		require.NoError(t, err)
-		assert.IsType(t, &mockProcessor{}, wp)
-		assert.NotImplements(t, (*Closer)(nil), wp)
+		assert.IsType(t, &sharedHandle{}, wp)
+		assert.Implements(t, (*Closer)(nil), wp)
+		require.NoError(t, Close(wp))
 	})
 
 	t.Run("wraps a closer processor", func(t *testing.T) {
@@ -119,8 +168,18 @@ func TestSafeWrap(t *testing.T) {
 		wrappedCloser := SafeWrap("closer processor", closer)
 		wcp, err := wrappedCloser(nil, nil)
 		require.NoError(t, err)
-		assert.IsType(t, &safeProcessorWithClose{}, wcp)
+		assert.IsType(t, &sharedHandle{}, wcp)
 		assert.Implements(t, (*Closer)(nil), wcp)
+		require.NoError(t, Close(wcp))
+	})
+
+	t.Run("wraps a closer path-setter processor without sharing", func(t *testing.T) {
+		cons, _ := newMockPathSetterCloserProcessor()
+		wrapped := SafeWrap("closer path-setter processor", cons)
+		wp, err := wrapped(nil, nil)
+		require.NoError(t, err)
+		assert.IsType(t, &safeProcessorWithClose{}, wp)
+		assert.Implements(t, (*Closer)(nil), wp)
 	})
 }
 
@@ -255,7 +314,12 @@ func TestSafeProcessorSetPathsClose(t *testing.T) {
 
 func TestSafeWrapSharedInstanceByNameAndHash(t *testing.T) {
 	cons, p := newMockCloserConstructor()
-	sw := SafeWrap("test-shared-instance", cons)
+	constructions := 0
+	counting := func(cfg *config.C, log *logp.Logger) (beat.Processor, error) {
+		constructions++
+		return cons(cfg, log)
+	}
+	sw := SafeWrap("test-shared-instance", counting)
 
 	proc1, err := sw(nil, nil)
 	require.NoError(t, err, "first SafeWrap call should succeed")
@@ -263,7 +327,8 @@ func TestSafeWrapSharedInstanceByNameAndHash(t *testing.T) {
 	proc2, err := sw(nil, nil)
 	require.NoError(t, err, "second SafeWrap call should succeed")
 
-	assert.Same(t, proc1, proc2, "same name+config should return the same processor pointer")
+	assert.Equal(t, 1, constructions, "same name+config must construct the underlying processor once")
+	assert.NotSame(t, proc1, proc2, "each call returns its own handle to the shared instance")
 
 	_, err = proc1.Run(nil)
 	require.NoError(t, err, "Run via proc1 should succeed")
@@ -272,6 +337,11 @@ func TestSafeWrapSharedInstanceByNameAndHash(t *testing.T) {
 	require.NoError(t, Close(proc1), "first Close should not error")
 	assert.Equal(t, 0, p.closeCount, "underlying processor should not be closed while a ref remains")
 
+	require.NoError(t, Close(proc1), "duplicate Close of the same handle should not error")
+	assert.Equal(t, 0, p.closeCount, "duplicate Close must not release another handle's reference")
+	_, err = proc2.Run(nil)
+	require.NoError(t, err, "proc2 must remain usable after proc1 was closed twice")
+
 	require.NoError(t, Close(proc2), "second Close should not error")
 	assert.Equal(t, 1, p.closeCount, "underlying processor should be closed once all refs are released")
 }
@@ -320,6 +390,234 @@ func TestSafeWrapRefCountingPreventsEarlyClose(t *testing.T) {
 	assert.Equal(t, 1, p.closeCount, "should close exactly once after last ref is released")
 }
 
+func TestSafeWrapConcurrentConstructionReservesReferences(t *testing.T) {
+	const (
+		name    = "test-concurrent-construction-reserves-references"
+		callers = 8
+	)
+
+	started := make(chan struct{})
+	release := make(chan struct{})
+	creatorClosed := make(chan struct{})
+	underlying := &concurrentMockCloserProcessor{}
+	var constructions atomic.Int32
+	cons := func(*config.C, *logp.Logger) (beat.Processor, error) {
+		if constructions.Add(1) == 1 {
+			close(started)
+		}
+		<-release
+		return underlying, nil
+	}
+	sw := SafeWrap(name, cons)
+	logger := logptest.NewTestingLogger(t, "")
+
+	type result struct {
+		constructErr error
+		runErr       error
+		closeErr     error
+	}
+	results := make(chan result, callers)
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		defer close(creatorClosed)
+		p, err := sw(nil, logger)
+		var closeErr error
+		if err == nil {
+			closeErr = Close(p)
+		}
+		results <- result{constructErr: err, closeErr: closeErr}
+	})
+	<-started
+	for range callers - 1 {
+		wg.Go(func() {
+			p, err := sw(nil, logger)
+			if err != nil {
+				results <- result{constructErr: err}
+				return
+			}
+			<-creatorClosed
+			_, runErr := p.Run(nil)
+			results <- result{runErr: runErr, closeErr: Close(p)}
+		})
+	}
+
+	allWaiting := waitForSharedReferences(t, name, callers)
+	close(release)
+	wg.Wait()
+
+	assert.True(t, allWaiting, "all callers must join the in-flight construction")
+	assert.Equal(t, int32(1), constructions.Load(),
+		"the creator closing before waiters run must not trigger another construction")
+	for range callers {
+		result := <-results
+		assert.NoError(t, result.constructErr, "every caller must receive a handle")
+		assert.NoError(t, result.runErr, "waiters must retain a live reference after the creator closes")
+		assert.NoError(t, result.closeErr, "every handle must close cleanly")
+	}
+	assert.Equal(t, int32(callers-1), underlying.runs.Load(), "every waiter must run once")
+	assert.Equal(t, int32(1), underlying.closes.Load(), "the underlying processor must close exactly once")
+}
+
+func TestSafeWrapConcurrentUnshareableConstruction(t *testing.T) {
+	const (
+		name    = "test-concurrent-unshareable-construction"
+		callers = 8
+	)
+
+	started := make(chan struct{})
+	release := make(chan struct{})
+	var constructions atomic.Int32
+	cons := func(*config.C, *logp.Logger) (beat.Processor, error) {
+		if constructions.Add(1) == 1 {
+			close(started)
+			<-release
+		}
+		return &mockUnshareableProcessor{}, nil
+	}
+	sw := SafeWrap(name, cons)
+	logger := logptest.NewTestingLogger(t, "")
+
+	type result struct {
+		processor beat.Processor
+		err       error
+	}
+	results := make(chan result, callers)
+	call := func() {
+		p, err := sw(nil, logger)
+		results <- result{processor: p, err: err}
+	}
+
+	var wg sync.WaitGroup
+	wg.Go(call)
+	<-started
+	for range callers - 1 {
+		wg.Go(call)
+	}
+
+	allWaiting := waitForSharedReferences(t, name, callers)
+	close(release)
+	wg.Wait()
+
+	assert.True(t, allWaiting, "all callers must join the in-flight construction")
+	assert.Equal(t, int32(callers), constructions.Load(), "every caller must construct an unshareable processor")
+	seen := make(map[beat.Processor]struct{}, callers)
+	for range callers {
+		result := <-results
+		assert.NoError(t, result.err, "unshareable construction must succeed")
+		_, exists := seen[result.processor]
+		assert.False(t, exists, "each caller must receive a distinct processor")
+		seen[result.processor] = struct{}{}
+	}
+}
+
+func TestSafeWrapConcurrentConstructionError(t *testing.T) {
+	const (
+		name    = "test-concurrent-construction-error"
+		callers = 8
+	)
+
+	expectedErr := errors.New("construction failed")
+	started := make(chan struct{})
+	release := make(chan struct{})
+	var constructions atomic.Int32
+	cons := func(*config.C, *logp.Logger) (beat.Processor, error) {
+		if constructions.Add(1) == 1 {
+			close(started)
+		}
+		<-release
+		return nil, expectedErr
+	}
+	sw := SafeWrap(name, cons)
+	logger := logptest.NewTestingLogger(t, "")
+
+	results := make(chan error, callers)
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		_, err := sw(nil, logger)
+		results <- err
+	})
+	<-started
+	for range callers - 1 {
+		wg.Go(func() {
+			_, err := sw(nil, logger)
+			results <- err
+		})
+	}
+
+	allWaiting := waitForSharedReferences(t, name, callers)
+	close(release)
+	wg.Wait()
+
+	assert.True(t, allWaiting, "all callers must join the in-flight construction")
+	assert.Equal(t, int32(1), constructions.Load(), "one failed construction must serve all waiting callers")
+	for range callers {
+		assert.ErrorIs(t, <-results, expectedErr, "all waiting callers must receive the construction error")
+	}
+}
+
+func TestSafeWrapConcurrentConstructionPanic(t *testing.T) {
+	const (
+		name    = "test-concurrent-construction-panic"
+		callers = 8
+	)
+
+	started := make(chan struct{})
+	release := make(chan struct{})
+	var constructions atomic.Int32
+	cons := func(*config.C, *logp.Logger) (beat.Processor, error) {
+		if constructions.Add(1) == 1 {
+			close(started)
+		}
+		<-release
+		panic("construction panic")
+	}
+	sw := SafeWrap(name, cons)
+	logger := logptest.NewTestingLogger(t, "")
+
+	type result struct {
+		err        error
+		panicValue any
+	}
+	results := make(chan result, callers)
+	call := func() {
+		var err error
+		defer func() {
+			results <- result{err: err, panicValue: recover()}
+		}()
+		_, err = sw(nil, logger)
+	}
+
+	var wg sync.WaitGroup
+	wg.Go(call)
+	<-started
+	for range callers - 1 {
+		wg.Go(call)
+	}
+
+	allWaiting := waitForSharedReferences(t, name, callers)
+	close(release)
+	wg.Wait()
+
+	assert.True(t, allWaiting, "all callers must join the in-flight construction")
+	assert.Equal(t, int32(1), constructions.Load(), "waiters must not retry a panicking construction")
+	for range callers {
+		result := <-results
+		assert.Nil(t, result.panicValue, "constructor panics must be converted to errors")
+		assert.ErrorContains(t, result.err, "processor constructor panicked: construction panic",
+			"all callers must receive the recovered constructor panic")
+	}
+}
+
+func waitForSharedReferences(t *testing.T, name string, want int) bool {
+	t.Helper()
+	return assert.Eventually(t, func() bool {
+		sharedProcessorMu.Lock()
+		defer sharedProcessorMu.Unlock()
+		core := sharedProcessors[sharedProcessorKey{name: name}]
+		return core != nil && core.refCount == want
+	}, time.Second, time.Millisecond, "shared core must reserve every waiting caller's reference")
+}
+
 func TestSafeWrapNewInstanceAfterAllRefsClosed(t *testing.T) {
 	sw := SafeWrap("test-recreate-after-close", mockCloserConstructor)
 
@@ -328,7 +626,6 @@ func TestSafeWrapNewInstanceAfterAllRefsClosed(t *testing.T) {
 
 	require.NoError(t, Close(proc1), "closing the only reference should succeed")
 
-	// Entry is removed from sharedProcessors; next call must build a fresh instance.
 	proc2, err := sw(nil, nil)
 	require.NoError(t, err, "SafeWrap after full close should succeed")
 
@@ -340,6 +637,47 @@ func TestSafeWrapNewInstanceAfterAllRefsClosed(t *testing.T) {
 	require.NoError(t, Close(proc2))
 }
 
+func TestSafeWrapUnshareableNotShared(t *testing.T) {
+	constructions := 0
+	cons := func(cfg *config.C, log *logp.Logger) (beat.Processor, error) {
+		constructions++
+		return &mockUnshareableProcessor{}, nil
+	}
+	sw := SafeWrap("test-unshareable", cons)
+
+	proc1, err := sw(nil, nil)
+	require.NoError(t, err, "first SafeWrap call should succeed")
+	proc2, err := sw(nil, nil)
+	require.NoError(t, err, "second SafeWrap call should succeed")
+
+	assert.Equal(t, 2, constructions, "an Unshareable processor must be constructed once per owner")
+	assert.NotSame(t, proc1, proc2, "an Unshareable processor must not be shared across owners")
+	assert.NotImplements(t, (*Closer)(nil), proc1)
+}
+
+func TestSafeWrapForwardsRunPdata(t *testing.T) {
+	p := &mockPdataProcessor{}
+	cons := func(cfg *config.C, log *logp.Logger) (beat.Processor, error) { return p, nil }
+	sw := SafeWrap("test-pdata", cons)
+
+	wp, err := sw(nil, nil)
+	require.NoError(t, err, "SafeWrap of a pdata processor should succeed")
+
+	assert.IsType(t, &sharedPdataHandle{}, wp, "a shared PdataProcessor must return a pdata-aware handle")
+	pp, ok := wp.(PdataProcessor)
+	require.True(t, ok, "shared handle must implement PdataProcessor")
+
+	drop, err := pp.RunPdata(pcommon.NewMap())
+	require.NoError(t, err, "RunPdata should forward without error")
+	assert.False(t, drop, "mock does not drop the event")
+	assert.Equal(t, 1, p.pdataCount, "RunPdata must reach the underlying processor")
+
+	require.NoError(t, Close(wp))
+	_, err = pp.RunPdata(pcommon.NewMap())
+	assert.ErrorIs(t, err, ErrClosed, "RunPdata must return ErrClosed after the handle is closed")
+	assert.Equal(t, 1, p.pdataCount, "RunPdata must not reach the underlying after close")
+}
+
 func TestSafeProcessorSetPaths(t *testing.T) {
 	cons, p := newMockPathSetterProcessor()
 	var (
diff --git a/libbeat/processors/shared_churn_stress_test.go b/libbeat/processors/shared_churn_stress_test.go
new file mode 100644
index 000000000000..32779b267d19
--- /dev/null
+++ b/libbeat/processors/shared_churn_stress_test.go
@@ -0,0 +1,151 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package processors_test
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/processors"
+	"github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
+)
+
+// churnProcessor reports Run-after-Close and double-Close lifecycle violations.
+type churnProcessor struct {
+	closed atomic.Bool
+	runs   atomic.Int64
+	closes atomic.Int64
+}
+
+func (p *churnProcessor) Run(event *beat.Event) (*beat.Event, error) {
+	if p.closed.Load() {
+		return nil, errors.New("BUG: Run called on underlying processor after Close")
+	}
+	p.runs.Add(1)
+	return event, nil
+}
+
+func (p *churnProcessor) Close() error {
+	p.closes.Add(1)
+	if p.closed.Swap(true) {
+		return errors.New("BUG: underlying processor closed twice")
+	}
+	return nil
+}
+
+func (p *churnProcessor) String() string { return "churn-processor" }
+
+// TestSharedProcessorConcurrentChurn stresses sharing and isolation across configurations.
+func TestSharedProcessorConcurrentChurn(t *testing.T) {
+	const (
+		procName       = "test-shared-churn-processor"
+		churnWorkers   = 20
+		churnIters     = 200
+		runsPerIter    = 5
+		steadyWorkers  = 30
+		runsPerHold    = 20
+		steadyReCycles = 50
+	)
+
+	var (
+		createdMu sync.Mutex
+		created   []*churnProcessor
+	)
+	wrapped := processors.SafeWrap(procName, func(*config.C, *logp.Logger) (beat.Processor, error) {
+		p := &churnProcessor{}
+		createdMu.Lock()
+		defer createdMu.Unlock()
+		created = append(created, p)
+		return p, nil
+	})
+
+	cfgChurn := config.MustNewConfigFrom(map[string]any{"id": "churn"})
+	cfgSteady := config.MustNewConfigFrom(map[string]any{"id": "steady"})
+
+	logger := logptest.NewTestingLogger(t, "")
+	event := &beat.Event{}
+
+	var wg sync.WaitGroup
+
+	// Churn workers: repeatedly construct, run, close the same config.
+	for w := range churnWorkers {
+		wg.Go(func() {
+			for i := range churnIters {
+				p, err := wrapped(cfgChurn, logger)
+				if !assert.NoErrorf(t, err, "churn worker %d iter %d: construct failed", w, i) {
+					return
+				}
+				for r := range runsPerIter {
+					out, err := p.Run(event)
+					assert.NoErrorf(t, err,
+						"churn worker %d iter %d run %d: unexpected Run error on live ref", w, i, r)
+					assert.Samef(t, event, out,
+						"churn worker %d iter %d run %d: event not passed through", w, i, r)
+				}
+				assert.NoErrorf(t, processors.Close(p), "churn worker %d iter %d: Close failed", w, i)
+			}
+		})
+	}
+
+	// Steady workers: use a different config of the same processor name.
+	for w := range steadyWorkers {
+		wg.Go(func() {
+			for c := range steadyReCycles {
+				p, err := wrapped(cfgSteady, logger)
+				if !assert.NoErrorf(t, err, "steady worker %d cycle %d: construct failed", w, c) {
+					return
+				}
+				for r := range runsPerHold {
+					out, err := p.Run(event)
+					assert.NoErrorf(t, err,
+						"steady worker %d cycle %d run %d: unexpected Run error on live ref", w, c, r)
+					assert.Samef(t, event, out,
+						"steady worker %d cycle %d run %d: event not passed through", w, c, r)
+				}
+				assert.NoErrorf(t, processors.Close(p), "steady worker %d cycle %d: Close failed", w, c)
+			}
+		})
+	}
+
+	wg.Wait()
+
+	var totalRuns int64
+	for i, p := range created {
+		assert.Truef(t, p.closed.Load(), "underlying processor %d must be closed after all refs released", i)
+		assert.EqualValuesf(t, 1, p.closes.Load(), "underlying processor %d must be closed exactly once", i)
+		totalRuns += p.runs.Load()
+	}
+	expectedRuns := int64(churnWorkers*churnIters*runsPerIter + steadyWorkers*steadyReCycles*runsPerHold)
+	assert.Equal(t, expectedRuns, totalRuns, "every Run must have been processed exactly once")
+
+	before := len(created)
+	p, err := wrapped(cfgChurn, logger)
+	require.NoError(t, err)
+	assert.Len(t, created, before+1, "a fresh underlying instance must be constructed after full release")
+	_, err = p.Run(event)
+	assert.NoError(t, err, "fresh instance must be runnable")
+	assert.NoError(t, processors.Close(p))
+}

From c7f67a821f1a1558086d6320a0ec139bf3fa360e Mon Sep 17 00:00:00 2001
From: Orestis Floros <orestis.floros@elastic.co>
Date: Thu, 30 Jul 2026 08:26:52 +0200
Subject: [PATCH 5/5] Update changelog

---
 ...are-identically-configured-processors-across-inputs.yaml | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/changelog/fragments/1783069200-share-identically-configured-processors-across-inputs.yaml b/changelog/fragments/1783069200-share-identically-configured-processors-across-inputs.yaml
index c206f387eca3..146ac5f1695d 100644
--- a/changelog/fragments/1783069200-share-identically-configured-processors-across-inputs.yaml
+++ b/changelog/fragments/1783069200-share-identically-configured-processors-across-inputs.yaml
@@ -13,7 +13,7 @@ kind: bug-fix
 
 # REQUIRED for all kinds
 # Change summary; a 80ish characters long description of the change.
-summary: fix excessive memory and goroutine usage when many inputs use the same processor configuration
+summary: fix excessive memory usage when many inputs use the same processor configuration
 
 # REQUIRED for breaking-change, deprecation, known-issue
 # Long description; in case the summary is not enough to describe the change
@@ -21,9 +21,7 @@ summary: fix excessive memory and goroutine usage when many inputs use the same
 description: >
   Memory usage and goroutine count no longer grow with the number of log files
   when inputs use processors such as add_kubernetes_metadata, for example
-  filestream collecting container logs. Previously each harvester ran its own
-  copy of the configured processors, which on hosts with many log files could
-  consume hundreds of megabytes and cause OOM restarts.
+  filestream collecting container logs.
 
 # REQUIRED for breaking-change, deprecation, known-issue
 # impact: