pkg/tools/batcher: stop scheduler panicking when b.data is closed externally (#3714)

* pkg/tools/batcher: stop scheduler panicking when b.data is closed externally

scheduler()'s defer unconditionally calls close(b.data). If the channel
was closed by the caller (or an upstream producer) instead of via the
normal Close()-sends-nil path, the receive on b.data returns ok == false,
scheduler returns, and the deferred close(b.data) then fires on an
already-closed channel:

    panic: close of closed channel

reliably reproducible under the #3653 steps (manually closing b.data
while Start() is running).

Track whether we observed the external-close via a local
`externallyClosed` flag set in the `ok == false` branch. The defer
only closes b.data when that flag is false, i.e. when the scheduler
exited through the nil-message or ticker paths and still owns the
channel. No behaviour change on the graceful Close() path.

Fixes #3653

* ci: retrigger after transient gha outage

Signed-off-by: SAY-5 <say.apm35@gmail.com>

---------

Signed-off-by: SAY-5 <say.apm35@gmail.com>
(cherry picked from commit b3a7342a42ca7daf3e01e275fb0e7d166fb16a58)

pkg/tools/batcher/batcher.go

@@ -149,12 +149,21 @@ func (b *Batcher[T]) Put(ctx context.Context, data *T) error {
 
 func (b *Batcher[T]) scheduler() {
 	ticker := time.NewTicker(b.config.interval)
+	// Track whether b.data was closed by an external caller so the
+	// cleanup below does not close it a second time. The only routes
+	// out of this function that leave b.data open are the nil-message
+	// and ticker paths; the ok == false branch means someone already
+	// closed the channel, and calling close(b.data) again there would
+	// panic with "close of closed channel" (#3653).
+	externallyClosed := false
 	defer func() {
 		ticker.Stop()
 		for _, ch := range b.chArrays {
 			close(ch)
 		}
-		close(b.data)
+		if !externallyClosed {
+			close(b.data)
+		}
 		b.wait.Done()
 	}()
 
@@ -167,6 +176,7 @@ func (b *Batcher[T]) scheduler() {
 		case data, ok := <-b.data:
 			if !ok {
 				// If the data channel is closed unexpectedly
+				externallyClosed = true
 				return
 			}
 			if data == nil {