Repository: broadwaylamb/OpenCombine (swift)

Issue:
Should change unfair lock to unfair recursive lock in Sink?
when AnyCancellable calls cancel function automatically in DEINIT, Sink's cancel function will be called, then terminateAndConsumeLock will be called. As the result, the receiveValue block and receiveCompletion block will be set to nil. Then objects in receiveValue block or receiveCompletion block may be destroyed. If there is an object that in block, it will call same sink's receive completion to finish in object's DEINIT function. That will make a crash cause call lock's lock function twice without unlock.

Pull request description:
Fixes #208.

Patch:
diff --git a/Sources/OpenCombine/Publishers/Publishers.Concatenate.swift b/Sources/OpenCombine/Publishers/Publishers.Concatenate.swift
index f8d1cd29b..4c844a2f2 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Concatenate.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Concatenate.swift
@@ -233,7 +233,7 @@ extension Publishers.Concatenate {
 
         private var suffixState = SubscriptionStatus.awaitingSubscription
 
-        private let suffix: Suffix
+        private var suffix: Suffix?
 
         private var pending = Subscribers.Demand.none
 
@@ -266,8 +266,18 @@ extension Publishers.Concatenate {
                 prefixState.subscription ?? suffixState.subscription
             prefixState = .terminal
             suffixState = .terminal
-            lock.unlock()
-            upstreamSubscription?.cancel()
+
+            // We MUST release the object AFTER unlocking the lock,
+            // since releasing it may trigger execution of arbitrary code,
+            // for example, if the object has a deinit.
+            // When the object deallocates, its deinit is called, and holding
+            // the lock at that moment can lead to deadlocks.
+
+            withExtendedLifetime(suffix) {
+                suffix = nil
+                lock.unlock()
+                upstreamSubscription?.cancel()
+            }
         }
 
         var description: String { return "Concatenate" }
@@ -320,7 +330,7 @@ extension Publishers.Concatenate {
             lock.unlock()
             switch completion {
             case .finished:
-                suffix.subscribe(SuffixSubscriber(inner: self))
+                suffix?.subscribe(SuffixSubscriber(inner: self))
             case .failure:
                 downstream.receive(completion: completion)
             }
diff --git a/Sources/OpenCombine/Subscribers/Subscribers.Assign.swift b/Sources/OpenCombine/Subscribers/Subscribers.Assign.swift
index e2768dfce..5211e91f8 100644
--- a/Sources/OpenCombine/Subscribers/Subscribers.Assign.swift
+++ b/Sources/OpenCombine/Subscribers/Subscribers.Assign.swift
@@ -112,8 +112,17 @@ extension Subscribers {
             lock.assertOwner()
 #endif
             status = .terminal
-            object = nil
-            lock.unlock()
+
+            // We MUST release the object AFTER unlocking the lock,
+            // since releasing it may trigger execution of arbitrary code,
+            // for example, if the object has a deinit.
+            // When the object deallocates, its deinit is called, and holding
+            // the lock at that moment can lead to deadlocks.
+
+            withExtendedLifetime(object) {
+                object = nil
+                lock.unlock()
+            }
         }
     }
 }
diff --git a/Sources/OpenCombine/Subscribers/Subscribers.Sink.swift b/Sources/OpenCombine/Subscribers/Subscribers.Sink.swift
index 3e72dc29a..d2ae500de 100644
--- a/Sources/OpenCombine/Subscribers/Subscribers.Sink.swift
+++ b/Sources/OpenCombine/Subscribers/Subscribers.Sink.swift
@@ -73,8 +73,22 @@ extension Subscribers {
 
         public func receive(completion: Subscribers.Completion<Failure>) {
             lock.lock()
+            status = .terminal
             let receiveCompletion = self.receiveCompletion
-            terminateAndConsumeLock()
+            self.receiveCompletion = { _ in }
+
+            // We MUST release the closures AFTER unlocking the lock,
+            // since releasing a closure may trigger execution of arbitrary code,
+            // for example, if the closure captures an object with a deinit.
+            // When closure deallocates, the object's deinit is called, and holding
+            // the lock at that moment can lead to deadlocks.
+            // See https://github.com/OpenCombine/OpenCombine/issues/208
+
+            withExtendedLifetime(receiveValue) {
+                receiveValue = { _ in }
+                lock.unlock()
+            }
+
             receiveCompletion(completion)
         }
 
@@ -84,18 +98,21 @@ extension Subscribers {
                 lock.unlock()
                 return
             }
-            terminateAndConsumeLock()
-            subscription.cancel()
-        }
-
-        private func terminateAndConsumeLock() {
-#if DEBUG
-            lock.assertOwner()
-#endif
             status = .terminal
-            receiveValue = { _ in }
-            receiveCompletion = { _ in }
-            lock.unlock()
+
+            // We MUST release the closures AFTER unlocking the lock,
+            // since releasing a closure may trigger execution of arbitrary code,
+            // for example, if the closure captures an object with a deinit.
+            // When closure deallocates, the object's deinit is called, and holding
+            // the lock at that moment can lead to deadlocks.
+            // See https://github.com/OpenCombine/OpenCombine/issues/208
+
+            withExtendedLifetime((receiveValue, receiveCompletion)) {
+                receiveCompletion = { _ in }
+                receiveValue = { _ in }
+                lock.unlock()
+            }
+            subscription.cancel()
         }
     }
 }


Test patch:
diff --git a/Tests/OpenCombineTests/Helpers/AutomaticallyFinish.swift b/Tests/OpenCombineTests/Helpers/AutomaticallyFinish.swift
new file mode 100644
index 000000000..21f45d1e4
--- /dev/null
+++ b/Tests/OpenCombineTests/Helpers/AutomaticallyFinish.swift
@@ -0,0 +1,52 @@
+//
+//  AutomaticallyFinish.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 08.07.2021.
+//
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+#endif
+
+final class AutomaticallyFinish<Output, Failure: Error> {
+
+    let subscription: CustomSubscription
+    let publisher: CustomPublisherBase<Output, Failure>
+
+    init() {
+        subscription = .init()
+        publisher = .init(subscription: subscription)
+    }
+
+    deinit {
+        publisher.send(completion: .finished)
+    }
+
+    func notify(_ value: Output) {
+        _ = publisher.send(value)
+    }
+
+    func listen(receiveCompletion: @escaping (Subscribers.Completion<Failure>) -> Void,
+                receiveValue: @escaping (Output) -> Void) -> AnyCancellable {
+        return publisher.sink(receiveCompletion: receiveCompletion,
+                              receiveValue: receiveValue)
+    }
+}
+
+extension AutomaticallyFinish: Publisher {
+    func receive<Downstream: Subscriber>(subscriber: Downstream)
+        where Downstream.Failure == Failure, Downstream.Input == Output
+    {
+        publisher.subscribe(subscriber)
+    }
+}
+
+extension AutomaticallyFinish where Failure == Never {
+    func assign<Root>(to keyPath: ReferenceWritableKeyPath<Root, Output>,
+                      on object: Root) -> AnyCancellable {
+        return publisher.assign(to: keyPath, on: object)
+    }
+}
diff --git a/Tests/OpenCombineTests/Helpers/CustomPublisher.swift b/Tests/OpenCombineTests/Helpers/CustomPublisher.swift
index 7c77043b6..0d3cfe436 100644
--- a/Tests/OpenCombineTests/Helpers/CustomPublisher.swift
+++ b/Tests/OpenCombineTests/Helpers/CustomPublisher.swift
@@ -30,10 +30,8 @@ import OpenCombine
 ///     publisher.subscribe(subscriber)
 ///
 ///     assert(subscription.history == [.requested(.max(42)), .cancelled])
-@available(macOS 10.15, iOS 13.0, *)
 typealias CustomPublisher = CustomPublisherBase<Int, TestingError>
 
-@available(macOS 10.15, iOS 13.0, *)
 class CustomPublisherBase<Output, Failure: Error>: Publisher, Cancellable {
 
     private(set) var subscriber: AnySubscriber<Output, Failure>?
@@ -83,7 +81,6 @@ class CustomPublisherBase<Output, Failure: Error>: Publisher, Cancellable {
     }
 }
 
-@available(macOS 10.15, iOS 13.0, *)
 extension CustomPublisherBase: Equatable {
     static func == (lhs: CustomPublisherBase<Output, Failure>,
                     rhs: CustomPublisherBase<Output, Failure>) -> Bool {
@@ -91,10 +88,8 @@ extension CustomPublisherBase: Equatable {
     }
 }
 
-@available(macOS 10.15, iOS 13.0, *)
 typealias CustomConnectablePublisher = CustomConnectablePublisherBase<Int, TestingError>
 
-@available(macOS 10.15, iOS 13.0, *)
 final class CustomConnectablePublisherBase<Output, Failure: Error>
     : CustomPublisherBase<Output, Failure>,
       ConnectablePublisher
diff --git a/Tests/OpenCombineTests/Helpers/CustomSubscription.swift b/Tests/OpenCombineTests/Helpers/CustomSubscription.swift
index 7afaa05a6..328deb59f 100644
--- a/Tests/OpenCombineTests/Helpers/CustomSubscription.swift
+++ b/Tests/OpenCombineTests/Helpers/CustomSubscription.swift
@@ -16,7 +16,6 @@ import OpenCombine
 ///
 /// In order to inject `CustomSubscription` into the chain of subscriptions,
 /// use the `CustomSubscriber` class.
-@available(macOS 10.15, iOS 13.0, *)
 final class CustomSubscription: Subscription, CustomStringConvertible {
 
     enum Event: Equatable, CustomStringConvertible {
@@ -79,7 +78,6 @@ final class CustomSubscription: Subscription, CustomStringConvertible {
     var description: String { return "CustomSubscription" }
 }
 
-@available(macOS 10.15, iOS 13.0, *)
 extension CustomSubscription: Equatable {
     static func == (lhs: CustomSubscription, rhs: CustomSubscription) -> Bool {
         return lhs === rhs
diff --git a/Tests/OpenCombineTests/PublisherTests/ConcatenateTests.swift b/Tests/OpenCombineTests/PublisherTests/ConcatenateTests.swift
index ef6c081be..cbedfee6c 100644
--- a/Tests/OpenCombineTests/PublisherTests/ConcatenateTests.swift
+++ b/Tests/OpenCombineTests/PublisherTests/ConcatenateTests.swift
@@ -377,6 +377,60 @@ final class ConcatenateTests: XCTestCase {
         XCTAssertEqual(publisher.prepend(CollectionOfOne(42)).prefix.sequence.first, 42)
     }
 
+    func testReleasesSuffixOnCancellation() throws {
+        var suffixIsDestroyed = false
+
+        var downstreamSubscription: Subscription?
+
+        do {
+            let prefix =
+                CustomPublisherBase<Int, Never>(subscription: CustomSubscription())
+            let suffix =
+                CustomPublisherBase<Int, Never>(subscription: CustomSubscription())
+            suffix.onDeinit = {
+                suffixIsDestroyed = true
+            }
+
+            let tracking = TrackingSubscriberBase<Int, Never>(
+                receiveSubscription: {
+                    downstreamSubscription = $0
+                }
+            )
+
+            prefix.append(suffix).subscribe(tracking)
+        }
+
+        XCTAssertFalse(suffixIsDestroyed)
+
+        try XCTUnwrap(downstreamSubscription).cancel()
+
+        XCTAssertTrue(suffixIsDestroyed)
+    }
+
+    func testReceiveCompletionWhileCancelling() throws {
+
+        var downstreamSubscription: Subscription?
+
+        do {
+            let prefix =
+                CustomPublisherBase<Int, Never>(subscription: CustomSubscription())
+            let autofinish = AutomaticallyFinish<Int, Never>()
+
+            let tracking = TrackingSubscriberBase<Int, Never>(
+                receiveSubscription: {
+                    downstreamSubscription = $0
+                }
+            )
+
+            prefix.append(autofinish).subscribe(tracking)
+
+            prefix.send(completion: .finished)
+        }
+
+        // autofinish is deallocated here, a completion is sent
+        try XCTUnwrap(downstreamSubscription).cancel()
+    }
+
     func testAppendReceiveValueBeforeSubscription() {
         testReceiveValueBeforeSubscription(
             value: 12,
diff --git a/Tests/OpenCombineTests/SubscribersTests/AssignTests.swift b/Tests/OpenCombineTests/SubscribersTests/AssignTests.swift
index 4ab78042c..7d2b33426 100644
--- a/Tests/OpenCombineTests/SubscribersTests/AssignTests.swift
+++ b/Tests/OpenCombineTests/SubscribersTests/AssignTests.swift
@@ -137,4 +137,40 @@ final class AssignTests: XCTestCase {
         publisher.send(100)
         XCTAssertEqual(object.value, 42)
     }
+
+    func testReceiveCompletionWhileCancelling() {
+        let cancellable: AnyCancellable
+
+        do {
+            let object = ObjectToModify()
+            cancellable = object.autofinish.assign(to: \.value, on: object)
+        }
+
+        // autofinish is deallocated here, a completion is sent to the sink
+        cancellable.cancel()
+    }
+
+    func testReceiveCompletionWhileCompleting() {
+        let cancellable: AnyCancellable
+
+        let finish: () -> Void
+
+        do {
+            let object = ObjectToModify()
+            cancellable = object.autofinish.assign(to: \.value, on: object)
+
+            let underlyingPublisher = object.autofinish.publisher
+
+            finish = { underlyingPublisher.send(completion: .finished) }
+        }
+
+        finish() // autofinish is deallocated here, a completion is sent to the sink
+
+        cancellable.cancel()
+    }
+}
+
+final class ObjectToModify {
+    let autofinish = AutomaticallyFinish<Int, Never>()
+    var value = 0
 }
diff --git a/Tests/OpenCombineTests/SubscribersTests/SinkTests.swift b/Tests/OpenCombineTests/SubscribersTests/SinkTests.swift
index 88045b779..974de7dde 100644
--- a/Tests/OpenCombineTests/SubscribersTests/SinkTests.swift
+++ b/Tests/OpenCombineTests/SubscribersTests/SinkTests.swift
@@ -210,6 +210,82 @@ final class SinkTests: XCTestCase {
                               releasesClosures: true)
     }
 
+    func testReceiveCompletionWhileCancelling() {
+        // https://github.com/OpenCombine/OpenCombine/issues/208
+
+        let cancellable: AnyCancellable
+
+        do {
+            let autofinish = AutomaticallyFinish<Int, Never>()
+
+            cancellable = autofinish.listen(
+                receiveCompletion: { _ in },
+                receiveValue: { _ in
+                    XCTFail("Should not be called")
+                    _ = autofinish // capture
+                }
+            )
+        }
+
+        // autofinish is deallocated here, a completion is sent to the sink
+        cancellable.cancel()
+    }
+
+    func testReceiveCompletionWhileCancelling2() {
+        // https://github.com/OpenCombine/OpenCombine/issues/208
+
+        let cancellable: AnyCancellable
+
+        do {
+            let autofinish = AutomaticallyFinish<Int, Never>()
+
+            cancellable = autofinish.listen(
+                receiveCompletion: { _ in
+                    XCTFail("Should not be called")
+                    _ = autofinish // capture
+                },
+                receiveValue: { _ in }
+            )
+        }
+
+        // autofinish is deallocated here, a completion is sent to the sink
+        cancellable.cancel()
+    }
+
+    func testReceiveCompletionWhileCompleting() {
+        // https://github.com/OpenCombine/OpenCombine/issues/208
+
+        let cancellable: AnyCancellable
+
+        let finish: () -> Void
+
+        var receiveCompletionCalled = false
+
+        do {
+            let autofinish = AutomaticallyFinish<Int, Never>()
+
+            cancellable = autofinish.listen(
+                receiveCompletion: { _ in
+                    receiveCompletionCalled = true
+                },
+                receiveValue: { _ in
+                    XCTFail("Should not be called")
+                    _ = autofinish // capture
+                }
+            )
+
+            let underlyingPublisher = autofinish.publisher
+
+            finish = { underlyingPublisher.send(completion: .finished) }
+        }
+
+        finish() // autofinish is deallocated here, a completion is sent to the sink
+
+        cancellable.cancel()
+
+        XCTAssertTrue(receiveCompletionCalled)
+    }
+
     func testRecursiveCompletion() {
         var recursionCounter = 10
         var delayedSink: Sut?