complexity-reduction 2/4: destination + factory-seam coverage (P2/P3/L/C/D)

cmd/clickhouse_protobuflist/clickhouse_protobuflist.go

@@ -12,6 +12,13 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+// marshalFn is the proto-marshaller seam encodeLengthDelimitedProtobufList
+// uses. Defaults to proto.Marshal; tests can swap in a deliberately
+// failing marshaller to exercise the error-return branch (which is
+// unreachable with proto.Marshal on a valid struct). Production code
+// never mutates this.
+var marshalFn = proto.Marshal
+
 func encodeLengthDelimitedProtobufList(r *clickhouse_protolist.Envelope_Record) (result []byte, err error) {
 
 	// for _, record := range e.Rows {
@@ -20,7 +27,7 @@ func encodeLengthDelimitedProtobufList(r *clickhouse_protolist.Envelope_Record)
 	// 		return nil, fmt.Errorf("error marshaling Record: %v", err)
 	// 	}
 
-	recordBytes, err := proto.Marshal(r)
+	recordBytes, err := marshalFn(r)
 	if err != nil {
 		return nil, fmt.Errorf("error marshaling Record: %v", err)
 	}

cmd/clickhouse_protobuflist/clickhouse_protobuflist_test.go

@@ -2,12 +2,14 @@ package main
 
 import (
 	"bytes"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
 	"testing"
 
 	"github.com/randomizedcoder/xtcp2/pkg/clickhouse_protolist"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestEncodeLengthDelimitedProtobufList(t *testing.T) {
@@ -106,3 +108,45 @@ func TestRunMain_writeEnvelopeError(t *testing.T) {
 		t.Errorf("rc = %d, want 1", rc)
 	}
 }
+
+// TestEncodeLengthDelimitedProtobufList_marshalErr swaps the
+// marshalFn seam for a failing fake to cover the proto-marshal
+// error-return branch (unreachable from production where proto.Marshal
+// can't fail on this struct).
+func TestEncodeLengthDelimitedProtobufList_marshalErr(t *testing.T) {
+	orig := marshalFn
+	marshalFn = func(_ proto.Message) ([]byte, error) {
+		return nil, fmt.Errorf("synthetic marshal err")
+	}
+	defer func() { marshalFn = orig }()
+
+	_, err := encodeLengthDelimitedProtobufList(&clickhouse_protolist.Envelope_Record{MyUint32: 1})
+	if err == nil {
+		t.Fatal("expected err from failing marshaller")
+	}
+	if !strings.Contains(err.Error(), "error marshaling Record") {
+		t.Errorf("err = %q, want substring 'error marshaling Record'", err)
+	}
+}
+
+// TestRunMain_encodeError exercises runMain's error-handling branch
+// when encodeLengthDelimitedProtobufList fails. rc=1 means the
+// encode-error branch fired.
+func TestRunMain_encodeError(t *testing.T) {
+	orig := marshalFn
+	marshalFn = func(_ proto.Message) ([]byte, error) {
+		return nil, fmt.Errorf("synthetic")
+	}
+	defer func() { marshalFn = orig }()
+
+	dir := t.TempDir()
+	out := filepath.Join(dir, "out.bin")
+	var stderr bytes.Buffer
+	rc := runMain([]string{"-filename", out}, &bytes.Buffer{}, &stderr)
+	if rc != 1 {
+		t.Errorf("rc = %d, want 1 (encode error)", rc)
+	}
+	if !strings.Contains(stderr.String(), "Error encoding") {
+		t.Errorf("stderr = %q, want substring 'Error encoding'", stderr.String())
+	}
+}

cmd/kafka_to_clickhouse/kafka_to_clickhouse.go

@@ -152,7 +152,10 @@ func runMain(ctx context.Context, args []string, stdout, stderr io.Writer) int {
 		// bounded by a 5s timeout so a wedged broker doesn't block
 		// shutdown indefinitely.
 		defer func() {
-			flushCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			// Derive flushCtx from the caller's ctx so cancellation
+			// propagates correctly; cap at 5s so a wedged broker
+			// doesn't block teardown indefinitely.
+			flushCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 			defer cancel()
 			if err := kClient.Flush(flushCtx); err != nil {
 				fmt.Fprintf(stderr, "kgo Flush on exit: %v\n", err)

cmd/kafka_to_clickhouse/kafka_to_clickhouse_test.go

@@ -160,6 +160,37 @@ func TestRunMain_fileMode(t *testing.T) {
 	}
 }
 
+// TestRunMain_kafkaMode drives runMain through the kafka=true branch
+// using a bogus broker + a short-lived ctx so destKafka's wg.Wait
+// (which blocks on Produce's callback) is unblocked by ctx
+// cancellation rather than waiting for franz-go's connection retries.
+// Covers the `if c.kafka { InitDestKafka + defer flush+close }` block
+// inside runMain that the file-mode test skips.
+func TestRunMain_kafkaMode(t *testing.T) {
+	prevClient := kClient
+	t.Cleanup(func() {
+		if kClient != prevClient && kClient != nil {
+			kClient.Close()
+		}
+		kClient = prevClient
+	})
+	ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
+	defer cancel()
+	dir := t.TempDir()
+	out := filepath.Join(dir, "out.bin")
+	var stderr bytes.Buffer
+	rc := runMain(ctx, []string{
+		"-filename", out, "-values", "1",
+		"-kafka=true", "-broker", "127.0.0.1:1", // refused
+		"-loops", "1", "-loopsSleep", "1ms",
+	}, &bytes.Buffer{}, &stderr)
+	// rc=0 — runMain swallows broker errors; the assertion is just
+	// "does not hang, returns cleanly".
+	if rc != 0 {
+		t.Errorf("rc = %d, want 0 (kafka mode w/ unreachable broker)", rc)
+	}
+}
+
 func TestInitDestKafka_noopWhenDisabled(t *testing.T) {
 	// kafka=false → InitDestKafka short-circuits without trying to create
 	// a client.

cmd/xtcp2_kafka_client/xtcp2_kafka_client.go

@@ -65,9 +65,17 @@ func runMain(ctx context.Context, args []string, stderr io.Writer) int {
 	return 0
 }
 
+// kafkaFetcher is the surface pollLoop needs from a Kafka consumer
+// client. Lifting it to an interface lets tests drive pollLoop's
+// happy-path EachRecord closure without a real broker. *kgo.Client
+// satisfies this interface.
+type kafkaFetcher interface {
+	PollFetches(ctx context.Context) kgo.Fetches
+}
+
 // pollLoop is the Kafka consume body. Extracted so test code can call it
 // against a fake client (with a pre-canceled ctx for a quick exit).
-func pollLoop(ctx context.Context, cl *kgo.Client) {
+func pollLoop(ctx context.Context, cl kafkaFetcher) {
 	for i := 0; ; i++ {
 		select {
 		case <-ctx.Done():

cmd/xtcp2_kafka_client/xtcp2_kafka_client_test.go

@@ -162,3 +162,99 @@ func TestPollLoop_fetchErrorsThenCancel(t *testing.T) {
 		t.Fatal("pollLoop did not exit after cancel")
 	}
 }
+
+// fakeFetcher implements the kafkaFetcher interface so pollLoop can be
+// driven with synthetic records — exercises the EachRecord closure
+// body that broker-bound tests can't reach without real kafka.
+type fakeFetcher struct {
+	fetches  []kgo.Fetches
+	calls    int
+	onCancel context.CancelFunc
+}
+
+func (f *fakeFetcher) PollFetches(_ context.Context) kgo.Fetches {
+	f.calls++
+	if f.calls > len(f.fetches) {
+		if f.onCancel != nil {
+			f.onCancel()
+		}
+		return kgo.Fetches{}
+	}
+	return f.fetches[f.calls-1]
+}
+
+func makeFetchWithRecord(value []byte) kgo.Fetches {
+	return kgo.Fetches{
+		{
+			Topics: []kgo.FetchTopic{
+				{
+					Topic: "test-topic",
+					Partitions: []kgo.FetchPartition{
+						{Records: []*kgo.Record{{Value: value}}},
+					},
+				},
+			},
+		},
+	}
+}
+
+// TestPollLoop_eachRecordClosureFires drives pollLoop with one
+// synthetic Fetch containing a single valid Confluent-framed record.
+// The EachRecord closure (processRecord call) fires, then the second
+// PollFetches call signals the fake to cancel ctx.
+func TestPollLoop_eachRecordClosureFires(t *testing.T) {
+	value := make([]byte, KafkaHeaderSizeCst+1)
+	value[0] = 0x00 // magic
+	// schemaID bytes [1:5] = 0; length byte [5] = 0 → empty proto
+	// envelope; processRecord parses it successfully.
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	fake := &fakeFetcher{
+		fetches:  []kgo.Fetches{makeFetchWithRecord(value)},
+		onCancel: cancel,
+	}
+	done := make(chan struct{})
+	go func() {
+		pollLoop(ctx, fake)
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("pollLoop did not exit after fake fetcher exhausted")
+	}
+	if fake.calls < 1 {
+		t.Errorf("expected ≥1 PollFetches call; got %d", fake.calls)
+	}
+}
+
+// TestPollLoop_fakeFetcherErrors drives pollLoop with a Fetches that
+// surfaces an error via FetchPartition.Err, then exhausts to cancel.
+// Exercises the `if errs := fetches.Errors(); ...` branch with a
+// non-empty Errors() result.
+func TestPollLoop_fakeFetcherErrors(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	errFetch := kgo.Fetches{
+		{Topics: []kgo.FetchTopic{
+			{Topic: "test-topic", Partitions: []kgo.FetchPartition{
+				{Err: errors.New("fetch err")},
+			}},
+		}},
+	}
+	fake := &fakeFetcher{
+		fetches:  []kgo.Fetches{errFetch},
+		onCancel: cancel,
+	}
+	done := make(chan struct{})
+	go func() {
+		pollLoop(ctx, fake)
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("pollLoop did not exit on fake-fetcher exhaust")
+	}
+}