Tests channel cache backfill with slow query, validates that a request that is terminated while waiting for the view lock doesn't trigger a view query. Runs multiple goroutines, using a channel and two waitgroups to ensure expected ordering of events, as follows 1. Define a PostQueryCallback (via l
(t *testing.T)
| 895 | // - releases the view lock, second changes request is unblocked |
| 896 | // - since it's been terminated, should return error before executing a second view query |
| 897 | func TestChannelQueryCancellation(t *testing.T) { |
| 898 | |
| 899 | if !base.UnitTestUrlIsWalrus() { |
| 900 | t.Skip("Skip test with LeakyBucket dependency test when running in integration") |
| 901 | } |
| 902 | base.SetUpTestLogging(t, base.LevelInfo, base.KeyCache) |
| 903 | |
| 904 | // Set up PostQueryCallback on bucket - will be invoked when changes triggers the cache backfill view query |
| 905 | |
| 906 | // Use queryWg to pause the query |
| 907 | var queryWg sync.WaitGroup |
| 908 | queryWg.Add(1) |
| 909 | |
| 910 | // Use queryBlocked to detect when the first called has reached queryWg.Wait |
| 911 | queryBlocked := make(chan struct{}) |
| 912 | |
| 913 | // Use changesWg to block until test goroutines are both complete |
| 914 | var changesWg sync.WaitGroup |
| 915 | |
| 916 | postQueryCallback := func(ddoc, viewName string, params map[string]interface{}) { |
| 917 | close(queryBlocked) // Notifies that a query is blocked, trigger to initiate second changes request |
| 918 | queryWg.Wait() // Waits until second changes request attempts to make a view query |
| 919 | } |
| 920 | |
| 921 | // Use leaky bucket to inject callback in query invocation |
| 922 | queryCallbackConfig := base.LeakyBucketConfig{ |
| 923 | PostQueryCallback: postQueryCallback, |
| 924 | } |
| 925 | |
| 926 | db, ctx := setupTestLeakyDBWithCacheOptions(t, DefaultCacheOptions(), queryCallbackConfig) |
| 927 | defer db.Close(ctx) |
| 928 | collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db) |
| 929 | collection.ChannelMapper = channels.NewChannelMapper(ctx, channels.DocChannelsSyncFunction, db.Options.JavascriptTimeout) |
| 930 | |
| 931 | // Write a handful of docs/sequences to the bucket |
| 932 | _, _, err := collection.Put(ctx, "key1", Body{"channels": "ABC"}) |
| 933 | assert.NoError(t, err, "Put failed with error: %v", err) |
| 934 | _, _, err = collection.Put(ctx, "key2", Body{"channels": "ABC"}) |
| 935 | assert.NoError(t, err, "Put failed with error: %v", err) |
| 936 | _, _, err = collection.Put(ctx, "key3", Body{"channels": "ABC"}) |
| 937 | assert.NoError(t, err, "Put failed with error: %v", err) |
| 938 | _, _, err = collection.Put(ctx, "key4", Body{"channels": "ABC"}) |
| 939 | assert.NoError(t, err, "Put failed with error: %v", err) |
| 940 | require.NoError(t, db.changeCache.waitForSequence(ctx, 4, base.DefaultWaitForSequence)) |
| 941 | |
| 942 | // Issue two one-shot since=0 changes request. Both will attempt a view query. The first will block based on queryWg, |
| 943 | // the second will block waiting for the view lock |
| 944 | initialQueryCount := db.DbStats.Cache().ViewQueries.Value() |
| 945 | changesWg.Add(1) |
| 946 | go func() { |
| 947 | defer changesWg.Done() |
| 948 | var options ChangesOptions |
| 949 | options.Since = SequenceID{Seq: 0} |
| 950 | options.ChangesCtx = base.TestCtx(t) |
| 951 | options.Continuous = false |
| 952 | options.Wait = false |
| 953 | options.Limit = 2 // Avoid prepending results in cache, as we don't want second changes to serve results from cache |
| 954 | _ = getChanges(t, collection, base.SetOf("ABC"), options) |
nothing calls this directly
no test coverage detected