TestProcessSkippedEntryStats: - Creates change cache with minimal pending seq wait time to push sequences to skipped quick - Push a sequence higher than expected to cache - Push some sequences that are skipped onto the cache and assert that the stats: number of current skipped sequences, length of s
(t *testing.T)
| 2092 | // - Push some sequences that are skipped onto the cache and assert that the stats: number of current skipped sequences, |
| 2093 | // length of slice, cumulative number of skipped sequences and capacity of slice are all updated as expected |
| 2094 | func TestProcessSkippedEntryStats(t *testing.T) { |
| 2095 | base.SetUpTestLogging(t, base.LevelDebug, base.KeyAll) |
| 2096 | |
| 2097 | ctx := base.TestCtx(t) |
| 2098 | bucket := base.GetTestBucket(t) |
| 2099 | dbContext, err := NewDatabaseContext(ctx, "db", bucket, false, DatabaseContextOptions{ |
| 2100 | Scopes: GetScopesOptions(t, bucket, 1), |
| 2101 | }) |
| 2102 | require.NoError(t, err) |
| 2103 | defer dbContext.Close(ctx) |
| 2104 | |
| 2105 | ctx = dbContext.AddDatabaseLogContext(ctx) |
| 2106 | err = dbContext.StartOnlineProcesses(ctx) |
| 2107 | require.NoError(t, err) |
| 2108 | |
| 2109 | testChangeCache := &changeCache{} |
| 2110 | if err := testChangeCache.Init(ctx, dbContext, dbContext.channelCache, nil, &CacheOptions{ |
| 2111 | CachePendingSeqMaxWait: 5 * time.Millisecond, |
| 2112 | CacheSkippedSeqMaxWait: 2 * time.Minute, |
| 2113 | }, dbContext.MetadataKeys); err != nil { |
| 2114 | log.Printf("Init failed for testChangeCache: %v", err) |
| 2115 | t.Fail() |
| 2116 | } |
| 2117 | |
| 2118 | if err := testChangeCache.Start(0); err != nil { |
| 2119 | log.Printf("Start error for testChangeCache: %v", err) |
| 2120 | t.Fail() |
| 2121 | } |
| 2122 | defer testChangeCache.Stop(ctx) |
| 2123 | require.NoError(t, err) |
| 2124 | |
| 2125 | // add cache entry that is higher than expected |
| 2126 | highEntry := &LogEntry{ |
| 2127 | Sequence: 20, |
| 2128 | DocID: fmt.Sprintf("doc_%d", 50), |
| 2129 | RevID: "1-abcdefabcdefabcdef", |
| 2130 | TimeReceived: channels.NewFeedTimestampFromNow(), |
| 2131 | } |
| 2132 | _ = testChangeCache.processEntry(ctx, highEntry) |
| 2133 | |
| 2134 | // update cache stats for assertions |
| 2135 | testChangeCache.updateStats(ctx) |
| 2136 | |
| 2137 | // assert this pushes an entry on the skipped sequence slice |
| 2138 | require.EventuallyWithT(t, func(c *assert.CollectT) { |
| 2139 | assert.Equal(c, 1, testChangeCache.skippedSeqs.list.GetLength()) |
| 2140 | }, time.Second*10, time.Millisecond*100) |
| 2141 | |
| 2142 | // expected values for stats on skipped slice |
| 2143 | arrivingSeqs := []uint64{3, 15, 18, 2, 1} |
| 2144 | expSliceLen := []int64{2, 3, 4, 4, 3} |
| 2145 | |
| 2146 | numSeqsInList := dbContext.DbStats.CacheStats.NumCurrentSeqsSkipped.Value() |
| 2147 | for j := 0; j < len(arrivingSeqs); j++ { |
| 2148 | newEntry := &LogEntry{ |
| 2149 | DocID: fmt.Sprintf("doc_%d", arrivingSeqs), |
| 2150 | RevID: "1-abcdefabcdefabcdef", |
| 2151 | Sequence: arrivingSeqs[j], |
nothing calls this directly
no test coverage detected