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") +}