Method waitForCatchUp

modules/livestore/live_store.go:513–576

(ctx context.Context)

}

func (s *LiveStore) waitForCatchUp(ctx context.Context) error {
	// If disabled (ReadinessTargetLag == 0), mark ready immediately
	// This preserves backward compatibility
	if s.cfg.ReadinessTargetLag == 0 {
		level.Info(s.logger).Log("msg", "catch-up waiting disabled (readiness_target_lag=0)")
		return nil
	}

	startTime := time.Now()

	ticker := time.NewTicker(time.Second) // Check every second
	defer ticker.Stop()

	level.Info(s.logger).Log(
		"msg", "waiting for Kafka catch-up",
		"target_lag", s.cfg.ReadinessTargetLag,
		"max_wait", s.cfg.ReadinessMaxWait,
	)

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
			elapsed := time.Since(startTime)

			// Check max wait timeout
			if elapsed >= s.cfg.ReadinessMaxWait {
				level.Warn(s.logger).Log(
					"msg", "max catch-up wait exceeded, proceeding anyway",
					"elapsed", elapsed,
					"max_wait", s.cfg.ReadinessMaxWait,
				)
				metricCatchUpDuration.Set(elapsed.Seconds())
				return nil
			}

			// Calculate current lag
			lag := s.calculateTimeLag(1000)
			if lag == nil {
				level.Debug(s.logger).Log("msg", "catch-up lag could not be determined, waiting")
				continue
			}

			level.Debug(s.logger).Log(
				"msg", "catch-up progress",
				"current_lag", *lag,
				"target_lag", s.cfg.ReadinessTargetLag,
				"elapsed", elapsed,
			)

			if *lag <= s.cfg.ReadinessTargetLag {
				level.Info(s.logger).Log(
					"msg", "caught up with Kafka",
					"final_lag", *lag,
					"target_lag", s.cfg.ReadinessTargetLag,
					"elapsed", elapsed,
				)