<commit_before>package db

import (
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/couchbase/sync_gateway/base"
	"github.com/couchbase/sync_gateway/channels"
)

var (
	DefaultChannelCacheMinLength = 50               // Keep at least this many entries in cache
	DefaultChannelCacheMaxLength = 500              // Don't put more than this many entries in cache
	DefaultChannelCacheAge       = 60 * time.Second // Keep entries at least this long
)

const NoSeq = uint64(0x7FFFFFFFFFFFFFFF)

// Minimizes need to perform GSI/View queries to respond to the changes feed by keeping a per-channel
// cache of the most recent changes in the channel.  Changes might be received over DCP won't necessarily
// be in sequence order, but the changes are buffered and re-ordered before inserting into the cache, and
// so the cache is always kept in ascending order by sequence.  (this is the global sync sequence, not the
// per-vbucket sequence).
//
// Uniqueness guarantee: a given document should only have _one_ entry in the cache, which represents the
// most recent revision (the revision with the highest sequence number).
//
// Completeness guarantee: the cache is guaranteed to have _every_ change in a channel from the validFrom sequence
// up to the current known latest change in that channel.  Eg, there are no gaps/holes.
//
// The validFrom state variable tracks the oldest sequence for which the cache is complete.  This may be earlier
// than the oldest entry in the cache.  Used to determine whether a GSI/View query is required for a given
// getChanges request.
//
// Shortly after startup and receiving a few changes from DCP, the cache might look something like this:
//
// ┌───────────────────────────────────────────────────────────────────────────────────────┐
// │                       Changes for Channel X (CBServer + Cache)                        │
// │   ┌─────────────────────────────────────────────────┐                                 │
// │   │            Cache Subset (MaxSize=3)             │                                 │
// │   │                                                 │                                 │
// │   │          ┌─────────┐  ┌─────────┐               │                                 │
// │   │          │ Seq: 8  │  │ Seq: 25 │               │                                 │
// │   │    ▲     │ Doc: A  │  │ Doc: B  │               │                                 │
// │   │    │     │ Rev: 1  │  │ Rev: 4  │               │                                 │
// │   │    │     └─────────┘  └─────────┘               │                                 │
// │   │                                                 │                                 │
// │   │       ValidFrom = Seq 5                         │                                 │
// │   │     (_sync:Seq at Startup)                      │                                 │
// │   │                                                 │                                 │
// │   └─────────────────────────────────────────────────┘                                 │
// │                                                                                       │
// └───────────────────────────────────────────────────────────────────────────────────────┘
//
// All changes for the channel currently fit in the cache.  The validFrom is set to the value of the _sync:Seq that
// was read on startup.
//
// Later after more data has been added that will fit in the cache, it will look more like this:
//
// ┌───────────────────────────────────────────────────────────────────────────────────────┐
// │                       Changes for Channel X (CBServer + Cache)                        │
// │                                     ┌─────────────────────────────────────────────┐   │
// │                                     │          Cache Subset (MaxSize=3)           │   │
// │                                     │                                             │   │
// │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │         ┌─────────┐┌─────────┐ ┌─────────┐  │   │
// │ │ Seq: 8  │ │ Seq: 25 │ │Seq: ... │ │         │Seq: 6002││Seq: 7022│ │Seq: 7027│  │   │
// │ │ Doc: A  │ │ Doc: B  │ │Doc: ... │ │   ▲     │ Doc: A  ││ Doc: H  │ │ Doc: M  │  │   │
// │ │ Rev: 1  │ │ Rev: 4  │ │Rev: ... │ │   │     │Rev: 345 ││ Rev: 4  │ │ Rev: 47 │  │   │
// │ └─────────┘ └─────────┘ └─────────┘ │   │     └─────────┘└─────────┘ └─────────┘  │   │
// │                                     │                                             │   │
// │                                     │  ValidFrom = Seq 5989 (cache known          │   │
// │                                     │  to be complete + valid from seq)           │   │
// │                                     │                                             │   │
// │                                     └─────────────────────────────────────────────┘   │
// │                                                                                       │
// └───────────────────────────────────────────────────────────────────────────────────────┘
//
// If a calling function wanted to get all of the changes for the channel since Seq 3000, they would
// be forced to issue a GSI/View backfill query for changes is in the channel between 3000 and 5989,
// since the cache is only validFrom the sequence 5989.  In this case, none of those backfilled values
// will fit in the cache, which is already full, however if it did have capacity available, then those
// values would be _prepended_ to the front of the cache and the validFrom sequence would be lowered
// to account for the fact that it's now valid from a lower sequence value.  Entries that violated
// the guarantees described above would possibly be discarded.
type channelCache struct {
	channelName      string               // The channel name, duh
	context          *DatabaseContext     // Database connection (used for view queries)
	logs             LogEntries           // Log entries in sequence order
	validFrom        uint64               // First sequence that logs is valid for, not necessarily the seq number of a change entry.
	lock             sync.RWMutex         // Controls access to logs, validFrom
	viewLock         sync.Mutex           // Ensures only one view query is made at a time
	lateLogs         []*lateLogEntry      // Late arriving LogEntries, stored in the order they were received
	lastLateSequence uint64               // Used for fast check of whether listener has the latest
	lateLogLock      sync.RWMutex         // Controls access to lateLogs
	options          *ChannelCacheOptions // Cache size/expiry settings
	cachedDocIDs     map[string]struct{}
}

func newChannelCache(context *DatabaseContext, channelName string, validFrom uint64) *channelCache {
	cache := &channelCache{context: context, channelName: channelName, validFrom: validFrom}
	cache.initializeLateLogs()
	cache.cachedDocIDs = make(map[string]struct{})
	cache.options = &ChannelCacheOptions{
		ChannelCacheMinLength: DefaultChannelCacheMinLength,
		ChannelCacheMaxLength: DefaultChannelCacheMaxLength,
		ChannelCacheAge:       DefaultChannelCacheAge,
	}
	cache.logs = make(LogEntries, 0, cache.options.ChannelCacheMaxLength)

	return cache
}

func newChannelCacheWithOptions(context *DatabaseContext, channelName string, validFrom uint64, options CacheOptions) *channelCache {
	cache := newChannelCache(context, channelName, validFrom)

	// Update cache options when present
	if options.ChannelCacheMinLength > 0 {
		cache.options.ChannelCacheMinLength = options.ChannelCacheMinLength
	}

	if options.ChannelCacheMaxLength > 0 {
		cache.options.ChannelCacheMaxLength = options.ChannelCacheMaxLength<commit_msg>Remove the default capacity for channel cache log slice. Reduces empty channel cache memory overhead from 4392 bytes to 296 bytes (#3944)<commit_after>package db

import (
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/couchbase/sync_gateway/base"
	"github.com/couchbase/sync_gateway/channels"
)

var (
	DefaultChannelCacheMinLength = 50               // Keep at least this many entries in cache
	DefaultChannelCacheMaxLength = 500              // Don't put more than this many entries in cache
	DefaultChannelCacheAge       = 60 * time.Second // Keep entries at least this long
)

const NoSeq = uint64(0x7FFFFFFFFFFFFFFF)

// Minimizes need to perform GSI/View queries to respond to the changes feed by keeping a per-channel
// cache of the most recent changes in the channel.  Changes might be received over DCP won't necessarily
// be in sequence order, but the changes are buffered and re-ordered before inserting into the cache, and
// so the cache is always kept in ascending order by sequence.  (this is the global sync sequence, not the
// per-vbucket sequence).
//
// Uniqueness guarantee: a given document should only have _one_ entry in the cache, which represents the
// most recent revision (the revision with the highest sequence number).
//
// Completeness guarantee: the cache is guaranteed to have _every_ change in a channel from the validFrom sequence
// up to the current known latest change in that channel.  Eg, there are no gaps/holes.
//
// The validFrom state variable tracks the oldest sequence for which the cache is complete.  This may be earlier
// than the oldest entry in the cache.  Used to determine whether a GSI/View query is required for a given
// getChanges request.
//
// Shortly after startup and receiving a few changes from DCP, the cache might look something like this:
//
// ┌───────────────────────────────────────────────────────────────────────────────────────┐
// │                       Changes for Channel X (CBServer + Cache)                        │
// │   ┌─────────────────────────────────────────────────┐                                 │
// │   │            Cache Subset (MaxSize=3)             │                                 │
// │   │                                                 │                                 │
// │   │          ┌─────────┐  ┌─────────┐               │                                 │
// │   │          │ Seq: 8  │  │ Seq: 25 │               │                                 │
// │   │    ▲     │ Doc: A  │  │ Doc: B  │               │                                 │
// │   │    │     │ Rev: 1  │  │ Rev: 4  │               │                                 │
// │   │    │     └─────────┘  └─────────┘               │                                 │
// │   │                                                 │                                 │
// │   │       ValidFrom = Seq 5                         │                                 │
// │   │     (_sync:Seq at Startup)                      │                                 │
// │   │                                                 │                                 │
// │   └─────────────────────────────────────────────────┘                                 │
// │                                                                                       │
// └───────────────────────────────────────────────────────────────────────────────────────┘
//
// All changes for the channel currently fit in the cache.  The validFrom is set to the value of the _sync:Seq that
// was read on startup.
//
// Later after more data has been added that will fit in the cache, it will look more like this:
//
// ┌───────────────────────────────────────────────────────────────────────────────────────┐
// │                       Changes for Channel X (CBServer + Cache)                        │
// │                                     ┌─────────────────────────────────────────────┐   │
// │                                     │          Cache Subset (MaxSize=3)           │   │
// │                                     │                                             │   │
// │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │         ┌─────────┐┌─────────┐ ┌─────────┐  │   │
// │ │ Seq: 8  │ │ Seq: 25 │ │Seq: ... │ │         │Seq: 6002││Seq: 7022│ │Seq: 7027│  │   │
// │ │ Doc: A  │ │ Doc: B  │ │Doc: ... │ │   ▲     │ Doc: A  ││ Doc: H  │ │ Doc: M  │  │   │
// │ │ Rev: 1  │ │ Rev: 4  │ │Rev: ... │ │   │     │Rev: 345 ││ Rev: 4  │ │ Rev: 47 │  │   │
// │ └─────────┘ └─────────┘ └─────────┘ │   │     └─────────┘└─────────┘ └─────────┘  │   │
// │                                     │                                             │   │
// │                                     │  ValidFrom = Seq 5989 (cache known          │   │
// │                                     │  to be complete + valid from seq)           │   │
// │                                     │                                             │   │
// │                                     └─────────────────────────────────────────────┘   │
// │                                                                                       │
// └───────────────────────────────────────────────────────────────────────────────────────┘
//
// If a calling function wanted to get all of the changes for the channel since Seq 3000, they would
// be forced to issue a GSI/View backfill query for changes is in the channel between 3000 and 5989,
// since the cache is only validFrom the sequence 5989.  In this case, none of those backfilled values
// will fit in the cache, which is already full, however if it did have capacity available, then those
// values would be _prepended_ to the front of the cache and the validFrom sequence would be lowered
// to account for the fact that it's now valid from a lower sequence value.  Entries that violated
// the guarantees described above would possibly be discarded.
type channelCache struct {
	channelName      string               // The channel name, duh
	context          *DatabaseContext     // Database connection (used for view queries)
	logs             LogEntries           // Log entries in sequence order
	validFrom        uint64               // First sequence that logs is valid for, not necessarily the seq number of a change entry.
	lock             sync.RWMutex         // Controls access to logs, validFrom
	viewLock         sync.Mutex           // Ensures only one view query is made at a time
	lateLogs         []*lateLogEntry      // Late arriving LogEntries, stored in the order they were received
	lastLateSequence uint64               // Used for fast check of whether listener has the latest
	lateLogLock      sync.RWMutex         // Controls access to lateLogs
	options          *ChannelCacheOptions // Cache size/expiry settings
	cachedDocIDs     map[string]struct{}
}

func newChannelCache(context *DatabaseContext, channelName string, validFrom uint64) *channelCache {
	cache := &channelCache{context: context, channelName: channelName, validFrom: validFrom}
	cache.initializeLateLogs()
	cache.cachedDocIDs = make(map[string]struct{})
	cache.options = &ChannelCacheOptions{
		ChannelCacheMinLength: DefaultChannelCacheMinLength,
		ChannelCacheMaxLength: DefaultChannelCacheMaxLength,
		ChannelCacheAge:       DefaultChannelCacheAge,
	}
	cache.logs = make(LogEntries, 0)

	return cache
}

func newChannelCacheWithOptions(context *DatabaseContext, channelName string, validFrom uint64, options CacheOptions) *channelCache {
	cache := newChannelCache(context, channelName, validFrom)

	// Update cache options when present
	if options.ChannelCacheMinLength > 0 {
		cache.options.ChannelCacheMinLength = options.ChannelCacheMinLength
	}

	if options.ChannelCacheMaxLength > 0 {
		cache.options.ChannelCacheMaxLength = options.ChannelCacheMaxLength
{'fraction_non_alphanumeric': 0.16491589869175743, 'fraction_numerical': 0.01011793516788039, 'mean_word_length': 1.6054398925453324, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 100, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5679416', 'n_tokens_mistral': 4489, 'n_tokens_neox': 3902, 'n_words': 1603}