mirror of
https://github.com/christianselig/apollo-backend
synced 2024-11-13 07:27:43 +00:00
make sure we don't run jobs that are too old
This commit is contained in:
parent
fc0d290735
commit
199643449d
1 changed files with 18 additions and 14 deletions
|
@ -157,20 +157,6 @@ func (nc *notificationsConsumer) Consume(delivery rmq.Delivery) {
|
|||
_ = nc.statsd.Incr("apollo.consumer.executions", notificationTags, 0.1)
|
||||
}()
|
||||
|
||||
// Measure queue latency
|
||||
key := fmt.Sprintf("locks:accounts:%s", id)
|
||||
ttl := nc.redis.PTTL(ctx, key).Val()
|
||||
age := (domain.NotificationCheckTimeout - ttl)
|
||||
_ = nc.statsd.Histogram("apollo.dequeue.latency", float64(age.Milliseconds()), notificationTags, 0.1)
|
||||
|
||||
defer func() {
|
||||
if err := nc.redis.Del(ctx, key).Err(); err != nil {
|
||||
logger.Error("failed to remove account lock", zap.Error(err), zap.String("key", key))
|
||||
}
|
||||
}()
|
||||
|
||||
logger.Debug("starting job")
|
||||
|
||||
defer func(ctx context.Context) {
|
||||
_, span := nc.tracer.Start(ctx, "queue:ack")
|
||||
defer span.End()
|
||||
|
@ -182,6 +168,24 @@ func (nc *notificationsConsumer) Consume(delivery rmq.Delivery) {
|
|||
}
|
||||
}(ctx)
|
||||
|
||||
// Measure queue latency
|
||||
key := fmt.Sprintf("locks:accounts:%s", id)
|
||||
ttl := nc.redis.PTTL(ctx, key).Val()
|
||||
if ttl == 0 {
|
||||
logger.Debug("job is too old, skipping")
|
||||
return
|
||||
}
|
||||
age := (domain.NotificationCheckTimeout - ttl)
|
||||
_ = nc.statsd.Histogram("apollo.dequeue.latency", float64(age.Milliseconds()), notificationTags, 0.1)
|
||||
|
||||
defer func() {
|
||||
if err := nc.redis.Del(ctx, key).Err(); err != nil {
|
||||
logger.Error("failed to remove account lock", zap.Error(err), zap.String("key", key))
|
||||
}
|
||||
}()
|
||||
|
||||
logger.Debug("starting job")
|
||||
|
||||
account, err := nc.accountRepo.GetByRedditID(ctx, id)
|
||||
if err != nil {
|
||||
if err != domain.ErrNotFound {
|
||||
|
|
Loading…
Reference in a new issue