CSHARP-2092: Cursor iteration should complete (abnormally) when another thread closes the cursor.

Author: DmitryLukyanov

src/MongoDB.Driver.Core/Core/Misc/Feature.cs

@@ -76,6 +76,7 @@ public class Feature
         private static readonly Feature __scramSha256Authentication = new Feature("ScramSha256Authentication", new SemanticVersion(4, 0, 0, ""));
         private static readonly Feature __serverExtractsUsernameFromX509Certificate = new Feature("ServerExtractsUsernameFromX509Certificate", new SemanticVersion(3, 3, 12));
         private static readonly Feature __shardedTransactions = new Feature("ShardedTransactions", new SemanticVersion(4, 1, 6));
+        private static readonly Feature __tailableCursor = new Feature("TailableCursor", new SemanticVersion(3, 2, 0));
         private static readonly Feature __transactions = new Feature("Transactions", new SemanticVersion(4, 0, 0));
         private static readonly Feature __userManagementCommands = new Feature("UserManagementCommands", new SemanticVersion(2, 6, 0));
         private static readonly Feature __views = new Feature("Views", new SemanticVersion(3, 3, 11));
@@ -347,6 +348,11 @@ public class Feature
         /// </summary>
         public static Feature ShardedTransactions => __shardedTransactions;
 
+        /// <summary>
+        /// Gets the tailable cursor feature.
+        /// </summary>
+        public static Feature TailableCursor => __tailableCursor;
+
         /// <summary>
         /// Gets the transactions feature.
         /// </summary>

src/MongoDB.Driver.Core/Core/Operations/AsyncCursor.cs

@@ -555,6 +555,7 @@ private async Task KillCursorsAsync(CancellationToken cancellationToken)
         public bool MoveNext(CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
+            cancellationToken.ThrowIfCancellationRequested();
 
             bool hasMore;
             if (TryMoveNext(out hasMore))
@@ -571,6 +572,7 @@ public bool MoveNext(CancellationToken cancellationToken)
         public async Task<bool> MoveNextAsync(CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
+            cancellationToken.ThrowIfCancellationRequested();
 
             bool hasMore;
             if (TryMoveNext(out hasMore))

tests/MongoDB.Driver.Tests/AsyncCursorTests.cs

@@ -13,7 +13,9 @@
 * limitations under the License.
 */
 
+using System;
 using System.Collections.Generic;
+using System.Threading;
 using FluentAssertions;
 using FluentAssertions.Common;
 using MongoDB.Bson;
@@ -78,17 +80,69 @@ public void KillCursor_should_actually_work()
                 cursor = collection.FindSync("{}");
                 cursor.MoveNext();
 
-                var cursorId = ((AsyncCursor<BsonDocument>) cursor)._cursorId();
+                var cursorId = ((AsyncCursor<BsonDocument>)cursor)._cursorId();
                 cursorId.Should().NotBe(0);
                 cursor.Dispose();
 
                 var desiredResult = BsonDocument.Parse($"{{ \"cursorsKilled\" : [{cursorId}], \"cursorsNotFound\" : [], " +
                     $"\"cursorsAlive\" : [], \"cursorsUnknown\" : [], \"ok\" : 1.0 }}");
-                var result = ((CommandSucceededEvent) eventCapturer.Events[0]).Reply;
+                var result = ((CommandSucceededEvent)eventCapturer.Events[0]).Reply;
                 result.IsSameOrEqualTo(desiredResult);
             }
         }
 
+        [SkippableFact]
+        public void Tailable_cursor_should_be_able_to_be_cancelled_from_a_different_thread_with_expected_result()
+        {
+            RequireServer.Check().Supports(Feature.TailableCursor);
+
+            string testCollectionName = "test";
+            string testDatabaseName = "test";
+
+            var client = DriverTestConfiguration.Client;
+            var database = client.GetDatabase(testDatabaseName);
+            var collection = database.GetCollection<BsonDocument>(testCollectionName);
+
+            DropCollection(client, testDatabaseName, testCollectionName);
+            var createCollectionOptions = new CreateCollectionOptions()
+            {
+                Capped = true,
+                MaxSize = 1000
+            };
+            database.CreateCollection(testCollectionName, createCollectionOptions);
+            collection.InsertOne(new BsonDocument()); // tailable cursors don't work on an empty collection
+
+            using (var cancellationTokenSource = new CancellationTokenSource())
+            {
+                var findOptions = new FindOptions<BsonDocument>()
+                {
+                    BatchSize = 1,
+                    CursorType = CursorType.TailableAwait
+                };
+                var cursor = collection.FindSync(FilterDefinition<BsonDocument>.Empty, findOptions);
+                var enumerator = cursor.ToEnumerable(cancellationTokenSource.Token).GetEnumerator();
+
+                var semaphore = new SemaphoreSlim(0);
+                var thread = new Thread(() =>
+                {
+                    semaphore.Wait();
+                    cancellationTokenSource.Cancel();
+                });
+                thread.Start();
+
+                var exception = Record.Exception((Action)(() =>
+                {
+                    while (true)
+                    {
+                        _ = enumerator.MoveNext();
+                        semaphore.Release(1);
+                    }
+                }));
+
+                exception.Should().BeAssignableTo<OperationCanceledException>();
+            }
+        }
+
         //private methods
         private IMongoClient CreateClient()
         {
@@ -105,6 +159,6 @@ private void DropCollection(IMongoClient client, string databaseName, string col
     public static class AsyncCursorReflector
     {
         public static long _cursorId(this AsyncCursor<BsonDocument> obj) =>
-            (long) Reflector.GetFieldValue(obj, nameof(_cursorId));
+            (long)Reflector.GetFieldValue(obj, nameof(_cursorId));
     }
 }