Prepends an array of entries to this one, skipping ones that I already have. The new array needs to overlap with my current log, i.e. must contain the same sequence as the oldest entry in the cache (c.logs[0]), otherwise nothing will be added because the method can't confirm that there are no missin
(logCtx context.Context, changes LogEntries, changesValidFrom uint64, changesValidTo uint64)
| 580 | // |
| 581 | // Returns the number of entries actually prepended. |
| 582 | func (c *singleChannelCacheImpl) prependChanges(logCtx context.Context, changes LogEntries, changesValidFrom uint64, changesValidTo uint64) int { |
| 583 | c.lock.Lock() |
| 584 | defer c.lock.Unlock() |
| 585 | |
| 586 | // If set of changes to prepend is empty, check whether validFrom should be updated |
| 587 | if len(changes) == 0 { |
| 588 | if changesValidFrom < c.validFrom && changesValidTo >= c.validFrom { |
| 589 | base.DebugfCtx(logCtx, base.KeyCache, " changesValidFrom (%d) < c.validFrom < changesValidTo (%d), setting c.validFrom from %v -> %v for %q", |
| 590 | changesValidFrom, changesValidTo, c.validFrom, changesValidFrom, base.UD(c.channelID)) |
| 591 | c.validFrom = changesValidFrom |
| 592 | } |
| 593 | return 0 |
| 594 | } |
| 595 | |
| 596 | // Ensure changes are valid to the cache's validFrom, otherwise unsafe to prepend |
| 597 | if changesValidTo < c.validFrom { |
| 598 | return 0 |
| 599 | } |
| 600 | |
| 601 | // If my cache is empty, just copy the new changes |
| 602 | if len(c.logs) == 0 { |
| 603 | if excess := len(changes) - c.options.ChannelCacheMaxLength; excess > 0 { |
| 604 | changes = changes[excess:] |
| 605 | changesValidFrom = changes[0].Sequence |
| 606 | } |
| 607 | c.logs = make(LogEntries, len(changes)) |
| 608 | copy(c.logs, changes) |
| 609 | base.InfofCtx(logCtx, base.KeyCache, " Initialized cache of %q with %d entries from query (#%d--#%d)", |
| 610 | base.UD(c.channelID), len(changes), changes[0].Sequence, changes[len(changes)-1].Sequence) |
| 611 | |
| 612 | for _, change := range changes { |
| 613 | c.cachedDocIDs[change.DocID] = struct{}{} |
| 614 | c.UpdateCacheUtilization(change, 1) |
| 615 | } |
| 616 | |
| 617 | c.validFrom = changesValidFrom |
| 618 | return len(changes) |
| 619 | |
| 620 | } |
| 621 | |
| 622 | // Prepending changes to a non-empty cache |
| 623 | // Check whether there's capacity to prepend |
| 624 | cacheCapacity := c.options.ChannelCacheMaxLength - len(c.logs) |
| 625 | if cacheCapacity <= 0 { |
| 626 | return 0 |
| 627 | } |
| 628 | |
| 629 | // Check whether the results to prepend are contiguous with the cache |
| 630 | if changesValidFrom >= c.validFrom { |
| 631 | return 0 |
| 632 | } |
| 633 | |
| 634 | // Iterate backward over changes set, building set to prepend. |
| 635 | // - Don't prepend any sequence values already in the cache (later than c.validFrom) |
| 636 | // - Ignore docIDs already in the cache |
| 637 | // - Stop when we have enough to fill to ChannelCacheMaxLength (or run out of query results) |
| 638 | entriesToPrepend := make(LogEntries, 0, cacheCapacity) |
| 639 | for i := len(changes) - 1; i >= 0; i-- { |