fix(syslog): don't close the shared acquisition output channel

pkg/acquisition/modules/no_shared_channel_close_test.go

@@ -0,0 +1,95 @@
+package modules
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestDatasourcesDoNotCloseSharedOutputChannel statically checks every
+// datasource module for `close(out)` inside a Stream() or StreamingAcquisition()
+// method. `out` is the pipeline output channel shared across every configured
+// datasource (see acquisition.StartAcquisition); it is owned by the acquisition
+// orchestrator, not by any individual datasource, so a datasource closing it
+// races with any other datasource still sending on it and panics with "send on
+// closed channel" (this happened in production with the syslog datasource).
+//
+// This is independent of any one datasource and covers every module directory
+// under pkg/acquisition/modules automatically, including future ones, without
+// needing to configure or run any of them.
+func TestDatasourcesDoNotCloseSharedOutputChannel(t *testing.T) {
+	_, thisFile, _, ok := runtime.Caller(0)
+	require.True(t, ok)
+
+	modulesDir := filepath.Dir(thisFile)
+
+	entries, err := os.ReadDir(modulesDir)
+	require.NoError(t, err)
+
+	for _, entry := range entries {
+		if !entry.IsDir() {
+			continue
+		}
+
+		moduleDir := filepath.Join(modulesDir, entry.Name())
+
+		fset := token.NewFileSet()
+
+		pkgs, err := parser.ParseDir(fset, moduleDir, func(fi os.FileInfo) bool {
+			return filepath.Ext(fi.Name()) == ".go"
+		}, 0)
+		require.NoError(t, err, "parsing module %s", entry.Name())
+
+		for _, pkg := range pkgs {
+			for path, file := range pkg.Files {
+				checkFileDoesNotCloseOut(t, fset, entry.Name(), path, file)
+			}
+		}
+	}
+}
+
+// checkFileDoesNotCloseOut walks a single file's AST for Stream/StreamingAcquisition
+// method declarations and fails if their body contains close(out).
+func checkFileDoesNotCloseOut(t *testing.T, fset *token.FileSet, module string, path string, file *ast.File) {
+	t.Helper()
+
+	for _, decl := range file.Decls {
+		fn, ok := decl.(*ast.FuncDecl)
+		if !ok || fn.Body == nil {
+			continue
+		}
+
+		if fn.Name.Name != "Stream" && fn.Name.Name != "StreamingAcquisition" {
+			continue
+		}
+
+		ast.Inspect(fn.Body, func(n ast.Node) bool {
+			call, ok := n.(*ast.CallExpr)
+			if !ok {
+				return true
+			}
+
+			ident, ok := call.Fun.(*ast.Ident)
+			if !ok || ident.Name != "close" || len(call.Args) != 1 {
+				return true
+			}
+
+			arg, ok := call.Args[0].(*ast.Ident)
+			if ok && arg.Name == "out" {
+				t.Errorf(
+					"%s: %s.%s must not close(out) -- it is the shared acquisition "+
+						"output channel, owned by the orchestrator, not by this datasource (%s:%d)",
+					module, module, fn.Name.Name, path, fset.Position(call.Pos()).Line,
+				)
+			}
+
+			return true
+		})
+	}
+}

pkg/acquisition/modules/syslog/run.go

@@ -43,8 +43,6 @@ func (s *Source) Stream(ctx context.Context, out chan pipeline.Event) error {
 	})
 
 	g.Go(func() error {
-		defer close(out)
-
 		for {
 			select {
 			case <-ctx.Done():

pkg/acquisition/types/types.go

@@ -76,6 +76,10 @@ type Fetcher interface {
 //    failures, but treat them as errors. The caller is responsible for supervising
 //    Stream(), and restarting it as needed. There is currently no way to differentiate
 //    retryable vs permanent errors.
+//  - never close the output channel: it is shared and fed by every configured
+//    datasource, so it is owned by the acquisition orchestrator, not by any single
+//    Stream() implementation. Closing it here would race with (and can panic) any
+//    other datasource still sending on it.
 type RestartableStreamer interface {
 	// Start live acquisition (eg, tail a file)
 	Stream(ctx context.Context, out chan pipeline.Event) error
@@ -84,6 +88,9 @@ type RestartableStreamer interface {
 // Tailer has the same pupose as RestartableStreamer (provide ongoing events) but
 // is responsible for spawning its own goroutines, and handling errors and retries.
 // New datasources are expected to implement RestartableStreamer instead.
+//
+// As with RestartableStreamer, the output channel is shared and orchestrator-owned:
+// an implementation must never close it.
 type Tailer interface {
 	StreamingAcquisition(ctx context.Context, out chan pipeline.Event, acquisTomb *tomb.Tomb) error
 }