ronantakizawa/codereview-bench huggingface dataset:
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/kubectl/pkg/cmd/logs/logs.go (Go)

Code before:

	for {
		err := o.ConsumeRequestFn(ctx, request, out)
		if err == nil {
			return nil
		}

		if !isWaitingForContainerError(err) {
			if o.IgnoreLogErrors {
				fmt.Fprint(internalErrOut, "error: "+err.Error()+"\n")
				return nil
			}
			return err
		}

		fmt.Fprintf(o.ErrOut, "%v\n", err)

		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}

func isWaitingForContainerError(err error) bool {
	if status, ok := err.(apierrors.APIStatus); ok {
		return strings.Contains(status.Status().Message, "is waiting to start")
	}
	return false
}

func (o LogsOptions) addPrefixIfNeeded(ref corev1.ObjectReference, writer io.Writer) io.Writer {
	if !o.Prefix || ref.FieldPath == "" || ref.Name == "" {
		return writer
	}

	// We rely on ref.FieldPath to contain a reference to a container
	// including a container name (not an index) so we can get a container name
	// without making an extra API request.
	var containerName string
	containerNameMatches := o.containerNameFromRefSpecRegexp.FindStringSubmatch(ref.FieldPath)
	if len(containerNameMatches) == 2 {
		containerName = containerNameMatches[1]
	}

	prefix := fmt.Sprintf("[pod/%s/%s] ", ref.Name, containerName)
	return &prefixingWriter{
		prefix: []byte(prefix),
		writer: writer,
	}

Review comment (suggestion):
Drop this check, as mentioned above, let's just use the user-provided timeout to wait for the log output. 

Code after:
			return true, nil
		}

		if o.IgnoreLogErrors {
			_, _ = fmt.Fprint(internalErrOut, "error: "+err.Error()+"\n")
			return true, nil
		}

		_, _ = o.ErrOut.Write([]byte(err.Error() + "\n"))

		if _, ok := err.(apierrors.APIStatus); ok {
			return false, nil
		}

		return false, err
	})
}

func (o LogsOptions) addPrefixIfNeeded(ref corev1.ObjectReference, writer io.Writer) io.Writer {
	if !o.Prefix || ref.FieldPath == "" || ref.Name == "" {
		return writer
	}

	// We rely on ref.FieldPath to contain a reference to a container
	// including a container name (not an index) so we can get a container name
	// without making an extra API request.
	var containerName string
	containerNameMatches := o.containerNameFromRefSpecRegexp.FindStringSubmatch(ref.FieldPath)
	if len(containerNameMatches) == 2 {
		containerName = containerNameMatches[1]
	}

	prefix := fmt.Sprintf("[pod/%s/%s] ", ref.Name, containerName)
	return &prefixingWriter{
		prefix: []byte(prefix),
		writer: writer,
	}
}

// DefaultConsumeRequest reads the data from request and writes into
// the out writer. It buffers data from requests until the newline or io.EOF
// occurs in the data, so it doesn't interleave logs sub-line
// when running concurrently.
//
// A successful read returns err == nil, not err == io.EOF.
// Because the function is defined to read from request until io.EOF, it does
// not treat an io.EOF as an error to be reported.
func DefaultConsumeRequest(ctx context.Context, request rest.ResponseWrapper, out io.Writer) error {
	readCloser, err := request.Stream(ctx)
	if err != nil {
		return err
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/kubectl/pkg/cmd/logs/logs.go (Go)

Code before:
			}

		}(objRef, request)
	}

	go func() {
		wg.Wait()
		writer.Close()
	}()

	_, err := io.Copy(o.Out, reader)
	return err
}

func (o LogsOptions) sequentialConsumeRequest(ctx context.Context, requests map[corev1.ObjectReference]rest.ResponseWrapper) error {
	for objRef, request := range requests {
		out := o.addPrefixIfNeeded(objRef, o.Out)
		if err := o.ConsumeRequestFn(ctx, request, out); err != nil {
			if !o.IgnoreLogErrors {
				return err
			}

			fmt.Fprintf(o.Out, "error: %v\n", err)
		}
	}

	return nil
}

func (o LogsOptions) addPrefixIfNeeded(ref corev1.ObjectReference, writer io.Writer) io.Writer {
	if !o.Prefix || ref.FieldPath == "" || ref.Name == "" {
		return writer
	}

	// We rely on ref.FieldPath to contain a reference to a container
	// including a container name (not an index) so we can get a container name
	// without making an extra API request.
	var containerName string
	containerNameMatches := o.containerNameFromRefSpecRegexp.FindStringSubmatch(ref.FieldPath)
	if len(containerNameMatches) == 2 {
		containerName = containerNameMatches[1]
	}

	prefix := fmt.Sprintf("[pod/%s/%s] ", ref.Name, containerName)
	return &prefixingWriter{
		prefix: []byte(prefix),
		writer: writer,
	}
}

// DefaultConsumeRequest reads the data from request and writes into

Review comment (suggestion):
Nit: next time, instead of creating those throw-away vars you can silence the linter with `//nolint:errcheck` after the line 😉 

Code after:

	go func() {
		wg.Wait()
		writer.Close()
	}()

	_, err := io.Copy(o.Out, reader)
	return err
}

func (o LogsOptions) sequentialConsumeRequest(ctx context.Context, requests map[corev1.ObjectReference]rest.ResponseWrapper) error {
	for objRef, request := range requests {
		out := o.addPrefixIfNeeded(objRef, o.Out)
		if err := o.consumeWithRetry(ctx, request, out, o.Out); err != nil {
			return err
		}
	}

	return nil
}

func (o LogsOptions) consumeWithRetry(ctx context.Context, request rest.ResponseWrapper, out io.Writer, internalErrOut io.Writer) error {
	if !o.Follow {
		err := o.ConsumeRequestFn(ctx, request, out)
		if err != nil && o.IgnoreLogErrors {
			_, _ = fmt.Fprint(internalErrOut, "error: "+err.Error()+"\n")
			return nil
		}
		return err
	}

	return wait.PollUntilContextTimeout(ctx, 1*time.Second, o.GetPodTimeout, true, func(_ context.Context) (bool, error) {
		err := o.ConsumeRequestFn(ctx, request, out)
		if err == nil {
			return true, nil
		}

		if o.IgnoreLogErrors {
			_, _ = fmt.Fprint(internalErrOut, "error: "+err.Error()+"\n")
			return true, nil
		}

		_, _ = o.ErrOut.Write([]byte(err.Error() + "\n"))

		if _, ok := err.(apierrors.APIStatus); ok {
			return false, nil
		}

		return false, err
	})
}
Repository: kubernetes/kubernetes
File: pkg/scheduler/framework/plugins/helper/workload_test.go (Go)

Code before:
			},
			expectedOk: true,
		},
		{
			name:         "basic policy",
			podGroupName: "pg2",
			expectedPolicy: schedulingapi.PodGroupPolicy{
				Basic: &schedulingapi.BasicSchedulingPolicy{},
			},
			expectedOk: true,
		},
		{
			name:           "pod group not found - return empty policy and false",
			podGroupName:   "pg3",
			expectedPolicy: schedulingapi.PodGroupPolicy{},
			expectedOk:     false,
		},
	}
	for _, tt := range testCases {
		t.Run(tt.name, func(t *testing.T) {
			got, ok := PodGroupPolicy(workload, tt.podGroupName)
			if ok != tt.expectedOk {
				t.Errorf("PodGroupPolicy() ok: %v, want: %v", ok, tt.expectedOk)
			}
			if diff := cmp.Diff(got, tt.expectedPolicy); diff != "" {
				t.Errorf("PodGroupPolicy() policy (-want,+got): %s", diff)
			}
		})
	}
}


Review comment (suggestion):
Add a new line:
```suggestion
				t.Errorf("PodGroupPolicy() policy (-want,+got):\n%s", diff)
```

Code after:
			},
			expectedOk: true,
		},
		{
			name:         "basic policy",
			podGroupName: "pg2",
			expectedPolicy: schedulingapi.PodGroupPolicy{
				Basic: &schedulingapi.BasicSchedulingPolicy{},
			},
			expectedOk: true,
		},
		{
			name:           "pod group not found - return empty policy and false",
			podGroupName:   "pg3",
			expectedPolicy: schedulingapi.PodGroupPolicy{},
			expectedOk:     false,
		},
	}
	for _, tt := range testCases {
		t.Run(tt.name, func(t *testing.T) {
			got, ok := PodGroupPolicy(workload, tt.podGroupName)
			if ok != tt.expectedOk {
				t.Errorf("PodGroupPolicy() ok: %v, want: %v", ok, tt.expectedOk)
			}
			if diff := cmp.Diff(got, tt.expectedPolicy); diff != "" {
				t.Errorf("PodGroupPolicy() policy (-want,+got):\n%s", diff)
			}
		})
	}
}
Repository: kubernetes/kubernetes
File: test/utils/ktesting/ktesting.go (Go)

Code before:
	_ "k8s.io/component-base/logs/testinit"
)

// SetDefaultVerbosity can be called during init to modify the default
// log verbosity of the program.
//
// Note that this immediately reconfigures the klog verbosity, already before
// flag parsing. If the verbosity is non-zero and SetDefaultVerbosity is called
// during init, then other init functions might start logging where normally
// they wouldn't log anything. Should this occur, then the right fix is to
// remove those log calls because logging during init is discouraged. It leads
// to unpredictable output (init order is not specified) and/or is useless
// (logging not initialized during init and thus conditional log output gets
// omitted).
func SetDefaultVerbosity(v int) {
	f := flag.CommandLine.Lookup("v")
	_ = f.Value.Set(fmt.Sprintf("%d", v))
}

// NewTestContext is a replacement for ktesting.NewTestContext
// which returns a more versatile context.
func NewTestContext(tb testing.TB) (klog.Logger, context.Context) {
	tCtx := Init(tb)
	return tCtx.Logger(), tCtx
}


Review comment (suggestion):
The comment needs to be updated:
```
NewTestContext is a drop-in replacement for ktesting.NewTestContext.

The result can be cast to a TContext and
has the same functionality (for example,
it gets cancelled at the end of the test).
It's returned as a context.Context to avoid
breaking code which expects that type.
```




Code after:
	_ "k8s.io/component-base/logs/testinit"
)

// SetDefaultVerbosity can be called during init to modify the default
// log verbosity of the program.
//
// Note that this immediately reconfigures the klog verbosity, already before
// flag parsing. If the verbosity is non-zero and SetDefaultVerbosity is called
// during init, then other init functions might start logging where normally
// they wouldn't log anything. Should this occur, then the right fix is to
// remove those log calls because logging during init is discouraged. It leads
// to unpredictable output (init order is not specified) and/or is useless
// (logging not initialized during init and thus conditional log output gets
// omitted).
func SetDefaultVerbosity(v int) {
	f := flag.CommandLine.Lookup("v")
	_ = f.Value.Set(fmt.Sprintf("%d", v))
}

// NewTestContext is a drop-in replacement for ktesting.NewTestContext.
//
// The result can be cast to a TContext and has the same functionality
// (for example, it gets cancelled at the end of the test).
// It's returned as a context.Context to avoid breaking code which expects that type.
func NewTestContext(tb testing.TB) (klog.Logger, context.Context) {
	tCtx := Init(tb)
	return tCtx.Logger(), tCtx
}
Repository: kubernetes/kubernetes
File: pkg/kubelet/cm/container_manager_test.go (Go)

Code before:
//go:build linux

/*
Copyright 2026 The Kubernetes Authors.

Licensed 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 cm

import (
	"testing"
	"github.com/stretchr/testify/require"
)

func TestParsePercentage(t *testing.T) {

Review comment (suggestion):
we have three separate implementation of `parsePercentage`. Maybe worth looking into improving the situation.

Code after:
//go:build linux

/*
Copyright The Kubernetes Authors.

Licensed 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 cm

import (
	"testing"

	"github.com/stretchr/testify/require"
)
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/apiserver/pkg/server/options/serving_with_loopback_test.go (Go)

Code before:
	}

	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			cfg := server.NewConfig(serializer.NewCodecFactory(runtime.NewScheme()))

			ssowlb := (&SecureServingOptions{
				BindAddress: netutils.ParseIPSloppy("127.0.0.1"),
			}).WithLoopback()
			now := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.Local)
			ssowlb.clock = clocktesting.NewFakeClock(now)

			ln, err := net.Listen("tcp", "127.0.0.1:0")
			if err != nil {
				t.Fatalf("failed to listen on 127.0.0.1:0")
			}
			defer ln.Close()
			ssowlb.Listener = ln
			ssowlb.BindPort = ln.Addr().(*net.TCPAddr).Port

			if err := ssowlb.ApplyToConfig(cfg); err != nil {
				t.Errorf("unexpected error: %v", err)
			}

			if tc.expired {
				ssowlb.clock = clocktesting.NewFakeClock(now.Add(maxAge * 2))
			}

			var foundChecker healthz.HealthChecker

			for _, livezCheck := range cfg.LivezChecks {
				if livezCheck.Name() == "loopback-certificate-expiry" {
					foundChecker = livezCheck
					break
				}
			}

			if foundChecker == nil {
				t.Fatal("loopback-certificate-expiry health checker not found")
			}

			err = foundChecker.Check(nil)
			if tc.expired {
				if !errors.Is(err, LoopbackCertificateExpiredError{}) {
					t.Errorf("expected error and received error do not match. expected: %v , received: %v", LoopbackCertificateExpiredError{}, err)
				}
				return
			}

			if err != nil {
				t.Errorf("received an unexpected error: %v", err)

Review comment (suggestion):
don't override the clock instance, save a handle to the fake clock and call `c.Advance(maxAge+1)`, etc

Code after:
	}

	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			cfg := server.NewConfig(serializer.NewCodecFactory(runtime.NewScheme()))

			ssowlb := (&SecureServingOptions{
				BindAddress: netutils.ParseIPSloppy("127.0.0.1"),
			}).WithLoopback()
			now := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.Local)
			fakeClock := clocktesting.NewFakeClock(now)
			ssowlb.clock = fakeClock

			ln, err := net.Listen("tcp", "127.0.0.1:0")
			if err != nil {
				t.Fatalf("failed to listen on 127.0.0.1:0")
			}
			defer ln.Close() //nolint:errcheck
			ssowlb.Listener = ln
			ssowlb.BindPort = ln.Addr().(*net.TCPAddr).Port

			if err := ssowlb.ApplyToConfig(cfg); err != nil {
				t.Errorf("unexpected error: %v", err)
			}

			if tc.expired {
				fakeClock.Step(maxAge + 1)
			}

			checkSet := map[string][]healthz.HealthChecker{
				"/livez":   cfg.LivezChecks,
				"/healthz": cfg.HealthzChecks,
			}

			for endpoint, checks := range checkSet {
				var foundChecker healthz.HealthChecker

				for _, check := range checks {
					if check.Name() == "loopback-serving-certificate" {
						foundChecker = check
						break
					}
				}

				if foundChecker == nil {
					t.Fatalf("loopback-serving-certificate health checker not found for %s endpoint", endpoint)
				}

				err = foundChecker.Check(nil)
				if tc.expired {
					if !errors.Is(err, LoopbackCertificateExpiredError{}) {
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/apiserver/pkg/server/options/audit_test.go (Go)

Code before:
					assert.Equal(t, int(math.MaxInt32), logger.MaxSize)
				} else {
					assert.Equal(t, options.LogOptions.MaxSize, logger.MaxSize)
				}
			}
		})
	}
}

func TestAuditLogMaxSizeZero(t *testing.T) {
	tmpDir := t.TempDir()
	auditPath := filepath.Join(tmpDir, "audit")
	policy := makeTmpPolicy(t)
	defer os.Remove(policy)

	o := NewAuditOptions()
	o.LogOptions.Path = auditPath
	o.PolicyFile = policy
	o.LogOptions.MaxSize = 0

	w, err := o.LogOptions.getWriter()
	require.NoError(t, err)
	require.NotNil(t, w)

	logger, ok := w.(*lumberjack.Logger)
	require.True(t, ok)

	assert.Equal(t, int(math.MaxInt32), logger.MaxSize, "MaxSize should be effectively unlimited (MaxInt32)")
}

func TestAuditInvalidOptions(t *testing.T) {
	tmpDir := t.TempDir()
	auditPath := filepath.Join(tmpDir, "audit")

	testCases := []struct {
		name    string
		options func() *AuditOptions
	}{{
		name: "invalid log format",
		options: func() *AuditOptions {
			o := NewAuditOptions()
			o.LogOptions.Path = auditPath
			o.LogOptions.Format = "foo"
			return o
		},
	}, {
		name: "invalid log mode",
		options: func() *AuditOptions {
			o := NewAuditOptions()
			o.LogOptions.Path = auditPath
			o.LogOptions.BatchOptions.Mode = "foo"

Review comment (bug):
```
{Failed  === RUN   TestAuditLogMaxSizeZero
    audit_test.go:214: 
        	Error Trace:	/home/prow/go/src/k8s.io/kubernetes/staging/src/k8s.io/apiserver/pkg/server/options/audit_test.go:214
        	Error:      	Should be true
        	Test:       	TestAuditLogMaxSizeZero
```

Code after:
				assert.FileExists(t, options.LogOptions.Path)
				assert.Equal(t, options.LogOptions.MaxSize, logger.MaxSize)
			}
		})
	}
}

func TestAuditLogMaxSizeZero(t *testing.T) {
	tmpDir := t.TempDir()
	auditPath := filepath.Join(tmpDir, "audit")
	policy := makeTmpPolicy(t)
	defer func() { _ = os.Remove(policy) }()

	o := NewAuditOptions()
	o.LogOptions.Path = auditPath
	o.PolicyFile = policy
	o.LogOptions.MaxSize = 0

	w, err := o.LogOptions.getWriter()
	require.NoError(t, err)
	require.NotNil(t, w)

	// When MaxSize is 0, we return a raw file writer (os.File) for unlimited size
	file, ok := w.(*os.File)
	require.True(t, ok, "Should be an os.File when MaxSize is 0")

	// Verify that the file was opened correctly
	require.NotNil(t, file)
	require.NoError(t, file.Close())
}

func TestAuditInvalidOptions(t *testing.T) {
	tmpDir := t.TempDir()
	auditPath := filepath.Join(tmpDir, "audit")

	testCases := []struct {
		name    string
		options func() *AuditOptions
	}{{
		name: "invalid log format",
		options: func() *AuditOptions {
			o := NewAuditOptions()
			o.LogOptions.Path = auditPath
			o.LogOptions.Format = "foo"
			return o
		},
	}, {
		name: "invalid log mode",
		options: func() *AuditOptions {
			o := NewAuditOptions()
			o.LogOptions.Path = auditPath
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/client-go/tools/cache/shared_informer.go (Go)

Code before:
		// No listeners are registered, do nothing
		return nil
	} else if _, exists := p.listeners[listener]; !exists {
		// Listener is not registered, just do nothing
		return nil
	}

	delete(p.listeners, listener)

	if p.listenersStarted {
		close(listener.addCh)
	}

	return nil
}

func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
	p.listenersLock.RLock()
	defer p.listenersLock.RUnlock()

	// Before we start blocking on writes to the listeners' channels,
	// ensure that they all have been started. If the processor stops,
	// p.listeners gets cleared, in which case we also continue here
	// and return without doing anything.
	for !p.listenersStarted && len(p.listeners) > 0 {
		p.listenersRCond.Wait()
	}

	for listener, isSyncing := range p.listeners {
		switch {
		case !sync:
			// non-sync messages are delivered to every listener
			listener.add(obj)
		case isSyncing:
			// sync messages are delivered to every syncing listener
			listener.add(obj)
		default:
			// skipping a sync obj for a non-syncing listener
		}
	}
}

// sharedProcessorRunDelay is used in a synctest bubble to achieve a certain ordering of
// steps in different goroutines.
var sharedProcessorRunDelay atomic.Pointer[time.Duration]

func (p *sharedProcessor) run(ctx context.Context) {
	func() {
		delay := sharedProcessorRunDelay.Load()
		if delay != nil {
			time.Sleep(*delay)

Review comment (suggestion):
Nice, this is what I was hoping we could do with a read lock so we don't have to grab a write lock on distribute.

Code after:
		// No listeners are registered, do nothing
		return nil
	} else if _, exists := p.listeners[listener]; !exists {
		// Listener is not registered, just do nothing
		return nil
	}

	delete(p.listeners, listener)

	if p.listenersStarted {
		close(listener.addCh)
	}

	return nil
}

func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
	p.listenersLock.RLock()
	defer p.listenersLock.RUnlock()

	// Before we start blocking on writes to the listeners' channels,
	// ensure that they all have been started. If the processor stops,
	// p.listeners gets cleared, in which case we also continue here
	// and return without doing anything.
	for !p.listenersStarted && len(p.listeners) > 0 {
		p.listenersRCond.Wait()
	}

	for listener, isSyncing := range p.listeners {
		switch {
		case !sync:
			// non-sync messages are delivered to every listener
			listener.add(obj)
		case isSyncing:
			// sync messages are delivered to every syncing listener
			listener.add(obj)
		default:
			// skipping a sync obj for a non-syncing listener
		}
	}
}

// sharedProcessorRunHook can be used inside tests to execute additional code
// at the start of sharedProcessor.run.
var sharedProcessorRunHook atomic.Pointer[func()]

func (p *sharedProcessor) run(ctx context.Context) {
	func() {
		hook := sharedProcessorRunHook.Load()
		if hook != nil {
			(*hook)()
Repository: kubernetes/kubernetes
File: staging/src/k8s.io/client-go/tools/cache/shared_informer.go (Go)

Code before:
		p.listenersRCond.Wait()
	}

	for listener, isSyncing := range p.listeners {
		switch {
		case !sync:
			// non-sync messages are delivered to every listener
			listener.add(obj)
		case isSyncing:
			// sync messages are delivered to every syncing listener
			listener.add(obj)
		default:
			// skipping a sync obj for a non-syncing listener
		}
	}
}

// sharedProcessorRunDelay is used in a synctest bubble to achieve a certain ordering of
// steps in different goroutines.
var sharedProcessorRunDelay atomic.Pointer[time.Duration]

func (p *sharedProcessor) run(ctx context.Context) {
	func() {
		delay := sharedProcessorRunDelay.Load()
		if delay != nil {
			time.Sleep(*delay)
		}
		// Changing listenersStarted needs a write lock.
		p.listenersLock.Lock()
		defer p.listenersLock.Unlock()
		for listener := range p.listeners {
			p.wg.Start(listener.run)
			p.wg.Start(listener.pop)
		}
		p.listenersStarted = true
		p.listenersRCond.Signal()
	}()
	<-ctx.Done()

	p.listenersLock.Lock()
	defer p.listenersLock.Unlock()
	for listener := range p.listeners {
		close(listener.addCh) // Tell .pop() to stop. .pop() will tell .run() to stop
	}

	// Wipe out list of listeners since they are now closed
	// (processorListener cannot be re-used)
	p.listeners = nil

	// Reset to false since no listeners are running
	p.listenersStarted = false

Review comment (bug):
Not sure how much of a fan I am of adding a sleep here, while it shouldn't impact anything since it's nil it still feels wrong to add a sleep like this to production code. Is this approach guidance from synctest on how to order goroutines? Is there some way to wrap run and have a sleep there maybe?

Code after:
		p.listenersRCond.Wait()
	}

	for listener, isSyncing := range p.listeners {
		switch {
		case !sync:
			// non-sync messages are delivered to every listener
			listener.add(obj)
		case isSyncing:
			// sync messages are delivered to every syncing listener
			listener.add(obj)
		default:
			// skipping a sync obj for a non-syncing listener
		}
	}
}

// sharedProcessorRunHook can be used inside tests to execute additional code
// at the start of sharedProcessor.run.
var sharedProcessorRunHook atomic.Pointer[func()]

func (p *sharedProcessor) run(ctx context.Context) {
	func() {
		hook := sharedProcessorRunHook.Load()
		if hook != nil {
			(*hook)()
		}
		// Changing listenersStarted needs a write lock.
		p.listenersLock.Lock()
		defer p.listenersLock.Unlock()
		for listener := range p.listeners {
			p.wg.Start(listener.run)
			p.wg.Start(listener.pop)
		}
		p.listenersStarted = true
		p.listenersRCond.Signal()
	}()
	<-ctx.Done()

	p.listenersLock.Lock()
	defer p.listenersLock.Unlock()
	for listener := range p.listeners {
		close(listener.addCh) // Tell .pop() to stop. .pop() will tell .run() to stop
	}

	// Wipe out list of listeners since they are now closed
	// (processorListener cannot be re-used)
	p.listeners = nil

	// Reset to false since no listeners are running
	p.listenersStarted = false