[PR] Fix goroutine link when sync part between liaison and data nodes [skywalking-banyandb]

["mrproliu (via GitHub)" <gi...@apache.org>]

mrproliu opened a new pull request, #1217:
URL: https://github.com/apache/skywalking-banyandb/pull/1217

   ## Summary
   
   The liaison leaks one gRPC client-stream goroutine on every write-queue part sync. Over days this accumulates into hundreds of thousands of parked goroutines, driving GC CPU up several-fold. This PR cancels the `SyncPart` stream context on every return path and adds a regression test that guards against it.
   
   ## How it was found
   
   A production liaison raised a CPU alert. Investigation on the live node:
   
   - `liaison-0` CPU rose from **0.67 → 3.4 cores** (~5.5×) over 24h, sustained — while query/write request rates stayed flat (query ~5 ops/s, write ~300 msg/s). So it was **not** load-driven.
   - A 20s CPU profile showed **~94% of CPU in `runtime.gcDrain`** (`scanobject` / `markroot` / `findObject`) — GC, not application work.
   - `go_goroutines` on `liaison-0` climbed monotonically at ~**+1,900/hour** with no plateau: **347,877** goroutines at capture time (`liaison-1` similarly affected, ~+39k/24h).
   - A goroutine profile showed **347,877 goroutines all parked at the same frame**:
   
     ```
     google.golang.org/grpc.newClientStreamWithParams.func4
         google.golang.org/grpc@v1.80.0/stream.go:426
     ```
   
     This is gRPC's per-stream **client-side** cleanup ("reaper") goroutine. GC must scan every one of those goroutine stacks as a root each cycle, which explains the GC-bound CPU.
   
   ## Root cause
   
   `chunkedSyncClient.SyncStreamingParts` (`banyand/queue/pub/chunked_sync.go`) opens a `SyncPart` client stream and, in its `defer`, only calls `stream.CloseSend()` — it never cancels the stream context.
   
   The gRPC reaper spawned by `newClientStreamWithParams` (for streaming RPCs) exits **only** when the per-call context is cancelled or the `ClientConn` is closed:
   
   ```go
   // grpc stream.go
   go func() {
       select {
       case <-cc.ctx.Done():
           cs.finish(ErrClientConnClosing)
       case <-ctx.Done():
           cs.finish(toRPCErr(ctx.Err()))
       }
   }()
   ```
   
   `CloseSend()` does **not** release it. It is released on a clean finish only if the client drains the stream to the server's final status (`io.EOF`) — but `SyncStreamingParts` returns without reaching `io.EOF` on several paths (empty-sync early return, stream error, and the normal path which breaks on the `SyncResult` frame before reading `io.EOF`).
   
   Crucially, the context passed in is the write-queue syncer's **process-lifetime** context: `measure/stream/trace` syncers call `SyncStreamingParts(tst.loopCloser.Ctx(), …)`, and `loopCloser` is `context.WithCancel(context.Background())` (`pkg/run/closer.go`), cancelled only at shutdown.
   
   So: streaming RPC + never-cancelled context + never drained to EOF ⇒ **one reaper goroutine leaked per sync, forever**, on both liaisons. The leak rate tracks the sync cadence rather than request volume, matching the observed signature (client-side, monotonic, server-side stream counts balanced).
   
   ## Fix
   
   Derive a per-call cancellable context for the stream and `defer cancel()`, so the reaper is released on every return path:
   
   ```go
   // banyand/queue/pub/chunked_sync.go
   streamCtx, cancel := context.WithCancel(ctx)
   defer cancel()
   
   stream, err := chunkedClient.SyncPart(streamCtx)
   ```
   
   Minimal and behavior-preserving: the context stays live for the whole call (the entire stream lifetime), and `cancel()` runs only after the function returns. This covers the empty-sync, error, and `SyncResult`-break paths uniformly.
   
   ## Test
   
   Adds `TestSyncStreamingPartsDoesNotLeakReaperGoroutines` (`banyand/queue/pub/chunked_sync_leak_test.go`) — a standard, deterministic unit test that runs with `go test`/CI on every commit:
   
   - Reproduces the production trigger: many syncs sharing a single **never-cancelled** context.
   - Counts live goroutines parked in `newClientStreamWithParams` via `runtime.Stack` (whole-block counting, robust to grpc's closure numbering and unrelated goroutine noise).
   - Asserts the count returns to baseline (no leak). Per-iteration `result.Success` assertions guarantee the stream path actually executed, so a clean count is not a vacuous pass.
   - Reuses the existing `setupChunkedSyncServer` / `createFileInfo` harness; no production code is touched by the test.
   
   ## Verification
   
   - **With the fix:** `PASS` — `baseline=0 after=0 leaked=0` over 30 syncs with a never-cancelled context.
   - **Reverting the fix:** `FAIL` — `leaked=30` (exactly one reaper per sync), with the leaked `newClientStreamWithParams.func4` stack printed. Confirms the test has teeth.
   - Existing chunked-sync retry specs still pass; `gofmt` and `go vet` clean.
   
   ## Notes
   
   - The reaper mechanism is unchanged between grpc `v1.80.0` (deployed) and `v1.82.0` (current `go.mod`) — the leak is a context-management bug on the BanyanDB side, not a grpc version issue.
   - Both liaison nodes were affected; a rolling restart clears the accumulated goroutines as a stop-gap, but this fix is required to stop recurrence.
   
   
   - [ ] If this pull request closes/resolves/fixes an existing issue, replace the issue number. Fixes apache/skywalking#<issue number>.
   - [x] Update the [`CHANGES` log](https://github.com/apache/skywalking-banyandb/blob/main/CHANGES.md).
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix goroutine link when sync part between liaison and data nodes [skywalking-banyandb]

["hanahmily (via GitHub)" <gi...@apache.org>]

hanahmily merged PR #1217:
URL: https://github.com/apache/skywalking-banyandb/pull/1217


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix goroutine link when sync part between liaison and data nodes [skywalking-banyandb]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #1217:
URL: https://github.com/apache/skywalking-banyandb/pull/1217#discussion_r3568429987


##########
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 {

Review Comment:
   Because defers run LIFO, the current `defer cancel()` will execute after the later `defer stream.CloseSend()` block. If `CloseSend()` ever blocks during transport teardown, the stream context won’t be cancelled until it returns, which can delay releasing gRPC’s per-stream cleanup goroutine. Consider cancelling before `CloseSend()` by folding `cancel()` into the same deferred block (and explicitly cancelling on the stream-creation error path).



##########
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
+}

Review Comment:
   `runtime.Stack` truncates output to the provided buffer size. With a fixed 4 MiB buffer, this test can undercount reaper goroutines (and potentially pass/fail incorrectly) if the package has enough background goroutines to exceed the buffer. Consider growing the buffer until `runtime.Stack` reports it fit, so the count is accurate and robust across different CI environments.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org