From 57ab9aefe7be8f093fe113542e8e07a12401c625 Mon Sep 17 00:00:00 2001
From: Nick <nick@myschuch.com>
Date: Fri, 4 Sep 2026 14:10:18 +1000
Subject: [PATCH] Close ring-buffer readers on tracer init failure

The FPM tracer opened the events reader and then the Drupal cache reader; if
the second open failed, the first was never closed, because ringreader.Run,
which takes ownership of both, was never reached. Across the collector
supervisor's restart loop a recurring failure leaked a kernel ring-buffer
mapping each attempt. Close the events reader on that error path.

ringreader.Run also leaked the readers it was handed if source validation
failed before its cleanup goroutine took over. Close all provided readers on
the validation-failure path too, via a shared closeAll helper, and add a test.
The single-reader CLI and Node tracers hand their reader straight to Run with
no intervening fallible step, so this covers them as well.
---
 pkg/tracer/php/fpm/tracer.go             |  3 +++
 pkg/tracer/ringreader/ringreader.go      | 18 +++++++++++---
 pkg/tracer/ringreader/ringreader_test.go | 30 ++++++++++++++++++++++++
 3 files changed, 48 insertions(+), 3 deletions(-)

diff --git a/pkg/tracer/php/fpm/tracer.go b/pkg/tracer/php/fpm/tracer.go
index bd41f1c..03c9742 100644
--- a/pkg/tracer/php/fpm/tracer.go
+++ b/pkg/tracer/php/fpm/tracer.go
@@ -336,6 +336,9 @@ func Run(ctx context.Context, plugin sink.Interface, extensionPath string, maxFu
 
 	drupalReader, err := ringbuf.NewReader(objs.DrupalCacheEvents)
 	if err != nil {
+		// The events reader is open but ringreader.Run, which would take
+		// ownership of both, is never reached on this path.
+		reader.Close()
 		return logger.WrapError(fmt.Errorf("failed to start drupal cache event reader: %w", err))
 	}
 
diff --git a/pkg/tracer/ringreader/ringreader.go b/pkg/tracer/ringreader/ringreader.go
index b645718..1a06153 100644
--- a/pkg/tracer/ringreader/ringreader.go
+++ b/pkg/tracer/ringreader/ringreader.go
@@ -52,6 +52,9 @@ func Run(ctx context.Context, sources ...Source) error {
 
 	for _, source := range sources {
 		if err := validate(source); err != nil {
+			// Run takes ownership of the readers on the success path, so close
+			// whatever was handed in rather than leaking it when rejecting.
+			closeAll(sources)
 			return err
 		}
 	}
@@ -66,15 +69,24 @@ func Run(ctx context.Context, sources ...Source) error {
 
 	group.Go(func() error {
 		<-groupCtx.Done()
-		for _, source := range sources {
-			_ = source.Reader.Close()
-		}
+		closeAll(sources)
 		return nil
 	})
 
 	return group.Wait()
 }
 
+// closeAll closes every reader which was provided, ignoring nil readers and
+// close errors. A reader may be closed twice across the lifetime of Run: once
+// here on shutdown is the norm, and calling Close again is harmless.
+func closeAll(sources []Source) {
+	for _, source := range sources {
+		if source.Reader != nil {
+			_ = source.Reader.Close()
+		}
+	}
+}
+
 func read(ctx context.Context, source Source) error {
 	for {
 		if ctx.Err() != nil {
diff --git a/pkg/tracer/ringreader/ringreader_test.go b/pkg/tracer/ringreader/ringreader_test.go
index 0c6e661..63ab8e3 100644
--- a/pkg/tracer/ringreader/ringreader_test.go
+++ b/pkg/tracer/ringreader/ringreader_test.go
@@ -165,3 +165,33 @@ func TestRun_FPMReadErrorClosesSiblingReader(t *testing.T) {
 		t.Fatal("FPM sibling reader was not closed")
 	}
 }
+
+func TestRun_ValidationFailureClosesReaders(t *testing.T) {
+	valid := &fakeReader{
+		readFn: func() (ringbuf.Record, error) {
+			return ringbuf.Record{}, ringbuf.ErrClosed
+		},
+	}
+	// A nil Handle fails validation, so Run rejects the sources before taking
+	// ownership. The readers handed in must still be closed rather than leaked.
+	invalid := &fakeReader{
+		readFn: func() (ringbuf.Record, error) {
+			return ringbuf.Record{}, ringbuf.ErrClosed
+		},
+	}
+
+	err := Run(t.Context(),
+		Source{
+			Reader: valid, Runtime: ingest.RuntimePHPFPM, Stream: StreamEvents,
+			Handle: func(context.Context, []byte) error { return nil },
+		},
+		Source{
+			Reader: invalid, Runtime: ingest.RuntimePHPFPM, Stream: StreamDrupalCache,
+			Handle: nil,
+		},
+	)
+
+	require.Error(t, err)
+	assert.Equal(t, int32(1), valid.closes.Load(), "the valid reader must be closed on validation failure")
+	assert.Equal(t, int32(1), invalid.closes.Load(), "the invalid reader must be closed on validation failure")
+}