protocol: embed-safe Broker.Serve() for the cloud in-process fold

Startup() was a blocking accept loop that log.Panic/os.Exit(1)d on failure and
never stored its listener, so a host could not run the Kafka adaptor in-process
nor stop it cleanly. Add Broker.Serve() error — the embed-safe form that returns
errors instead of exiting, stores the listener, and returns nil on a clean
Shutdown (which now closes ShutDownSignal + the listener so Accept unblocks).
Startup() becomes a thin fatal wrapper over Serve() so standalone main.go is
unchanged. The os import is no longer needed.

Test proves Serve() returns an error (never exits) on an unreachable PubSub.

protocol/broker.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"io"
 	"net"
-	"os"
 	"runtime/debug"
 	"sync"
 	"time"
@@ -20,6 +19,7 @@ type Broker struct {
 	Config         *types.Configuration
 	PubSub         *pubsub.Client
 	ShutDownSignal chan bool
+	listener       net.Listener
 	partitionMu    sync.Map // map[string]*sync.Mutex keyed by "topic-partition"
 }
 
@@ -130,36 +130,52 @@ func NewBroker(config *types.Configuration) *Broker {
 	}
 }
 
-// Startup initializes the broker, connects to NATS, and listens for incoming Kafka client connections
+// Startup initializes the broker and blocks serving Kafka clients. It is the
+// standalone entrypoint (main.go): any startup failure is fatal.
 func (b *Broker) Startup() {
+	if err := b.Serve(); err != nil {
+		log.Panic("Hanzo Kafka failed: %v", err)
+	}
+}
+
+// Serve is the embed-safe form of Startup: it connects to PubSub, starts the
+// admin server, and runs the Kafka accept loop, RETURNING errors instead of
+// exiting the process so a host binary (hanzoai/cloud) can run the adaptor
+// in-process. It stores the listener so Shutdown can stop it; a clean Shutdown
+// closes ShutDownSignal + the listener, so Accept fails and Serve returns nil.
+// Run it in a goroutine when embedding.
+func (b *Broker) Serve() error {
 	var err error
 
 	b.PubSub, err = pubsub.NewClient(b.Config.PubSubUrl)
 	if err != nil {
-		log.Panic("Failed to connect to Hanzo PubSub: %v", err)
+		return fmt.Errorf("connect pubsub: %w", err)
 	}
 
-	err = b.PubSub.EnsureOffsetBucket()
-	if err != nil {
-		log.Panic("Failed to ensure offset bucket: %v", err)
+	if err = b.PubSub.EnsureOffsetBucket(); err != nil {
+		return fmt.Errorf("ensure offset bucket: %w", err)
 	}
 
 	b.StartAdmin()
 
-	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", b.Config.BrokerPort))
+	ln, err := net.Listen("tcp", fmt.Sprintf(":%d", b.Config.BrokerPort))
 	if err != nil {
-		log.Error("Error starting server: %v", err)
-		os.Exit(1)
+		return fmt.Errorf("listen :%d: %w", b.Config.BrokerPort, err)
 	}
-	defer listener.Close()
+	b.listener = ln
 
 	log.Info("Hanzo Kafka listening on port %d (PubSub: %s)", b.Config.BrokerPort, b.Config.PubSubUrl)
 
 	for {
-		conn, err := listener.Accept()
+		conn, err := ln.Accept()
 		if err != nil {
-			log.Error("Error accepting connection: %v", err)
-			continue
+			select {
+			case <-b.ShutDownSignal:
+				return nil
+			default:
+				log.Error("Error accepting connection: %v", err)
+				continue
+			}
 		}
 		go b.HandleConnection(conn)
 	}
@@ -229,6 +245,9 @@ func (b *Broker) safeHandle(h APIKeyHandler, req types.Request) (response []byte
 // Shutdown gracefully shuts down the broker
 func (b *Broker) Shutdown() {
 	close(b.ShutDownSignal)
+	if b.listener != nil {
+		b.listener.Close()
+	}
 	if b.PubSub != nil {
 		b.PubSub.Close()
 	}

protocol/serve_test.go

@@ -0,0 +1,24 @@
+package protocol
+
+import (
+	"testing"
+
+	"github.com/hanzoai/stream/types"
+)
+
+// TestServeReturnsErrorOnUnreachablePubSub proves the embed-safety property: on
+// a startup failure Serve RETURNS an error rather than calling log.Panic /
+// os.Exit, so a host binary (hanzoai/cloud) can run the adaptor in-process
+// without the whole process dying. pubsub.NewClient wraps nats.Connect, which
+// fails fast against an unreachable URL, so Serve returns before it ever binds
+// the Kafka port or starts the admin server (AdminPort 0 disables it anyway).
+func TestServeReturnsErrorOnUnreachablePubSub(t *testing.T) {
+	b := NewBroker(&types.Configuration{
+		PubSubUrl:  "nats://127.0.0.1:1", // nothing listens on port 1
+		BrokerPort: 0,
+		AdminPort:  0,
+	})
+	if err := b.Serve(); err == nil {
+		t.Fatal("Serve returned nil for an unreachable PubSub; want an error")
+	}
+}