Method running

kv/memberlist/propagation_tracker.go:119–155

(ctx context.Context)

}

func (t *PropagationDelayTracker) running(ctx context.Context) error {
	t.startupTime = time.Now().UnixMilli()
	level.Info(t.logger).Log("msg", "propagation delay tracker started", "beacon_interval", t.cfg.BeaconInterval, "beacon_lifetime", t.cfg.BeaconLifetime)

	// Start the goroutine to track beacon arrivals in real-time
	watchCtx, cancelWatch := context.WithCancel(ctx)
	defer cancelWatch()

	watchDone := make(chan struct{})
	go func() {
		defer close(watchDone)
		t.watchBeacons(watchCtx)
	}()
	defer func() {
		select {
		case <-watchDone:
		case <-time.After(10 * time.Second):
			level.Warn(t.logger).Log("msg", "timed out waiting for watch goroutine to finish")
		}
	}()

	// Start the beacon publish ticker. The first tick uses a random delay between 1ns and
	// beaconInterval to distribute beacon publishing over time when multiple processes
	// start simultaneously (e.g., during a rollout).
	initialDelay := 1 + time.Duration(rand.Int63n(int64(t.cfg.BeaconInterval)))
	stopTicker, tickerChan := timeutil.NewVariableTicker(initialDelay, t.cfg.BeaconInterval)
	defer stopTicker()

	for {
		select {
		case <-ctx.Done():
			return nil
		case <-tickerChan:
			t.onBeaconInterval(ctx)
		}
	}
}

// watchBeacons uses WatchKey to receive beacon updates in real-time and measure delay.
// It loops with backoff until the context is canceled.