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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
