(ctx context.Context, job CronJob)
| 122 | } |
| 123 | |
| 124 | func (c *Cron) run(ctx context.Context, job CronJob) { |
| 125 | //nolint:gocritic // We are a publisher in this function |
| 126 | ctx = dbauthz.AsUsagePublisher(ctx) |
| 127 | defer c.wg.Done() |
| 128 | for { |
| 129 | boundary, delay := nextTick(c.clock.Now(), job.Interval, job.Jitter) |
| 130 | |
| 131 | // Use a quartz timer so the wait honors ctx cancellation and |
| 132 | // tests can advance time deterministically. |
| 133 | timer := c.clock.NewTimer(delay, job.Name) |
| 134 | |
| 135 | select { |
| 136 | case <-ctx.Done(): |
| 137 | if !timer.Stop() { |
| 138 | // Drain the channel if the timer already fired. |
| 139 | <-timer.C |
| 140 | } |
| 141 | return |
| 142 | case <-timer.C: |
| 143 | } |
| 144 | |
| 145 | // Use the boundary (not wall-clock "now") for the stable ID |
| 146 | // so all replicas targeting the same boundary produce the |
| 147 | // same key. |
| 148 | stableID := string(job.EventType) + ":" + boundary.UTC().Format(cronDateFormat) |
| 149 | |
| 150 | // Skip if this bucket was already recorded — avoids running |
| 151 | // the potentially expensive heartbeat function for a |
| 152 | // duplicate. |
| 153 | exists, err := c.db.UsageEventExistsByID(ctx, stableID) |
| 154 | if err != nil { |
| 155 | c.log.Warn(ctx, "cron heartbeat existence check failed", |
| 156 | slog.F("job", job.Name), |
| 157 | slog.Error(err), |
| 158 | ) |
| 159 | continue |
| 160 | } |
| 161 | if exists { |
| 162 | c.log.Debug(ctx, "cron heartbeat already recorded, skipping", |
| 163 | slog.F("job", job.Name), |
| 164 | slog.F("id", stableID), |
| 165 | ) |
| 166 | continue |
| 167 | } |
| 168 | |
| 169 | event, err := job.Fn(ctx) |
| 170 | if err != nil { |
| 171 | c.log.Error(ctx, "cron heartbeat func failed", |
| 172 | slog.F("job", job.Name), |
| 173 | slog.Error(err), |
| 174 | ) |
| 175 | continue |
| 176 | } |
| 177 | |
| 178 | if event.EventType() != job.EventType { |
| 179 | c.log.Error(ctx, "cron heartbeat func returned wrong event type", |
| 180 | slog.F("job", job.Name), |
| 181 | slog.F("expected", job.EventType), |
no test coverage detected