func (writer *messageWriter) Finish(bytesScanned, bytesProcessed int64) error {
	select {
	case <-writer.doneCh:
		return fmt.Errorf("messageWriter is done")
	default:
		writer.finBytesScanned = bytesScanned
		writer.finBytesProcessed = bytesProcessed
		close(writer.payloadCh)
		// Wait until the `start` go-routine is done.
		<-writer.doneCh
		return nil
	}
}

				}
				return
			}
		}
	}()
	return func(err error) {
		if doneCh == nil {
			return
		}

		// Indicate we are ready to write.
		doneCh <- err

		// Wait for channel to be closed so we don't race on writes.
		<-doneCh

		// Clear so we can be called multiple times without crashing.
		doneCh = nil
	}, &closeNotifier{rc: r.Body, done: bodyDoneCh}
}

		return nil
	}
	syncNodeStatus := func() {}

	var shutdownChan chan bool
	var shutdownChanMut sync.Mutex
	var connChan = make(chan struct{}, 1)

	lock.Lock()
	systemDbus = func() (dbusInhibiter, error) {
		defer func() {
			connChan <- struct{}{}
		}()
		ch := make(chan bool)
		shutdownChanMut.Lock()
		shutdownChan = ch
		shutdownChanMut.Unlock()

				}
			}(pod, group)
		}

		var (
			doneCh = make(chan struct{})
			timer  = m.clock.NewTimer(time.Duration(group.ShutdownGracePeriodSeconds) * time.Second)
		)
		go func() {
			defer close(doneCh)
			wg.Wait()
		}()

		select {
		case <-doneCh:
			timer.Stop()
		case <-timer.C():

		return ps.Participating && ps.Info.Status == rebalStarted
	}
	return false
}

func (z *erasureServerPools) rebalanceBuckets(ctx context.Context, poolIdx int) (err error) {
	doneCh := make(chan error, 1)
	defer xioutil.SafeClose(doneCh)

	// Save rebalance.bin periodically.
	go func() {
		// Update rebalance.bin periodically once every 5-10s, chosen randomly

	defer cancel()

	expectedError := errors.New("Expected error")
	var attempt int
	f := ConditionFunc(func() (bool, error) {
		attempt++
		return false, expectedError
	})

	doneCh := make(chan struct{})
	go func() {
		defer close(doneCh)
		if err := loopConditionUntilContext(ctx, timerWithClock(backoff.Timer(), fakeClock), false, true, f.WithContext()); err == nil || err != expectedError {
			t.Errorf("unexpected error: %v", err)
		}

type signalInterceptingTestStep struct {
	doneCh chan struct{}
}

func (ts signalInterceptingTestStep) done() <-chan struct{} {
	return ts.doneCh
}
func (ts signalInterceptingTestStep) execute(fn func()) {
	defer close(ts.doneCh)
	fn()
}
func newSignalInterceptingTestStep() *signalInterceptingTestStep {
	return &signalInterceptingTestStep{
		doneCh: make(chan struct{}),
	}
}

	lookupGroupCtx, lookupGroupCancel := context.WithCancel(withUnexpiredValuesPreserved(ctx))

	lookupKey := network + "\000" + host
	dnsWaitGroup.Add(1)
	ch := r.getLookupGroup().DoChan(lookupKey, func() (any, error) {
		return testHookLookupIP(lookupGroupCtx, resolverFunc, network, host)
	})

	dnsWaitGroupDone := func(ch <-chan singleflight.Result, cancelFn context.CancelFunc) {
		<-ch