(skywalking-banyandb) branch goroutine-leak created (now f2dac1647)

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mrproliu pushed a change to branch goroutine-leak
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git


      at f2dac1647 Fix goroutine link when sync part between liaison and data nodes

This branch includes the following new commits:

     new f2dac1647 Fix goroutine link when sync part between liaison and data nodes

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(skywalking-banyandb) 01/01: Fix goroutine link when sync part between liaison and data nodes

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mrproliu pushed a commit to branch goroutine-leak
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git

commit f2dac1647e434d8d2bb81bb96e384ee3c2183522
Author: mrproliu <74...@qq.com>
AuthorDate: Mon Jul 13 14:14:55 2026 +0800

    Fix goroutine link when sync part between liaison and data nodes
---
 CHANGES.md                                  |   1 +
 banyand/queue/pub/chunked_sync.go           |  12 +-
 banyand/queue/pub/chunked_sync_leak_test.go | 184 ++++++++++++++++++++++++++++
 3 files changed, 196 insertions(+), 1 deletion(-)

diff --git a/CHANGES.md b/CHANGES.md
index 41467ba54..d8cb94cbd 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -131,6 +131,7 @@ Release Notes.
 - Fix backup container OOM from overlapping scheduled runs; serialize runs and upload small snapshot files concurrently.
 - Fix block metadata reset before unmarshal for stream and trace.
 - Lifecycle only handles stream/measure/trace snapshots, skipping other catalogs.
+- Fix liaison goroutine leak from uncancelled gRPC SyncPart stream context in write-queue part sync.
 
 ### Document
 
diff --git a/banyand/queue/pub/chunked_sync.go b/banyand/queue/pub/chunked_sync.go
index cc77e2a68..0b1de486a 100644
--- a/banyand/queue/pub/chunked_sync.go
+++ b/banyand/queue/pub/chunked_sync.go
@@ -120,7 +120,17 @@ func (c *chunkedSyncClient) SyncStreamingParts(ctx context.Context, parts []queu
 
 	chunkedClient := clusterv1.NewChunkedSyncServiceClient(c.conn)
 
-	stream, err := chunkedClient.SyncPart(ctx)
+	// Bind the stream to a cancellable context scoped to this call. gRPC spawns a
+	// per-stream cleanup goroutine (newClientStreamWithParams) that only exits when
+	// this context is cancelled or the ClientConn closes — CloseSend() alone does
+	// NOT release it. Callers pass a process-lifetime context (the write-queue
+	// syncer's loopCloser.Ctx()), so without this cancel every sync would leak one
+	// reaper goroutine forever. defer cancel() guarantees release on every return
+	// path (early return, error, or SyncResult break before io.EOF).
+	streamCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	stream, err := chunkedClient.SyncPart(streamCtx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create sync stream: %w", err)
 	}
diff --git a/banyand/queue/pub/chunked_sync_leak_test.go b/banyand/queue/pub/chunked_sync_leak_test.go
new file mode 100644
index 000000000..bc709a2af
--- /dev/null
+++ b/banyand/queue/pub/chunked_sync_leak_test.go
@@ -0,0 +1,184 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pub
+
+import (
+	"context"
+	"runtime"
+	"strings"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/apache/skywalking-banyandb/banyand/queue"
+	"github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// reaperFrame is the gRPC client-side stream cleanup goroutine that
+// (*clientStream) spawns per streaming RPC in newClientStreamWithParams. It
+// blocks on `select { <-cc.ctx.Done(); <-ctx.Done() }` and only exits when the
+// per-call context is cancelled or the ClientConn is closed — NOT when the
+// stream is drained or CloseSend() is called (see grpc stream.go). SyncPart is a
+// streaming RPC, so each call spawns one; if its context is never cancelled the
+// goroutine leaks forever.
+const reaperFrame = "newClientStreamWithParams"
+
+// countReaperGoroutines returns the number of live goroutines currently parked
+// in the gRPC client-stream reaper. It counts whole goroutine blocks (not raw
+// substring hits) so it is robust to grpc's internal closure numbering and to
+// unrelated goroutine noise.
+func countReaperGoroutines() int {
+	buf := make([]byte, 1<<22)
+	n := runtime.Stack(buf, true)
+	count := 0
+	for _, g := range strings.Split(string(buf[:n]), "\n\ngoroutine ") {
+		if strings.Contains(g, reaperFrame) {
+			count++
+		}
+	}
+	return count
+}
+
+// stableReaperCount polls countReaperGoroutines until two consecutive readings
+// agree, so reapers that are mid-teardown (the fixed path cancels them
+// asynchronously) settle before we assert, keeping the test non-flaky.
+func stableReaperCount() int {
+	prev := -1
+	for i := 0; i < 100; i++ {
+		time.Sleep(20 * time.Millisecond)
+		cur := countReaperGoroutines()
+		if cur == prev {
+			return cur
+		}
+		prev = cur
+	}
+	return prev
+}
+
+// firstReaperStack returns one reaper goroutine's stack, surfaced in the failure
+// message so a regression is immediately actionable.
+func firstReaperStack() string {
+	buf := make([]byte, 1<<22)
+	n := runtime.Stack(buf, true)
+	for _, g := range strings.Split(string(buf[:n]), "\n\ngoroutine ") {
+		if strings.Contains(g, reaperFrame) {
+			return "goroutine " + strings.TrimSpace(g)
+		}
+	}
+	return "<none>"
+}
+
+func newLeakTestClient(t *testing.T) *chunkedSyncClient {
+	t.Helper()
+	address := getAddress()
+	_, cleanup := setupChunkedSyncServer(address, 0) // 0 mismatches -> success path
+	t.Cleanup(cleanup)
+
+	conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("failed to dial mock server: %v", err)
+	}
+	t.Cleanup(func() { _ = conn.Close() })
+
+	return &chunkedSyncClient{
+		conn:      conn,
+		log:       logger.GetLogger("test-chunked-sync-leak"),
+		chunkSize: 1024,
+	}
+}
+
+func leakTestParts() []queue.StreamingPartData {
+	return []queue.StreamingPartData{
+		{
+			ID:      1,
+			Group:   "test-group",
+			ShardID: 1,
+			Topic:   "stream_write",
+			Files: []queue.FileInfo{
+				createFileInfo("leak-test-file.dat", []byte("payload for chunked sync leak test")),
+			},
+		},
+	}
+}
+
+// TestSyncStreamingPartsDoesNotLeakReaperGoroutines is a regression guard for the
+// production goroutine leak where a liaison accumulated 347k goroutines parked in
+// grpc's newClientStreamWithParams reaper, all from SyncPart streams whose context
+// was never cancelled.
+//
+// SyncStreamingParts opens a SyncPart client stream and only CloseSend()s it; the
+// reaper exits solely on context cancellation / ClientConn close. In production
+// the caller passes the write-queue syncer's process-lifetime loopCloser.Ctx()
+// (context.WithCancel(context.Background()), cancelled only at shutdown), and the
+// client breaks out of the Recv loop on the SyncResult frame before reading the
+// trailing io.EOF — so the stream is never finished and the reaper leaks.
+//
+// This test reproduces exactly that condition: a single never-cancelled context
+// shared across many syncs. It passes only because SyncStreamingParts derives a
+// per-call cancellable context and defer-cancels it (chunked_sync.go). Reverting
+// that fix makes reapers accumulate ~1 per sync and this test fails.
+func TestSyncStreamingPartsDoesNotLeakReaperGoroutines(t *testing.T) {
+	const iterations = 30
+
+	client := newLeakTestClient(t)
+
+	// Warm up the lazy ClientConn/transport so its long-lived background
+	// goroutines exist before the baseline (they are not reapers, but this keeps
+	// the environment steady).
+	warmCtx, warmCancel := context.WithTimeout(context.Background(), 10*time.Second)
+	if _, err := client.SyncStreamingParts(warmCtx, leakTestParts()); err != nil {
+		warmCancel()
+		t.Fatalf("warm-up sync failed: %v", err)
+	}
+	warmCancel()
+
+	baseline := stableReaperCount()
+
+	// Reproduce the production trigger: the caller's context is NEVER cancelled
+	// while the syncs run (mirrors loopCloser.Ctx()).
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	for i := 0; i < iterations; i++ {
+		result, err := client.SyncStreamingParts(ctx, leakTestParts())
+		if err != nil {
+			t.Fatalf("sync %d failed: %v", i, err)
+		}
+		// Asserting success guarantees the SyncPart stream path actually ran, so a
+		// clean reaper count reflects proper cleanup rather than a vacuous no-op.
+		if !result.Success {
+			t.Fatalf("sync %d not successful: %+v", i, result)
+		}
+	}
+
+	after := stableReaperCount()
+	leaked := after - baseline
+	t.Logf("reaper goroutines: baseline=%d after=%d leaked=%d over %d syncs (never-cancelled context)",
+		baseline, after, leaked, iterations)
+
+	// Each leaked sync would park exactly one reaper; allow a tiny slack for
+	// scheduling. A number near `iterations` means the SyncPart stream context is
+	// no longer being cancelled — the leak has regressed.
+	if leaked > 2 {
+		t.Fatalf("SyncStreamingParts leaked %d gRPC reaper goroutines over %d syncs with a never-cancelled "+
+			"context — SyncPart's stream context must be cancelled on every return path (chunked_sync.go). "+
+			"Sample leaked stack:\n%s", leaked, iterations, firstReaperStack())
+	}
+}