Function TestChannelCacheBypass

db/channel_cache_test.go:467–518

TestChannelCacheBypass validates that the bypass 'cache' is used when the cache max_num_channels is reached. To force this scenario, HWM is set to 100%, which effectively disables compaction.

(t *testing.T)

// TestChannelCacheBypass validates that the bypass 'cache' is used when the cache max_num_channels is reached.
// To force this scenario, HWM is set to 100%, which effectively disables compaction.
func TestChannelCacheBypass(t *testing.T) {
	base.SetUpTestLogging(t, base.LevelWarn, base.KeyCache)

	// Define cache with max channels 20, watermarks 50/100
	options := DefaultCacheOptions().ChannelCacheOptions
	options.MaxNumChannels = 20
	options.CompactHighWatermarkPercent = 100
	options.CompactLowWatermarkPercent = 50

	stats, err := base.NewSyncGatewayStats()
	require.NoError(t, err)
	dbstats, err := stats.NewDBStats("", false, false, false, nil, nil)
	require.NoError(t, err)
	testStats := dbstats.Cache()
	queryHandler := &testQueryHandler{}
	activeChannelStat := &base.SgwIntStat{}
	activeChannels := channels.NewActiveChannels(activeChannelStat)
	ctx := base.TestCtx(t)
	cache, err := newChannelCache(ctx, "testDb", options, queryHandler.asFactory, activeChannels, testStats)
	require.NoError(t, err, "Background task error whilst creating channel cache")
	defer cache.Stop(ctx)

	channelCount := 100
	// define channel set
	channelNames := make([]string, 0)
	for i := 1; i <= channelCount; i++ {
		channelName := fmt.Sprintf("chan_%d", i)
		channelNames = append(channelNames, channelName)
	}

	// Seed the query handler with a single doc that's in all the channels
	queryEntry := testLogEntryForChannels(1, channelNames)
	queryHandler.seedEntries(LogEntries{queryEntry})

	// Send entry to the cache.  Don't reuse queryEntry here, as AddToCache strips out the channels property
	logEntry := testLogEntryForChannels(1, channelNames)
	cache.AddToCache(ctx, logEntry)

	// Issue queries for all channels.  First 20 should end up in the cache, remaining 80 should trigger bypass
	for c := 1; c <= channelCount; c++ {
		channel := channels.NewID(fmt.Sprintf("chan_%d", c), base.DefaultCollectionID)
		options := getChangesOptionsWithCtxOnly(t)
		changes, err := cache.GetChanges(base.TestCtx(t), channel, options)
		assert.NoError(t, err, fmt.Sprintf("Error getting changes for channel %q", channel))
		assert.True(t, len(changes) == 1, "Expected one change per channel")
	}

	// check bypass count stat
	bypassCountStat := testStats.ChannelCacheBypassCount
	require.NotNil(t, bypassCountStat)
	assert.Equal(t, 80, int(bypassCountStat.Value()))
}

func waitForCompaction(cache *channelCacheImpl) (compactionComplete bool) {
	for i := 0; i <= 10; i++ {