fix(kafka/sarama_broker): drain async producer Successes and Errors (#72) (#120)

The async Sarama producer was configured with Return.Successes=true and
Return.Errors=true but no goroutine drained those channels, so they
filled up and the producer blocked indefinitely. Two drainer goroutines
now consume both channels for the producer's lifetime. Errors are
logged and counted via the existing arcade_kafka_produce_errors_total
metric. Closes F-014.

Author: galt-tr

kafka/sarama_broker.go

@@ -4,23 +4,49 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 
 	"github.com/IBM/sarama"
+	"go.uber.org/zap"
+
+	"github.com/bsv-blockchain/arcade/metrics"
 )
 
 // saramaBroker is the production Broker backed by IBM Sarama. It owns both a
 // sync and async producer so Send/SendAsync/SendBatch can pick the appropriate
 // one without the caller caring.
+//
+// The async producer is configured with Return.Successes=true and
+// Return.Errors=true. Sarama routes every produced message's outcome onto
+// those channels, and if no goroutine drains them they fill up and the
+// producer blocks indefinitely on Input(). To keep SendAsync non-blocking,
+// the broker spawns two drain goroutines for the producer's lifetime: one
+// discards successes (the SendAsync caller already counted the produce in
+// the Producer wrapper) and one logs/counts errors via metrics.
+// Close() waits for both drainers to exit after closing the underlying
+// async producer, which closes both channels in turn.
 type saramaBroker struct {
 	syncProducer  sarama.SyncProducer
 	asyncProducer sarama.AsyncProducer
 	brokers       []string
 	consumerGroup string
+
+	logger     *zap.Logger
+	drainersWG sync.WaitGroup
 }
 
 // NewSaramaBroker constructs a Sarama-backed Broker with sensible defaults
-// (WaitForAll on sync, WaitForLocal on async, 5 retries).
+// (WaitForAll on sync, WaitForLocal on async, 5 retries). Errors from the
+// async producer are logged via the package-global zap logger; callers that
+// want a custom logger should use NewSaramaBrokerWithLogger.
 func NewSaramaBroker(brokers []string, consumerGroup string) (Broker, error) {
+	return NewSaramaBrokerWithLogger(brokers, consumerGroup, nil)
+}
+
+// NewSaramaBrokerWithLogger is like NewSaramaBroker but lets callers inject a
+// zap logger for async-producer error logging. A nil logger falls back to
+// zap.NewNop() so the broker is always safe to construct.
+func NewSaramaBrokerWithLogger(brokers []string, consumerGroup string, logger *zap.Logger) (Broker, error) {
 	syncCfg := sarama.NewConfig()
 	syncCfg.Producer.RequiredAcks = sarama.WaitForAll
 	syncCfg.Producer.Retry.Max = 5
@@ -44,12 +70,69 @@ func NewSaramaBroker(brokers []string, consumerGroup string) (Broker, error) {
 		return nil, fmt.Errorf("creating async producer: %w", err)
 	}
 
-	return &saramaBroker{
-		syncProducer:  syncProducer,
-		asyncProducer: asyncProducer,
+	return newSaramaBrokerFromProducers(syncProducer, asyncProducer, brokers, consumerGroup, logger), nil
+}
+
+// newSaramaBrokerFromProducers wires the broker around already-constructed
+// sync and async producers. Extracted so tests can substitute Sarama mocks
+// without standing up a real Kafka cluster. It also starts the async-producer
+// drainer goroutines, which is the only place those should be spawned —
+// duplicating that elsewhere would race for ownership of Successes/Errors.
+func newSaramaBrokerFromProducers(
+	sync sarama.SyncProducer,
+	async sarama.AsyncProducer,
+	brokers []string,
+	consumerGroup string,
+	logger *zap.Logger,
+) *saramaBroker {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	b := &saramaBroker{
+		syncProducer:  sync,
+		asyncProducer: async,
 		brokers:       brokers,
 		consumerGroup: consumerGroup,
-	}, nil
+		logger:        logger,
+	}
+	b.startAsyncDrainers()
+	return b
+}
+
+// startAsyncDrainers spawns the two goroutines that consume the async
+// producer's Successes and Errors channels for the producer's lifetime.
+// They return when the underlying channels close, which Sarama does as
+// part of asyncProducer.Close(). Close() then waits on drainersWG so the
+// broker does not return from Close until both goroutines have exited —
+// otherwise a test or a process restart could observe partial shutdown.
+func (b *saramaBroker) startAsyncDrainers() {
+	b.drainersWG.Add(2)
+	go func() {
+		defer b.drainersWG.Done()
+		// Successes are already accounted for by Producer.SendAsync at
+		// enqueue time, so we just discard them here. Draining is the
+		// whole point — a full Successes channel blocks Input().
+		successes := b.asyncProducer.Successes()
+		for {
+			if _, ok := <-successes; !ok {
+				return
+			}
+		}
+	}()
+	go func() {
+		defer b.drainersWG.Done()
+		for produceErr := range b.asyncProducer.Errors() {
+			topic := ""
+			if produceErr.Msg != nil {
+				topic = produceErr.Msg.Topic
+			}
+			metrics.KafkaProduceErrors.WithLabelValues(topic).Inc()
+			b.logger.Error("async kafka produce failed",
+				zap.String("topic", topic),
+				zap.Error(produceErr.Err),
+			)
+		}
+	}()
 }
 
 func (b *saramaBroker) Send(_ context.Context, topic, key string, value []byte) error {
@@ -137,9 +220,13 @@ func (b *saramaBroker) Close() error {
 	if err := b.syncProducer.Close(); err != nil {
 		errs = append(errs, err)
 	}
+	// Closing the async producer closes its Successes/Errors channels once
+	// in-flight messages have been flushed, which is what unblocks the
+	// drain goroutines below.
 	if err := b.asyncProducer.Close(); err != nil {
 		errs = append(errs, err)
 	}
+	b.drainersWG.Wait()
 	if len(errs) > 0 {
 		return fmt.Errorf("closing producers: %v", errs)
 	}

kafka/sarama_broker_test.go

@@ -0,0 +1,174 @@
+package kafka
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/IBM/sarama"
+	"github.com/IBM/sarama/mocks"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest"
+)
+
+// fakeSyncProducer satisfies sarama.SyncProducer so the saramaBroker can be
+// constructed without a real Kafka cluster. The async-channel-drainer fix
+// only exercises the async producer side, but Close() drives both producers.
+type fakeSyncProducer struct {
+	closed atomic.Bool
+}
+
+func (f *fakeSyncProducer) SendMessage(*sarama.ProducerMessage) (int32, int64, error) {
+	return 0, 0, nil
+}
+
+func (f *fakeSyncProducer) SendMessages([]*sarama.ProducerMessage) error { return nil }
+
+func (f *fakeSyncProducer) Close() error { f.closed.Store(true); return nil }
+
+func (f *fakeSyncProducer) AbortTxn() error { return nil }
+
+func (f *fakeSyncProducer) AddMessageToTxn(*sarama.ConsumerMessage, string, *string) error {
+	return nil
+}
+
+func (f *fakeSyncProducer) AddOffsetsToTxn(map[string][]*sarama.PartitionOffsetMetadata, string) error {
+	return nil
+}
+
+func (f *fakeSyncProducer) BeginTxn() error { return nil }
+
+func (f *fakeSyncProducer) CommitTxn() error { return nil }
+
+func (f *fakeSyncProducer) IsTransactional() bool { return false }
+
+func (f *fakeSyncProducer) TxnStatus() sarama.ProducerTxnStatusFlag { return 0 }
+
+func (f *fakeSyncProducer) TxnAddOffsetsToTxn(map[string][]*sarama.PartitionOffsetMetadata, string) error {
+	return nil
+}
+
+// newAsyncProducerMock builds a Sarama mock async producer with the same
+// Return.Successes/Return.Errors flags the production broker uses, so the
+// drainer goroutines see the same channel behavior they would in a real
+// deployment.
+func newAsyncProducerMock(t *testing.T) *mocks.AsyncProducer {
+	t.Helper()
+	cfg := sarama.NewConfig()
+	cfg.Producer.Return.Successes = true
+	cfg.Producer.Return.Errors = true
+	return mocks.NewAsyncProducer(t, cfg)
+}
+
+// TestSaramaBroker_AsyncDrainer_DoesNotBlockOnSuccesses sends N messages
+// through SendAsync. Without a Successes drainer the mock's success channel
+// (capacity = ChannelBufferSize, default 256) would fill and the next send
+// would block on Input(). N is chosen well above that buffer to make the
+// regression mode unambiguous.
+func TestSaramaBroker_AsyncDrainer_DoesNotBlockOnSuccesses(t *testing.T) {
+	mp := newAsyncProducerMock(t)
+	const n = 1024
+	for range n {
+		mp.ExpectInputAndSucceed()
+	}
+
+	b := newSaramaBrokerFromProducers(&fakeSyncProducer{}, mp, nil, "", zaptest.NewLogger(t))
+
+	done := make(chan error, 1)
+	go func() {
+		ctx := context.Background()
+		for i := 0; i < n; i++ {
+			if err := b.SendAsync(ctx, "tx.validated", "k", []byte("v")); err != nil {
+				done <- err
+				return
+			}
+		}
+		done <- nil
+	}()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("SendAsync returned error: %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("SendAsync blocked — Successes channel was not drained")
+	}
+
+	if err := b.Close(); err != nil {
+		t.Fatalf("Close returned error: %v", err)
+	}
+}
+
+// TestSaramaBroker_AsyncDrainer_HandlesErrors verifies that messages routed
+// to the Errors channel are drained (not blocking the producer) and logged
+// without panicking. The mock fails every input, so without an Errors
+// drainer the channel would fill and SendAsync would deadlock.
+func TestSaramaBroker_AsyncDrainer_HandlesErrors(t *testing.T) {
+	mp := newAsyncProducerMock(t)
+	const n = 512
+	produceErr := errors.New("simulated produce failure")
+	for range n {
+		mp.ExpectInputAndFail(produceErr)
+	}
+
+	b := newSaramaBrokerFromProducers(&fakeSyncProducer{}, mp, nil, "", zaptest.NewLogger(t))
+
+	done := make(chan error, 1)
+	go func() {
+		ctx := context.Background()
+		for i := 0; i < n; i++ {
+			if err := b.SendAsync(ctx, "tx.validated", "k", []byte("v")); err != nil {
+				done <- err
+				return
+			}
+		}
+		done <- nil
+	}()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("SendAsync returned error: %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("SendAsync blocked — Errors channel was not drained")
+	}
+
+	if err := b.Close(); err != nil {
+		t.Fatalf("Close returned error: %v", err)
+	}
+}
+
+// TestSaramaBroker_Close_WaitsForDrainers asserts that Close() blocks until
+// the drainer goroutines exit. After Close returns, no goroutine should
+// still be reading from Successes/Errors, which we approximate by checking
+// that the broker's WaitGroup has zero counter (drainersWG.Wait() returns
+// immediately on a second call).
+func TestSaramaBroker_Close_WaitsForDrainers(t *testing.T) {
+	mp := newAsyncProducerMock(t)
+	mp.ExpectInputAndSucceed()
+
+	b := newSaramaBrokerFromProducers(&fakeSyncProducer{}, mp, nil, "", zap.NewNop())
+
+	if err := b.SendAsync(context.Background(), "t", "k", []byte("v")); err != nil {
+		t.Fatalf("SendAsync: %v", err)
+	}
+	if err := b.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+
+	// A second Wait must be a no-op; if it blocks the drainers leaked.
+	waitDone := make(chan struct{})
+	go func() {
+		b.drainersWG.Wait()
+		close(waitDone)
+	}()
+	select {
+	case <-waitDone:
+	case <-time.After(time.Second):
+		t.Fatal("drainer goroutines leaked past Close()")
+	}
+}