processUnusedRange handles pushing unused range to pending or skipped lists
(ctx context.Context, fromSequence, toSequence uint64, timeReceived channels.FeedTimestamp)
| 649 | |
| 650 | // processUnusedRange handles pushing unused range to pending or skipped lists |
| 651 | func (c *changeCache) processUnusedRange(ctx context.Context, fromSequence, toSequence uint64, timeReceived channels.FeedTimestamp) []channels.ID { |
| 652 | c.lock.Lock() |
| 653 | defer c.lock.Unlock() |
| 654 | |
| 655 | var numSkipped int64 |
| 656 | var changedChannels []channels.ID |
| 657 | if toSequence < c.nextSequence { |
| 658 | // batch remove from skipped |
| 659 | numSkipped = c.skippedSeqs.processUnusedSequenceRangeAtSkipped(ctx, fromSequence, toSequence) |
| 660 | } else if fromSequence >= c.nextSequence { |
| 661 | // whole range to pending |
| 662 | c._pushRangeToPending(fromSequence, toSequence, timeReceived) |
| 663 | // unblock any pending sequences we can after new range(s) have been pushed to pending |
| 664 | changedChannels = append(changedChannels, c._addPendingLogs(ctx)...) |
| 665 | c.internalStats.pendingSeqLen = len(c.pendingLogs) |
| 666 | } else { |
| 667 | // An unused sequence range than includes c.nextSequence in the middle of the range |
| 668 | // isn't possible under normal processing - unused sequence ranges will normally be moved |
| 669 | // from pending to skipped in their entirety, as it's the processing of the pending sequence |
| 670 | // *after* the range that triggers the range to be skipped. A partial range in skipped means |
| 671 | // a duplicate entry with a sequence within the bounds of the range was previously present |
| 672 | // in pending. |
| 673 | base.WarnfCtx(ctx, "unused sequence range of #%d to %d contains duplicate sequences, will be ignored", fromSequence, toSequence) |
| 674 | } |
| 675 | if numSkipped == 0 { |
| 676 | c.db.BroadcastSlowMode.CompareAndSwap(true, false) |
| 677 | } |
| 678 | return changedChannels |
| 679 | } |
| 680 | |
| 681 | // _pushRangeToPending will push an unused sequence range to pendingLogs |
| 682 | func (c *changeCache) _pushRangeToPending(startSeq, endSeq uint64, timeReceived channels.FeedTimestamp) { |
no test coverage detected