Zip crash

[albertopeam]

Hi!
We are experiencing a crash after upgrading from 0.11.0(internal fork) to 0.13.0 in Zip operator.
We have created a demo project as a showcase, please take a look. It only contains a test case testExample.
Could you take a look @broadwaylamb?

OpenCombineTest.zip

[Kyle-Ye]

The same code will not pass on Combine(You may want to use delay instead of debounce) and will crash on OpenCombine.
A more simple case in Package form OpenCombineDemo.zip

// Pass with Combine but crash on OpenCombine
import XCTest
import Combine

@available(macOS 10.15, *)
class CombineTestTests: XCTestCase {
    var subscriptions: Set<AnyCancellable> = .init()
    
    func testExample() throws {
        let scheduler = DispatchQueue.main
        let expectation = self.expectation(description: #function)
        var result: (Int, Int)?
        
        let firstPublisher = Just(1)
            .delay(for: .milliseconds(600), scheduler: scheduler)
        let secondPublisher = Just(2)
            .delay(for: .milliseconds(600), scheduler: scheduler)
        Publishers.Zip(firstPublisher, secondPublisher)
            .sink(receiveValue: {
                result = ($0.0, $0.1)
                expectation.fulfill()
            })
            .store(in: &subscriptions)
        
        wait(for: [expectation], timeout: 5)
        
        XCTAssertEqual(result?.0, 1)
        XCTAssertEqual(result?.1, 2)
    }
}

[Kyle-Ye]

The problem is no one is strongly ref the Zip.Inner, so it will dealloc at some point.

// Publishers.Zip.swift
public func receive<Downstream: Subscriber>(subscriber: Downstream) where
    UpstreamB.Failure == Downstream.Failure,
    Downstream.Input == (UpstreamA.Output, UpstreamB.Output)
{
    _ = Inner<Downstream>(downstream: subscriber, a, b)
}

[Kyle-Ye]

Introduced by #222 and the question seems to be discussed at #109 (comment)

That struct holds a strong reference to the Inner. It is also a Subscriber, but not Subscription.

Could you help look at this? @MaxDesiatov