Method runOnce

internal/maintenance/queue_cleaner.go:155–197

(ctx context.Context)

}

func (s *QueueCleaner) runOnce(ctx context.Context) (*queueCleanerRunOnceResult, error) {
	res := &queueCleanerRunOnceResult{QueuesDeleted: make([]string, 0, 10)}

	for {
		// Wrapped in a function so that defers run as expected.
		queuesDeleted, err := func() ([]string, error) {
			ctx, cancelFunc := context.WithTimeout(ctx, riversharedmaintenance.TimeoutDefault)
			defer cancelFunc()

			queuesDeleted, err := s.exec.QueueDeleteExpired(ctx, &riverdriver.QueueDeleteExpiredParams{
				Max:              s.batchSize(),
				Schema:           s.Config.Schema,
				UpdatedAtHorizon: time.Now().Add(-s.Config.RetentionPeriod),
			})
			if err != nil {
				return nil, fmt.Errorf("error deleting expired queues: %w", err)
			}

			s.reducedBatchSizeBreaker.ResetIfNotOpen()

			return queuesDeleted, nil
		}()
		if err != nil {
			if errors.Is(err, context.DeadlineExceeded) {
				s.reducedBatchSizeBreaker.Trip()
			}

			return nil, err
		}

		s.TestSignals.DeletedBatch.Signal(struct{}{})

		res.QueuesDeleted = append(res.QueuesDeleted, queuesDeleted...)
		// Deleted was less than query `LIMIT` which means work is done.
		if len(queuesDeleted) < s.batchSize() {
			break
		}

		serviceutil.CancellableSleep(ctx, randutil.DurationBetween(riversharedmaintenance.BatchBackoffMin, riversharedmaintenance.BatchBackoffMax))
	}

	return res, nil
}