coverage-sweep 5/6: bug-fix wave A (Bug 27–51 + regression tests)

cmd/clickhouse_http_insert_protobuflist/clickhouse_http_insert_protobuflist.go

@@ -179,14 +179,24 @@ func writeDataToFile(ctx context.Context, filename string, data []byte) error {
 var ErrClickHouseHTTPPost = errors.New("clickhouse http post failed")
 
 func insertIntoCH(ctx context.Context, c config, binaryData []byte) error {
-	return insertIntoCHAt(ctx, http.DefaultClient, "http://"+c.connectStr, binaryData)
+	return insertIntoCHAt(ctx, http.DefaultClient, "http://"+c.connectStr, binaryData, c.envelope)
 }
 
 // insertIntoCHAt POSTs a binary protobuf payload to ClickHouse's HTTP
 // endpoint. Extracted so tests can drive it against httptest.Server (the
 // production wrapper picks the baseURL from config.connectStr).
-func insertIntoCHAt(ctx context.Context, client *http.Client, baseURL string, binaryData []byte) error {
-	clickhouseURL := baseURL + "/?query=INSERT%20INTO%20clickhouse_protolist.clickhouse_protolist%20FORMAT%20Protobuf&format_schema=clickhouse_protolist.proto:clickhouse_protolist.v1.Record"
+//
+// useEnvelope picks the ClickHouse format: ProtobufList for envelope mode
+// (a single Envelope with a repeated Rows field), Protobuf for the
+// length-prefixed per-row mode. The previous code hardwired FORMAT
+// Protobuf regardless, so the default envelope=true path mailed an
+// Envelope at a Record schema — ClickHouse rejected every insert.
+func insertIntoCHAt(ctx context.Context, client *http.Client, baseURL string, binaryData []byte, useEnvelope bool) error {
+	format := "Protobuf"
+	if useEnvelope {
+		format = "ProtobufList"
+	}
+	clickhouseURL := baseURL + "/?query=INSERT%20INTO%20clickhouse_protolist.clickhouse_protolist%20FORMAT%20" + format + "&format_schema=clickhouse_protolist.proto:clickhouse_protolist.v1.Record"
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, clickhouseURL, bytes.NewReader(binaryData))
 	if err != nil {

cmd/clickhouse_http_insert_protobuflist/clickhouse_http_insert_protobuflist_test.go

@@ -135,7 +135,7 @@ func TestInsertIntoCHAt_success(t *testing.T) {
 		w.WriteHeader(http.StatusOK)
 	}))
 	defer srv.Close()
-	if err := insertIntoCHAt(t.Context(), srv.Client(), srv.URL, []byte("payload")); err != nil {
+	if err := insertIntoCHAt(t.Context(), srv.Client(), srv.URL, []byte("payload"), true); err != nil {
 		t.Errorf("err = %v, want nil", err)
 	}
 }
@@ -146,7 +146,7 @@ func TestInsertIntoCHAt_serverError(t *testing.T) {
 		_, _ = w.Write([]byte("oops")) //nolint:errcheck // test plumbing
 	}))
 	defer srv.Close()
-	err := insertIntoCHAt(t.Context(), srv.Client(), srv.URL, []byte("payload"))
+	err := insertIntoCHAt(t.Context(), srv.Client(), srv.URL, []byte("payload"), true)
 	if err == nil {
 		t.Error("500 should produce error")
 	}
@@ -159,15 +159,41 @@ func TestInsertIntoCHAt_connRefused(t *testing.T) {
 	srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
 	url := srv.URL
 	srv.Close()
-	if err := insertIntoCHAt(t.Context(), http.DefaultClient, url, []byte("payload")); err == nil {
+	if err := insertIntoCHAt(t.Context(), http.DefaultClient, url, []byte("payload"), true); err == nil {
 		t.Error("conn-refused should produce error")
 	}
 }
 
 func TestInsertIntoCHAt_badURL(t *testing.T) {
 	// Malformed URL forces http.NewRequestWithContext to fail.
-	err := insertIntoCHAt(t.Context(), http.DefaultClient, "://not a url", []byte("p"))
+	err := insertIntoCHAt(t.Context(), http.DefaultClient, "://not a url", []byte("p"), true)
 	if err == nil {
 		t.Error("malformed URL should produce error")
 	}
 }
+
+// Verify the URL contains FORMAT ProtobufList when useEnvelope is true,
+// and FORMAT Protobuf when false — the bug under fix.
+func TestInsertIntoCHAt_formatSelection(t *testing.T) {
+	cases := []struct {
+		useEnvelope bool
+		wantFormat  string
+	}{
+		{true, "ProtobufList"},
+		{false, "Protobuf&"},
+	}
+	for _, tc := range cases {
+		var gotURL string
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			gotURL = r.URL.String()
+			w.WriteHeader(http.StatusOK)
+		}))
+		if err := insertIntoCHAt(t.Context(), srv.Client(), srv.URL, []byte("p"), tc.useEnvelope); err != nil {
+			t.Errorf("useEnvelope=%v: err = %v", tc.useEnvelope, err)
+		}
+		if !strings.Contains(gotURL, "FORMAT%20"+tc.wantFormat) {
+			t.Errorf("useEnvelope=%v: URL = %s, want FORMAT %s", tc.useEnvelope, gotURL, tc.wantFormat)
+		}
+		srv.Close()
+	}
+}

cmd/clickhouse_protobuflist/clickhouse_protobuflist.go

@@ -26,7 +26,14 @@ func encodeLengthDelimitedProtobufList(r *clickhouse_protolist.Envelope_Record)
 	}
 
 	log.Printf("AppendVarint of length:%d", len(recordBytes))
-	protowire.AppendVarint(result, uint64(len(recordBytes)))
+	// protowire.AppendVarint returns the appended slice — the previous
+	// code dropped the return value, so the length prefix was never
+	// actually written. Non-envelope mode emitted raw record bytes
+	// without the length-delim wrapper its name advertises (ClickHouse
+	// readers expecting LengthDelimited misparsed the file). Use the
+	// return value the way encodeLengthDelimitedEnvelope below already
+	// does.
+	result = protowire.AppendVarint(result, uint64(len(recordBytes)))
 
 	result = append(result, recordBytes...)
 

cmd/kafka_to_clickhouse/kafka_to_clickhouse.go

@@ -172,7 +172,14 @@ func primaryFunction(ctx context.Context, c config) {
 			log.Printf("primaryFunction i:%d", i)
 		}
 		fileOrKafka(ctx, c, &binaryData)
-		time.Sleep(c.loopsSleep)
+		// Bug fix: time.Sleep ignored ctx, so SIGTERM took up to
+		// loopsSleep (default 10s) to be observed. Use a ctx-aware
+		// wait so shutdown is prompt.
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(c.loopsSleep):
+		}
 	}
 }
 
@@ -442,6 +449,14 @@ func getLatestSchemaIDAt(ctx context.Context, client *http.Client, baseURL, subj
 	}
 	defer resp.Body.Close()
 
+	// Schema Registries return a JSON error body on 4xx/5xx; decoding it
+	// into the {id int} struct silently yields id:0. Reject non-2xx
+	// upfront so kafka_to_clickhouse's downstream Produce path doesn't
+	// stamp every Kafka record with a bogus schemaID:0 magic header.
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return 0, fmt.Errorf("getLatestSchemaIDAt %s: unexpected status:%d", url, resp.StatusCode)
+	}
+
 	var result struct {
 		ID int `json:"id"`
 	}

cmd/register_schema/register_schema.go

@@ -70,6 +70,14 @@ func getLatestSchemaIDAt(client *http.Client, baseURL, subject string) (int, err
 		return 0, err
 	}
 	defer resp.Body.Close()
+	// Schema Registries return a JSON error body on 4xx/5xx (e.g.
+	// {"error_code":40402,"message":"Subject not found."}). The previous
+	// code skipped the status check and decoded the error body straight
+	// into the {id int} struct, producing a silent id:0 success — the
+	// CLI printed "id:0" for a missing subject. Reject non-2xx upfront.
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return 0, fmt.Errorf("getLatestSchemaIDAt %s: unexpected status:%d", url, resp.StatusCode)
+	}
 	var result struct {
 		ID int `json:"id"`
 	}

cmd/xtcp2client/xtcp2client.go

@@ -222,11 +222,15 @@ breakPoint:
 				log.Printf("pollStreamRecv err:%v", err)
 			}
 
+			// A broken stream typically returns the same error immediately
+			// on every subsequent Recv (e.g. "rpc error: stream closed").
+			// The previous code looped without backoff, pegging a CPU core
+			// at 100% until ctx was canceled. Sleep briefly between
+			// retries so the loop is ctx-cancellable AND non-spinny.
 			select {
 			case <-ctx.Done():
 				break breakPoint
-			default:
-				// non-blocking
+			case <-time.After(100 * time.Millisecond):
 			}
 			continue
 		}
@@ -366,7 +370,12 @@ breakPoint:
 		var flatRecordsResponse *xtcp_flat_record.FlatRecordsResponse
 		flatRecordsResponse, err = stream.Recv()
 		if err == io.EOF {
-			continue
+			// End of stream — the server closed cleanly. Subsequent
+			// Recv calls keep returning io.EOF, so `continue` here
+			// pegged a CPU core. Break out of the loop so the
+			// reconnect-with-sleep path in singleStreamingClient
+			// re-establishes a fresh stream.
+			break breakPoint
 		}
 
 		if err != nil {

pkg/io_uring/ring.go

@@ -180,6 +180,26 @@ func (r *Ring) Close(drainTimeout time.Duration, onDrain func(Result)) {
 			}
 		}
 	}
+	// Drain any in-flight entries that didn't get a CQE within the
+	// deadline. The kernel still owns these buffers until QueueExit
+	// reclaims the ring, so we can't safely reuse them yet — but we
+	// MUST hand them back to the caller's drain callback before the
+	// inFlight map vanishes, otherwise the userspace-side packet pool
+	// leaks N buffers per ring close (up to inFlightCap each). Mark
+	// these as a synthetic ETIME-style result (Res=-syscall.ETIME) so
+	// the callback can tell apart "normal CQE with bytes" from "abandoned
+	// at teardown".
+	if onDrain != nil {
+		for reqID, entry := range r.inFlight {
+			onDrain(Result{
+				Op:       entry.op,
+				Res:      -int32(syscall.ETIME),
+				Buf:      entry.buf,
+				HdrBytes: entry.wvHdr,
+			})
+			delete(r.inFlight, reqID)
+		}
+	}
 	r.r.QueueExit()
 	r.r = nil
 }

pkg/xtcp/deserializers.go

@@ -154,9 +154,15 @@ func (x *XTCP) InitDeserializers(wg *sync.WaitGroup) {
 	}
 
 	// INET_DIAG_SOCKOPT 22
+	// Previously registered DeserializeCGroupIDXTCP here as a workaround
+	// because DeserializeSockOptXTCP had the wrong target type
+	// (*Envelope_XtcpFlatRecord). With the sockopt deserializer's
+	// signature corrected, register the actual SockOpt parser — the
+	// CGroupID one was decoding 8 bytes against the 2-byte SOCKOPT
+	// payload, silently erroring out and leaving SockOpt unpopulated.
 	key = dsKeySockopt
 	if _, exists := x.config.EnabledDeserializers.Enabled[key]; exists {
-		x.RTATypeDeserializer[xtcpnl.SockOptEnumValueCst] = xtcpnl.DeserializeCGroupIDXTCP
+		x.RTATypeDeserializer[xtcpnl.SockOptEnumValueCst] = xtcpnl.DeserializeSockOptXTCP
 		x.RTATypeDeserializerStr[xtcpnl.SockOptEnumValueCst] = key
 	}
 

pkg/xtcp/destinations_kafka.go

@@ -53,7 +53,7 @@ func newKafkaDest(ctx context.Context, x *XTCP) (Destination, error) {
 		return nil, err
 	}
 
-	schemaID, err := d.getLatestSchemaID()
+	schemaID, err := d.getLatestSchemaID(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("newKafkaDest getLatestSchemaID: %w", err)
 	}
@@ -155,25 +155,50 @@ func (d *kafkaDest) Send(ctx context.Context, b *[]byte) (int, error) {
 
 func (d *kafkaDest) Close() error {
 	if d.client != nil {
+		// franz-go's Close cancels in-flight produces without waiting for
+		// their broker acks — records buffered when shutdown fires were
+		// silently dropped. Flush first with a bounded timeout so the
+		// daemon's last poll cycle is durably delivered before teardown.
+		flushCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		if err := d.client.Flush(flushCtx); err != nil {
+			d.x.pC.WithLabelValues("destKafka", "FlushOnClose", "error").Inc()
+			if d.x.debugLevel > 10 {
+				log.Printf("destKafka Flush on Close: %v", err)
+			}
+		}
+		cancel()
 		d.client.Close()
 	}
 	return nil
 }
 
-func (d *kafkaDest) getLatestSchemaID() (int, error) {
+func (d *kafkaDest) getLatestSchemaID(ctx context.Context) (int, error) {
 	url := fmt.Sprintf("%s/subjects/%s-value/versions/latest",
 		d.x.config.KafkaSchemaUrl, d.x.config.Topic)
 	if d.x.debugLevel > 10 {
 		log.Printf("getLatestSchemaID url:%s\n", url)
 	}
-	resp, err := http.Get(url)
+	// http.Get used the DefaultClient which has no timeout — a hung
+	// Schema Registry would block daemon startup indefinitely. Build
+	// the request with the caller's ctx + a hard 10s ceiling so the
+	// init-time call observes shutdown and never wedges.
+	reqCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, url, nil)
+	if err != nil {
+		return 0, err
+	}
+	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return 0, err
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode == http.StatusNotFound {
 		return 0, fmt.Errorf("getLatestSchemaID http.StatusNotFound url:%s", url)
 	}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return 0, fmt.Errorf("getLatestSchemaID url:%s unexpected status:%d", url, resp.StatusCode)
+	}
 	var result struct {
 		ID int `json:"id"`
 	}

pkg/xtcp/grpc_configService.go

@@ -148,7 +148,24 @@ func (c *xtcpConfigService) SetPollFrequency(
 	c.config.PollFrequency = in.PollFrequency
 	c.config.PollTimeout = in.PollTimeout
 
-	*c.changePollFrequencyCh <- c.config.PollFrequency.AsDuration()
+	// Send the new poll frequency to the poller. The channel is buffered
+	// (size 2), so two sends can succeed without a reader; the third
+	// would block forever — pegging the gRPC handler goroutine — if the
+	// poller stopped reading (mid-shutdown, paused, etc.). Use ctx-aware
+	// select with a non-blocking default fallback so a coalesced
+	// frequency-change is dropped (the next caller will resend) rather
+	// than wedging the RPC.
+	select {
+	case *c.changePollFrequencyCh <- c.config.PollFrequency.AsDuration():
+	case <-ctx.Done():
+		c.pC.WithLabelValues("SetPollFrequency", "ctxDone", "count").Inc()
+		return nil, status.Error(codes.Canceled, ctx.Err().Error())
+	case <-c.ctx.Done():
+		c.pC.WithLabelValues("SetPollFrequency", "serverCtxDone", "count").Inc()
+		return nil, status.Error(codes.Unavailable, "server shutting down")
+	default:
+		c.pC.WithLabelValues("SetPollFrequency", "chFull", "count").Inc()
+	}
 
 	if c.debugLevel > 10 {
 		log.Printf("SetPollFrequency c.config.PollFrequency:%0.2f c.config.PollTimeout:%0.2f",

pkg/xtcp/grpc_flatRecordService.go

@@ -174,7 +174,22 @@ func (s *xtcpFlatRecordService) PollFlatRecords(
 			}
 			continue
 		}
-		*s.pollRequestCh <- struct{}{}
+		// Buffered channel of size 2 — third in-flight poke would block
+		// forever if the poller isn't draining (mid-shutdown, paused,
+		// already-polling state). Non-blocking send so the RPC handler
+		// stays responsive; observe both the stream ctx and the
+		// service-level ctx so coalesced pokes don't wedge teardown.
+		select {
+		case *s.pollRequestCh <- struct{}{}:
+		case <-ctx.Done():
+			s.pC.WithLabelValues("PollFlatRecords", "ctxDone", "count").Inc()
+			return ctx.Err()
+		case <-s.ctx.Done():
+			s.pC.WithLabelValues("PollFlatRecords", "serverCtxDone", "count").Inc()
+			return s.ctx.Err()
+		default:
+			s.pC.WithLabelValues("PollFlatRecords", "chFull", "count").Inc()
+		}
 		if s.debugLevel > 10 {
 			log.Printf("PollFlatRecords *s.pollRequestCh <- struct{}{}")
 		}

pkg/xtcp/netlinker.go

@@ -90,9 +90,14 @@ func (x *XTCP) netlinkerSyscall(ctx context.Context, wg *sync.WaitGroup, nsName
 
 		if wf > 0 {
 			now := time.Now()
+			// Capture only the n bytes Recvfrom actually filled. Writing the
+			// raw *packetBuffer here used to dump the full pool-buffer size
+			// (e.g. 8 KiB), trailing the real packet with stale bytes from a
+			// previous Recvfrom — pcap-like consumers parsed garbage past
+			// the real end of the message.
 			err = os.WriteFile(
 				x.config.CapturePath+"netlink."+now.Format(time.RFC3339Nano),
-				*(packetBuffer),
+				(*packetBuffer)[:n],
 				writeFilesPermissionsCst)
 			if err != nil {
 				// Diagnostic capture-to-file is a side feature; a disk-
@@ -146,7 +151,14 @@ func (x *XTCP) netlinkerSyscall(ctx context.Context, wg *sync.WaitGroup, nsName
 		}
 	}
 
-	*packetBuffer = (*packetBuffer)[:0]
+	// Restore the slice header to full capacity before returning it to
+	// the pool. (*packetBuffer)[:0] would Put a zero-length slice — a
+	// later Get from a fresh netlinker would call syscall.Recvfrom on
+	// it, which panics on &p[0] when len(p)==0. iouringPrefillRecvs
+	// (netlinker_iouring.go) already restores cap on Get as a defensive
+	// measure, but the syscall path is the producer of these buffers
+	// and must Put them in usable shape.
+	*packetBuffer = (*packetBuffer)[:cap(*packetBuffer)]
 	x.packetBufferPool.Put(packetBuffer)
 
 	x.pC.WithLabelValues("Netlinker", "complete", "count").Inc()