Function TestProcessSkippedEntry

db/change_cache_test.go:2018–2087

TestProcessSkippedEntry: - Creates change cache with minimal pending seq wait time to push sequences to skipped quick - Push a sequence higher than expected to cache - Have NewTestProcessEntryFeed push sequences to the cache and assert that the current skipped sequence count decrements each time - I

(t *testing.T)

//   - Assert that the length of slice stat is equal to the length of the skipped sequence slice + assert that number of
//     sequences skipped overall stat is correct
func TestProcessSkippedEntry(t *testing.T) {
	base.SetUpTestLogging(t, base.LevelDebug, base.KeyAll)

	ctx := base.TestCtx(t)
	bucket := base.GetTestBucket(t)
	dbContext, err := NewDatabaseContext(ctx, "db", bucket, false, DatabaseContextOptions{
		Scopes: GetScopesOptions(t, bucket, 1),
	})
	require.NoError(t, err)
	defer dbContext.Close(ctx)

	ctx = dbContext.AddDatabaseLogContext(ctx)
	err = dbContext.StartOnlineProcesses(ctx)
	require.NoError(t, err)

	testChangeCache := &changeCache{}
	if err := testChangeCache.Init(ctx, dbContext, dbContext.channelCache, nil, &CacheOptions{
		CachePendingSeqMaxWait: 5 * time.Millisecond,
		CacheSkippedSeqMaxWait: 2 * time.Minute,
	}, dbContext.MetadataKeys); err != nil {
		log.Printf("Init failed for testChangeCache: %v", err)
		t.Fail()
	}

	if err := testChangeCache.Start(0); err != nil {
		log.Printf("Start error for testChangeCache: %v", err)
		t.Fail()
	}
	defer testChangeCache.Stop(ctx)
	require.NoError(t, err)

	feed := NewTestProcessEntryFeed(1, 5)

	// add cache entry that is higher than expected
	highEntry := &LogEntry{
		Sequence:     20,
		DocID:        fmt.Sprintf("doc_%d", 50),
		RevID:        "1-abcdefabcdefabcdef",
		TimeReceived: channels.NewFeedTimestampFromNow(),
	}
	_ = testChangeCache.processEntry(ctx, highEntry)

	// update cache stats for assertions
	testChangeCache.updateStats(ctx)

	// assert this pushes an entry on the skipped sequence slice
	require.EventuallyWithT(t, func(c *assert.CollectT) {
		assert.Equal(c, 1, testChangeCache.skippedSeqs.list.GetLength())
	}, time.Second*10, time.Millisecond*100)

	// process some sequences over cache
	currNumSkippedSeqs := dbContext.DbStats.CacheStats.NumCurrentSeqsSkipped.Value()
	for j := 0; j < 10; j++ {
		en := feed.Next()
		_ = testChangeCache.processEntry(ctx, en)
		testChangeCache.updateStats(ctx)
		assert.Equal(t, currNumSkippedSeqs-1, dbContext.DbStats.CacheStats.NumCurrentSeqsSkipped.Value())
		currNumSkippedSeqs = dbContext.DbStats.CacheStats.NumCurrentSeqsSkipped.Value()