Method processEntry

db/change_cache.go:755–833

processEntry handles a newly-arrived LogEntry and returns the changes channels from this revision. This can be any existing, removed or newly added channels. Its possible for channels slice returned to have duplicates in it. It is the callers responsibility to de-duplicate before notifying any chang

(ctx context.Context, change *LogEntry)

// This can be any existing, removed or newly added channels. Its possible for channels slice returned to have duplicates
// in it. It is the callers responsibility to de-duplicate before notifying any changes.
func (c *changeCache) processEntry(ctx context.Context, change *LogEntry) []channels.ID {
	c.lock.Lock()
	defer c.lock.Unlock()
	if c.logsDisabled {
		return nil
	}

	sequence := change.Sequence
	if change.Sequence > c.internalStats.highSeqFeed {
		c.internalStats.highSeqFeed = change.Sequence
	}

	// Duplicate handling - there are a few cases where processEntry can be called multiple times for a sequence:
	//   - recentSequences for rapidly updated documents
	//   - principal mutations that don't increment sequence
	// We can cancel processing early in these scenarios.
	// Check if this is a duplicate of an already processed sequence
	if sequence < c.nextSequence && !change.Skipped {
		// check for presence in skippedSeqs, it's possible that change.skipped can be marked false in recent sequence handling
		// but this change is subsequently pushed to skipped before acquiring cache mutex in this function
		if !c.WasSkipped(sequence) {
			base.DebugfCtx(ctx, base.KeyCache, "  Ignoring duplicate of #%d", sequence)
			return nil
		} else {
			change.Skipped = true
		}
	}

	// Check if this is a duplicate of a pending sequence
	if _, found := c.receivedSeqs[sequence]; found {
		base.DebugfCtx(ctx, base.KeyCache, "  Ignoring duplicate of #%d", sequence)
		return nil
	}
	c.receivedSeqs[sequence] = struct{}{}

	changedChannels := make([]channels.ID, 0, len(change.Channels))
	if sequence == c.nextSequence || c.nextSequence == 0 {
		// This is the expected next sequence so we can add it now:
		changedChannels = c._addToCache(ctx, change)
		// Also add any pending sequences that are now contiguous:
		changedChannels = append(changedChannels, c._addPendingLogs(ctx)...)
	} else if sequence > c.nextSequence {
		// There's a missing sequence (or several), so put this one on ice until it arrives:
		heap.Push(&c.pendingLogs, change)
		numPending := len(c.pendingLogs)
		c.internalStats.pendingSeqLen = numPending
		if base.LogDebugEnabled(ctx, base.KeyCache) {
			base.DebugfCtx(ctx, base.KeyCache, "  Deferring #%d (%d now waiting for #%d...#%d) doc %q / %q",
				sequence, numPending, c.nextSequence, c.pendingLogs[0].Sequence-1, base.UD(change.DocID), change.RevID)
		}
		// Update max pending high watermark stat
		if numPending > c.internalStats.maxPending {
			c.internalStats.maxPending = numPending
		}

		if numPending > c.options.CachePendingSeqMaxNum {
			// Too many pending; add the oldest one:
			changedChannels = append(changedChannels, c._addPendingLogs(ctx)...)