| 169 | } |
| 170 | |
| 171 | func (s *RedisQueue) Push(ctx context.Context, data []byte, highPriority bool, minTargetBlock, maxTargetBlock uint64) error { |
| 172 | currentBlock := atomic.LoadUint64(s.currentBlock) |
| 173 | |
| 174 | if maxTargetBlock <= currentBlock { |
| 175 | metrics.IncSbundlesReceivedStale() |
| 176 | s.log.Debug("max target block is less than current block, skipping", zap.Uint64("max_target_block", maxTargetBlock), zap.Uint64("current_block", currentBlock)) |
| 177 | return ErrStaleItem |
| 178 | } |
| 179 | |
| 180 | // we schedule items for the next block |
| 181 | if nextBlock := currentBlock + 1; minTargetBlock < nextBlock { |
| 182 | minTargetBlock = nextBlock |
| 183 | } |
| 184 | |
| 185 | args := packArgs{ |
| 186 | data: data, |
| 187 | minTargetBlock: minTargetBlock, |
| 188 | maxTargetBlock: maxTargetBlock, |
| 189 | highPriority: highPriority, |
| 190 | timestamp: time.Now(), |
| 191 | iteration: 0, |
| 192 | } |
| 193 | err := s.pushToQueue(ctx, args) |
| 194 | if err != nil { |
| 195 | if errors.Is(err, ErrQueueFull) { |
| 196 | metrics.IncQueueFullSbundles() |
| 197 | } |
| 198 | return err |
| 199 | } |
| 200 | s.log.Debug("pushed to queue", zap.Uint64("min_target_block", minTargetBlock), zap.Uint64("max_target_block", maxTargetBlock), zap.Bool("high_priority", highPriority)) |
| 201 | return nil |
| 202 | } |
| 203 | |
| 204 | // queuedItems returns number of items in the queue that should be eventually processed |
| 205 | // processable are the items that can be processed now |