zapingest: the address is the switch — a linked library takes no port

hanzoai/o11y v1.5.41 shipped pkg/zapingest defaulting Enabled=true with
0.0.0.0:4317-4319 baked in, so importing the module SEIZED three well-known
ports as a side effect of being linked. hanzoai/cloud embeds o11y and runs its
own ingest on the same three ports in the same process; whichever bound first
won, the loser dropped telemetry silently, and log ingest fell ~200x. The
chart's O11Y_ZAPINGEST_ENABLED=false is a tourniquet on that default, and a
flag that must be set correctly in every chart is a third way to get it wrong.

The address IS the switch. `Enabled` is deleted: a boolean beside three
addresses is a second way to state one decision, and the two can disagree —
Validate() already had to reconcile them. Stating where spans arrive IS the act
of deciding that spans arrive here. Every address now defaults to EMPTY, so
linking this package binds nothing and the race cannot be reached by default.
A signal with no address is simply not received; the conventional ports
(4317/4318/4319) are documented rather than defaulted, so the deployment that
publishes them is the one that chooses them.

A collision is now LOUD instead of lost. holderOf() reads procfs to name the
process holding the port, and says so explicitly when that process is THIS one
— which is the actual production shape, two implementations in one binary, and
the case where the answer is "delete one" rather than "retry" or "move the
port". The error is returned (never swallowed) and logged, and a partial bind
is rolled back: a half-bound edge is worse than none, because it reports
partial traffic behind a healthy-looking surface.

The diagnostic is built into the CAUSE, not passed to Wrap*f. errors.base.Error()
returns the wrapped cause's text and DROPS the wrapper's own message, so a
message handed to Wrap*f is retained on the value and never printed by anything
rendering err.Error(). The first draft of this fix hit exactly that: a loud
message that was invisible. Folding it into the cause makes it print while %w
keeps errors.Is(err, syscall.EADDRINUSE) working.

Removing the field is safe for consumers still setting the env: pkg/config's
decoder does not set ErrorUnused, so a stale O11Y_ZAPINGEST_ENABLED is ignored
rather than fatal.

Tests cover the load-bearing default (binds nothing), signal selection,
validation, holder identification, and that the collision message names this
process.

Co-authored-by: Hanzo Dev <[email protected]>

pkg/zapingest/config.go

@@ -1,6 +1,7 @@
 package zapingest
 
 import (
+	"fmt"
 	"net"
 
 	"github.com/hanzoai/o11y/pkg/errors"
@@ -9,10 +10,27 @@ import (
 
 // Config declares where the three ZAP signals are received. One address per
 // signal, because each receiver runs its own zap.Node and therefore its own
-// listener. 4317 is the span port zapreceiver already defaults to; logs and
-// metrics take the next two.
+// listener.
+//
+// THE ADDRESS IS THE SWITCH. An empty address means that signal is not
+// received in this process. There is no separate `enabled` boolean, because a
+// boolean beside three addresses is a second way to state one decision and the
+// two can disagree — stating where spans arrive IS the act of deciding that
+// spans arrive here.
+//
+// Every address therefore defaults to EMPTY, and that default is load-bearing:
+// linking this package binds nothing. The previous default (enabled, with
+// 0.0.0.0:4317-4319 baked in) meant importing the o11y module seized three
+// well-known ports as a side effect of being linked, which is how a host
+// process and the library it embeds came to race for the same sockets. The
+// loser dropped telemetry silently and log ingest fell ~200x. A library must
+// not take a port because someone linked it.
+//
+// The ports a deployment usually wants are the fleet's conventional ones:
+// spans 4317, logs 4318, metrics 4319. They are documented here rather than
+// defaulted, so the choice is made once, visibly, by the deployment that also
+// publishes them.
 type Config struct {
-	Enabled bool   `mapstructure:"enabled"`
 	Spans   string `mapstructure:"spans"`
 	Logs    string `mapstructure:"logs"`
 	Metrics string `mapstructure:"metrics"`
@@ -23,25 +41,46 @@ func NewConfigFactory() factory.ConfigFactory {
 }
 
 func newConfig() factory.Config {
-	return Config{
-		Enabled: true,
-		Spans:   "0.0.0.0:4317",
-		Logs:    "0.0.0.0:4318",
-		Metrics: "0.0.0.0:4319",
-	}
+	return Config{}
 }
 
-func (c Config) Validate() error {
-	if !c.Enabled {
-		return nil
-	}
-	for name, addr := range map[string]string{"spans": c.Spans, "logs": c.Logs, "metrics": c.Metrics} {
-		if _, _, err := net.SplitHostPort(addr); err != nil {
-			return errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "zapingest %s address %q is not host:port", name, addr)
+// signal is one receivable signal and the address it is received on.
+type signal struct {
+	name string
+	addr string
+}
+
+// signals lists the signals this process receives, in a fixed order, omitting
+// the ones it does not. It is the ONE place that knows "configured" means
+// "non-empty address".
+func (c Config) signals() []signal {
+	out := make([]signal, 0, 3)
+	for _, s := range []signal{{"spans", c.Spans}, {"logs", c.Logs}, {"metrics", c.Metrics}} {
+		if s.addr != "" {
+			out = append(out, s)
 		}
 	}
-	if c.Spans == c.Logs || c.Spans == c.Metrics || c.Logs == c.Metrics {
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "zapingest needs one address per signal, got spans=%s logs=%s metrics=%s", c.Spans, c.Logs, c.Metrics)
+	return out
+}
+
+// Validate reports a bad or duplicated address. The message is folded into the
+// CAUSE rather than passed to Wrap*f: errors.base.Error() renders the wrapped
+// cause and drops the wrapper's own message, so "which signal" would otherwise
+// be kept on the value and never printed.
+func (c Config) Validate() error {
+	configured := c.signals()
+	for _, s := range configured {
+		if _, _, err := net.SplitHostPort(s.addr); err != nil {
+			cause := fmt.Errorf("zapingest %s address %q is not host:port: %w", s.name, s.addr, err)
+			return errors.WrapInvalidInputf(cause, errors.CodeInvalidInput, "%s", cause)
+		}
+	}
+	for i, a := range configured {
+		for _, b := range configured[i+1:] {
+			if a.addr == b.addr {
+				return errors.NewInvalidInputf(errors.CodeInvalidInput, "zapingest needs one address per signal, but %s and %s both use %s", a.name, b.name, a.addr)
+			}
+		}
 	}
 	return nil
 }

pkg/zapingest/holder.go

@@ -0,0 +1,153 @@
+package zapingest
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+)
+
+// holderOf names the process already listening on addr, so a bind collision
+// says WHO it lost to instead of just "address already in use".
+//
+// This exists because the failure it describes is invisible otherwise. Two
+// implementations of one ingest edge, linked into one binary, race for
+// :4317-:4319; whichever binds first wins and the loser's telemetry silently
+// goes nowhere. "address already in use" does not say that the winner is the
+// same PID — and when it is, the answer is not "retry" or "pick another port",
+// it is "delete one of the two implementations".
+//
+// Returns "" when the holder cannot be determined (no procfs, or the socket
+// belongs to another user). The caller must still report the bind error; this
+// only enriches it.
+func holderOf(addr string) string {
+	_, portStr, err := splitPort(addr)
+	if err != nil {
+		return ""
+	}
+	inodes := listenInodes(portStr)
+	if len(inodes) == 0 {
+		return ""
+	}
+	pid, ok := pidForInodes(inodes)
+	if !ok {
+		return ""
+	}
+
+	desc := fmt.Sprintf("pid %d", pid)
+	if cmd := processName(pid); cmd != "" {
+		desc += " (" + cmd + ")"
+	}
+	if pid == os.Getpid() {
+		desc += " — THIS SAME PROCESS: two ingest implementations are linked into one binary, and only one may own the edge"
+	}
+	return desc
+}
+
+func splitPort(addr string) (string, string, error) {
+	i := strings.LastIndex(addr, ":")
+	if i < 0 {
+		return "", "", fmt.Errorf("no port in %q", addr)
+	}
+	p := addr[i+1:]
+	if _, err := strconv.Atoi(p); err != nil {
+		return "", "", err
+	}
+	return addr[:i], p, nil
+}
+
+// listenInodes returns the socket inodes LISTENing on port, across IPv4 and
+// IPv6. procfs writes the port as uppercase hex.
+func listenInodes(port string) map[string]struct{} {
+	n, err := strconv.Atoi(port)
+	if err != nil {
+		return nil
+	}
+	want := fmt.Sprintf("%04X", n)
+	out := map[string]struct{}{}
+	for _, f := range []string{"/proc/net/tcp", "/proc/net/tcp6"} {
+		fh, err := os.Open(f)
+		if err != nil {
+			continue
+		}
+		sc := bufio.NewScanner(fh)
+		sc.Scan() // header
+		for sc.Scan() {
+			fields := strings.Fields(sc.Text())
+			if len(fields) < 10 {
+				continue
+			}
+			// fields[1] = local hexaddr:hexport, fields[3] = state (0A = LISTEN),
+			// fields[9] = inode.
+			local := fields[1]
+			if i := strings.LastIndex(local, ":"); i < 0 || local[i+1:] != want {
+				continue
+			}
+			if fields[3] != "0A" {
+				continue
+			}
+			out[fields[9]] = struct{}{}
+		}
+		fh.Close()
+	}
+	return out
+}
+
+// pidForInodes finds the process holding any of these socket inodes. Our own
+// PID is preferred when it matches, because "the holder is us" is the single
+// most useful thing this function can report.
+func pidForInodes(inodes map[string]struct{}) (int, bool) {
+	self := os.Getpid()
+	if holdsAny(self, inodes) {
+		return self, true
+	}
+	entries, err := os.ReadDir("/proc")
+	if err != nil {
+		return 0, false
+	}
+	for _, e := range entries {
+		pid, err := strconv.Atoi(e.Name())
+		if err != nil || pid == self {
+			continue
+		}
+		if holdsAny(pid, inodes) {
+			return pid, true
+		}
+	}
+	return 0, false
+}
+
+func holdsAny(pid int, inodes map[string]struct{}) bool {
+	fds, err := os.ReadDir(filepath.Join("/proc", strconv.Itoa(pid), "fd"))
+	if err != nil {
+		return false
+	}
+	for _, fd := range fds {
+		link, err := os.Readlink(filepath.Join("/proc", strconv.Itoa(pid), "fd", fd.Name()))
+		if err != nil {
+			continue
+		}
+		ino, ok := strings.CutPrefix(link, "socket:[")
+		if !ok {
+			continue
+		}
+		if _, hit := inodes[strings.TrimSuffix(ino, "]")]; hit {
+			return true
+		}
+	}
+	return false
+}
+
+func processName(pid int) string {
+	b, err := os.ReadFile(filepath.Join("/proc", strconv.Itoa(pid), "cmdline"))
+	if err == nil && len(b) > 0 {
+		return strings.Join(strings.FieldsFunc(string(b), func(r rune) bool { return r == 0 }), " ")
+	}
+	b, err = os.ReadFile(filepath.Join("/proc", strconv.Itoa(pid), "comm"))
+	if err != nil {
+		return ""
+	}
+	return strings.TrimSpace(string(b))
+}

pkg/zapingest/zapingest.go

@@ -8,29 +8,42 @@
 // pkg/zaplogreceiver and pkg/zapmetricreceiver decoded batches and handed them
 // to a Handler nobody supplied, while pkg/datastoretraces, pkg/datastorelogs and
 // pkg/datastoremetrics wrote event.span / event.log / event.metric for a caller
-// that did not exist. Nothing in the server constructed either. The only live
-// ingest path was cloud's embedded otelcol, which writes the OTLP-fork tables
-// the event-plane readers do not read — so the plane the query builders read was
-// only ever filled by hand.
+// that did not exist. Nothing in the server constructed either.
 //
 // This service is that missing edge and nothing else: three transports, three
 // writers, one factory.Service. It owns no decoding, no row building and no SQL
 // — each of those belongs to exactly one package already.
 //
+// It is also the ONLY ingest edge, and an embedding host must not carry a
+// second one. What a consumer gets here is the WHOLE edge: three signals, and
+// the support tables the read plane joins against — event.span_resource /
+// _attribute / _key, event.trace, event.operation, event.log_resource /
+// _attribute / _key / _resource_key, event.series. A partial re-implementation
+// of the write half fills the two headline tables and leaves those empty, and
+// an empty destination is indistinguishable from "no traffic". So a second
+// implementation does not announce itself: it quietly wins a port and drops
+// what it has nowhere to put.
+//
 // Each receiver runs its own zap.Node and therefore its own listener, so the
 // three signals get three addresses. They are separate ports rather than one
 // multiplexed node because a signal must be able to fail, saturate or be
 // firewalled on its own.
+//
+// NOTHING BINDS UNLESS AN ADDRESS SAYS SO. See Config: the address is the
+// switch, it defaults to empty, and linking this package takes no port.
 package zapingest
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"log/slog"
+	"syscall"
 
 	"github.com/hanzoai/o11y/pkg/datastorelogs"
 	"github.com/hanzoai/o11y/pkg/datastoremetrics"
 	"github.com/hanzoai/o11y/pkg/datastoretraces"
-	"github.com/hanzoai/o11y/pkg/errors"
+	o11yerrors "github.com/hanzoai/o11y/pkg/errors"
 	"github.com/hanzoai/o11y/pkg/factory"
 	"github.com/hanzoai/o11y/pkg/telemetrystore"
 	"github.com/hanzoai/o11y/pkg/zaplogreceiver"
@@ -38,16 +51,17 @@ import (
 	"github.com/hanzoai/o11y/pkg/zapreceiver"
 )
 
+// stopper is the only thing Service needs from a receiver once it is up.
+// Holding the three behind it keeps Stop uniform and the signal list open.
+type stopper interface{ Stop() }
+
 // Service is the running ZAP ingest plane.
 type Service struct {
 	config Config
 	logger *slog.Logger
 	store  telemetrystore.TelemetryStore
 
-	spans   *zapreceiver.Receiver
-	logs    *zaplogreceiver.Receiver
-	metrics *zapmetricreceiver.Receiver
-
+	started []stopper
 	stopped chan struct{}
 }
 
@@ -63,64 +77,97 @@ func New(settings factory.ProviderSettings, config Config, store telemetrystore.
 	}
 }
 
-// Start binds the three receivers with the three event-plane writers as their
-// handlers and blocks until Stop. A receiver's New returns as soon as its
-// listener is up, so a bind failure surfaces here rather than in a goroutine.
+// Start binds each CONFIGURED receiver with its event-plane writer as the
+// handler, then blocks until Stop. A receiver's New returns as soon as its
+// listener is up, so a bind failure surfaces here rather than in a goroutine —
+// and it is RETURNED, never swallowed. Losing a port means losing telemetry,
+// which is exactly the failure that must not be quiet.
 func (s *Service) Start(ctx context.Context) error {
-	if !s.config.Enabled {
-		s.logger.InfoContext(ctx, "zap ingest disabled; the event plane has no write side in this process")
+	configured := s.config.signals()
+	if len(configured) == 0 {
+		s.logger.InfoContext(ctx, "zap ingest receives nothing in this process: no signal address is configured (set zapingest spans/logs/metrics to bind)")
 		<-s.stopped
 		return nil
 	}
 
 	conn := s.store.Datastore()
-	var err error
-
-	if s.spans, err = zapreceiver.New(zapreceiver.Config{
-		Listen:  s.config.Spans,
-		OnBatch: datastoretraces.NewWriter(conn).WriteSpans,
-		Logger:  s.logger,
-	}); err != nil {
-		return errors.WrapInternalf(err, errors.CodeInternal, "cannot start zap span receiver on %s", s.config.Spans)
+	// One binder per signal, so the three read identically and a new signal is
+	// one entry rather than another copy of the same error handling.
+	binders := map[string]func(addr string) (stopper, error){
+		"spans": func(addr string) (stopper, error) {
+			return zapreceiver.New(zapreceiver.Config{
+				Listen:  addr,
+				OnBatch: datastoretraces.NewWriter(conn).WriteSpans,
+				Logger:  s.logger,
+			})
+		},
+		"logs": func(addr string) (stopper, error) {
+			return zaplogreceiver.New(zaplogreceiver.Config{
+				Listen:  addr,
+				OnBatch: datastorelogs.NewWriter(conn).WriteLogs,
+				Logger:  s.logger,
+			})
+		},
+		"metrics": func(addr string) (stopper, error) {
+			return zapmetricreceiver.New(zapmetricreceiver.Config{
+				Listen:  addr,
+				OnBatch: datastoremetrics.NewWriter(conn).WriteMetrics,
+				Logger:  s.logger,
+			})
+		},
 	}
 
-	if s.logs, err = zaplogreceiver.New(zaplogreceiver.Config{
-		Listen:  s.config.Logs,
-		OnBatch: datastorelogs.NewWriter(conn).WriteLogs,
-		Logger:  s.logger,
-	}); err != nil {
-		return errors.WrapInternalf(err, errors.CodeInternal, "cannot start zap log receiver on %s", s.config.Logs)
+	for _, sig := range configured {
+		rcv, err := binders[sig.name](sig.addr)
+		if err != nil {
+			// Release whatever already came up. A half-bound edge is worse than
+			// none: it reports partial traffic with a healthy-looking surface.
+			s.stopAll()
+			bindErr := s.bindError(err, sig)
+			// Logged as well as returned: an embedding host is free to ignore a
+			// service error, and telemetry vanishing must not depend on whether
+			// the caller chose to print it.
+			s.logger.ErrorContext(ctx, "zap ingest cannot take its port",
+				slog.String("signal", sig.name), slog.String("addr", sig.addr), slog.Any("error", bindErr))
+			return bindErr
+		}
+		s.started = append(s.started, rcv)
+		s.logger.InfoContext(ctx, "zap ingest receiving", slog.String("signal", sig.name), slog.String("addr", sig.addr))
 	}
 
-	if s.metrics, err = zapmetricreceiver.New(zapmetricreceiver.Config{
-		Listen:  s.config.Metrics,
-		OnBatch: datastoremetrics.NewWriter(conn).WriteMetrics,
-		Logger:  s.logger,
-	}); err != nil {
-		return errors.WrapInternalf(err, errors.CodeInternal, "cannot start zap metric receiver on %s", s.config.Metrics)
-	}
-
-	s.logger.InfoContext(ctx, "zap ingest listening; receivers write the event plane",
-		slog.String("spans", s.config.Spans),
-		slog.String("logs", s.config.Logs),
-		slog.String("metrics", s.config.Metrics),
-	)
-
 	<-s.stopped
 	return nil
 }
 
+// bindError turns a failed listen into an error that names what it lost to.
+// "address already in use" on its own sends you hunting for a stale process;
+// the holder's identity is what reveals that the winner is a second ingest
+// implementation inside this very binary.
+//
+// The message is built into the CAUSE, not into the wrapper. o11y's
+// errors.base.Error() returns the wrapped cause's text and drops the wrapper's
+// own message, so a message passed to Wrap*f is retained on the value but
+// never printed by anything that renders err.Error() — a diagnostic that
+// exists and is invisible. Folding it into the cause makes it print, and the
+// %w keeps errors.Is(err, syscall.EADDRINUSE) working for callers.
+func (s *Service) bindError(err error, sig signal) error {
+	if !errors.Is(err, syscall.EADDRINUSE) {
+		cause := fmt.Errorf("zapingest cannot start the %s receiver on %s: %w", sig.name, sig.addr, err)
+		return o11yerrors.WrapInternalf(cause, o11yerrors.CodeInternal, "%s", cause)
+	}
+	held := holderOf(sig.addr)
+	if held == "" {
+		held = "an unidentified process"
+	}
+	cause := fmt.Errorf(
+		"zapingest cannot bind %s on %s: already held by %s. One ingest edge, not two — remove the other listener rather than moving this one: %w",
+		sig.name, sig.addr, held, err)
+	return o11yerrors.WrapInternalf(cause, o11yerrors.CodeInternal, "%s", cause)
+}
+
 // Stop closes every listener that came up and releases Start.
 func (s *Service) Stop(_ context.Context) error {
-	if s.spans != nil {
-		s.spans.Stop()
-	}
-	if s.logs != nil {
-		s.logs.Stop()
-	}
-	if s.metrics != nil {
-		s.metrics.Stop()
-	}
+	s.stopAll()
 	select {
 	case <-s.stopped:
 	default:
@@ -128,3 +175,10 @@ func (s *Service) Stop(_ context.Context) error {
 	}
 	return nil
 }
+
+func (s *Service) stopAll() {
+	for _, r := range s.started {
+		r.Stop()
+	}
+	s.started = nil
+}

pkg/zapingest/zapingest_test.go

@@ -0,0 +1,172 @@
+package zapingest
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net"
+	"os"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/hanzoai/o11y/pkg/factory"
+)
+
+// The default must bind NOTHING. This is the regression that caused the
+// incident: a linked-in library that seizes :4317-:4319 as a side effect of
+// being imported races the host for the same sockets.
+func TestDefaultConfigBindsNothing(t *testing.T) {
+	c, ok := newConfig().(Config)
+	if !ok {
+		t.Fatalf("newConfig() is not a Config")
+	}
+	if got := c.signals(); len(got) != 0 {
+		t.Fatalf("default config receives %v; a linked library must take no port", got)
+	}
+	for name, addr := range map[string]string{"spans": c.Spans, "logs": c.Logs, "metrics": c.Metrics} {
+		if addr != "" {
+			t.Errorf("default %s address is %q; want empty", name, addr)
+		}
+	}
+	if err := c.Validate(); err != nil {
+		t.Errorf("default config must validate, got %v", err)
+	}
+}
+
+// A Service on the default config starts, binds nothing, and stops cleanly —
+// the shape every process that merely links o11y gets.
+func TestStartWithNoAddressBindsNothing(t *testing.T) {
+	s := New(settings(), Config{}, nil) // nil store: never dereferenced when nothing binds
+	done := make(chan error, 1)
+	go func() { done <- s.Start(context.Background()) }()
+
+	time.Sleep(50 * time.Millisecond)
+	if len(s.started) != 0 {
+		t.Fatalf("started %d receivers on an empty config", len(s.started))
+	}
+	if err := s.Stop(context.Background()); err != nil {
+		t.Fatalf("Stop: %v", err)
+	}
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Start returned %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Start did not return after Stop")
+	}
+}
+
+func TestSignalsOmitsUnconfigured(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		cfg  Config
+		want []string
+	}{
+		{"none", Config{}, nil},
+		{"logs only", Config{Logs: ":4318"}, []string{"logs"}},
+		{"spans+metrics", Config{Spans: ":4317", Metrics: ":4319"}, []string{"spans", "metrics"}},
+		{"all", Config{Spans: ":4317", Logs: ":4318", Metrics: ":4319"}, []string{"spans", "logs", "metrics"}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var got []string
+			for _, s := range tc.cfg.signals() {
+				got = append(got, s.name)
+			}
+			if strings.Join(got, ",") != strings.Join(tc.want, ",") {
+				t.Fatalf("signals()=%v want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestValidate(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		cfg     Config
+		wantErr string
+	}{
+		{"empty ok", Config{}, ""},
+		{"partial ok", Config{Logs: "0.0.0.0:4318"}, ""},
+		{"bad addr", Config{Spans: "4317"}, "not host:port"},
+		{"duplicate", Config{Spans: "0.0.0.0:4317", Logs: "0.0.0.0:4317"}, "both use"},
+		// An unconfigured signal must not collide with another unconfigured
+		// one: three empty strings are equal but none of them is an address.
+		{"two empty are not a collision", Config{Spans: "0.0.0.0:4317"}, ""},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.cfg.Validate()
+			if tc.wantErr == "" {
+				if err != nil {
+					t.Fatalf("want ok, got %v", err)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), tc.wantErr) {
+				t.Fatalf("want error containing %q, got %v", tc.wantErr, err)
+			}
+		})
+	}
+}
+
+// A bind collision must fail fast AND name the holder. Here the holder is this
+// test process itself — the same shape as two ingest implementations linked
+// into one binary, which is the case the message must call out by name.
+func TestBindCollisionNamesTheHolder(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	defer ln.Close()
+	addr := ln.Addr().String()
+
+	s := New(settings(), Config{Logs: addr}, nil)
+	sig := signal{name: "logs", addr: addr}
+	err = s.bindError(fmt.Errorf("zaplogreceiver: zap.Node start: listen tcp %s: bind: %w", addr, syscall.EADDRINUSE), sig)
+	if err == nil {
+		t.Fatal("bindError returned nil for an in-use address")
+	}
+	msg := err.Error()
+	for _, want := range []string{addr, "logs", "One ingest edge, not two"} {
+		if !strings.Contains(msg, want) {
+			t.Errorf("bind error missing %q:\n%s", want, msg)
+		}
+	}
+	// procfs is how the holder is identified; where it exists, the holder is us.
+	if _, statErr := os.Stat("/proc/net/tcp"); statErr == nil {
+		if !strings.Contains(msg, fmt.Sprintf("pid %d", os.Getpid())) {
+			t.Errorf("holder should be this process (pid %d):\n%s", os.Getpid(), msg)
+		}
+		if !strings.Contains(msg, "THIS SAME PROCESS") {
+			t.Errorf("an in-process collision must say so:\n%s", msg)
+		}
+	}
+}
+
+// holderOf finds the process listening on an address, and reports nothing for
+// an address nobody holds.
+func TestHolderOf(t *testing.T) {
+	if _, err := os.Stat("/proc/net/tcp"); err != nil {
+		t.Skip("no procfs")
+	}
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	addr := ln.Addr().String()
+	if got := holderOf(addr); !strings.Contains(got, fmt.Sprintf("pid %d", os.Getpid())) {
+		t.Errorf("holderOf(%s)=%q, want this pid", addr, got)
+	}
+	ln.Close()
+	// Give the socket a moment to leave LISTEN.
+	time.Sleep(50 * time.Millisecond)
+	if got := holderOf(addr); got != "" {
+		t.Errorf("holderOf(%s) after close = %q, want empty", addr, got)
+	}
+}
+
+func settings() factory.ProviderSettings {
+	return factory.ProviderSettings{Logger: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))}
+}