Method releaseUnusedSequenceRange

db/change_cache.go:619–648

releaseUnusedSequenceRange will handle unused sequence range arriving over DCP. It will batch remove from skipped or push a range to pending sequences, or both.

(ctx context.Context, fromSequence uint64, toSequence uint64, timeReceived channels.FeedTimestamp)

// releaseUnusedSequenceRange will handle unused sequence range arriving over DCP. It will batch remove from skipped or
// push a range to pending sequences, or both.
func (c *changeCache) releaseUnusedSequenceRange(ctx context.Context, fromSequence uint64, toSequence uint64, timeReceived channels.FeedTimestamp) {

	base.InfofCtx(ctx, base.KeyCache, "Received #%d-#%d (unused sequence range)", fromSequence, toSequence)

	allChangedChannels := channels.SetOfNoValidate(unusedSeqChannelID)

	// if range is single value, just run sequence through process entry and return early
	if fromSequence == toSequence {
		change := &LogEntry{
			Sequence:       toSequence,
			TimeReceived:   timeReceived,
			UnusedSequence: true,
		}
		changedChannels := c.processEntry(ctx, change)
		channelSet := channels.SetFromArrayNoValidate(changedChannels)
		allChangedChannels = allChangedChannels.Update(channelSet)
		if c.notifyChangeFunc != nil {
			c.notifyChangeFunc(ctx, allChangedChannels)
		}
		return
	}

	// push unused range to either pending or skipped lists based on current state of the change cache
	changedChannels := c.processUnusedRange(ctx, fromSequence, toSequence, timeReceived)
	allChangedChannels = allChangedChannels.Update(channels.SetFromArrayNoValidate(changedChannels))

	if c.notifyChangeFunc != nil {
		c.notifyChangeFunc(ctx, allChangedChannels)
	}
}

// processUnusedRange handles pushing unused range to pending or skipped lists
func (c *changeCache) processUnusedRange(ctx context.Context, fromSequence, toSequence uint64, timeReceived channels.FeedTimestamp) []channels.ID {