Forked Workflows call didComplete() multiple times

[jbrophy17]

When forking a Workflow, the didComplete() function is called each time one of the forks completes. I understand that this is happening because each fork needs to call .commit(), which adds a subscription, but intuitively I would expect didComplete() to only be called once per Workflow after all branches had completed.

A basic test case of this:

    func test_fork_verifySingleCompletion() {
        let workflow = TestWorkflow()
        
        let emptyObservable = Observable.just(((), ()))
        let rootStep = workflow
            .onStep { _ -> Observable<((), ())> in
                return emptyObservable
            }
        
        let firstFork: Step<(), (), ()>? = rootStep.asObservable().fork(workflow)
        _ = firstFork?
            .onStep { (_, _) -> Observable<((), ())> in
                return Observable.just(((), ()))
            }
            .commit()
        
        let secondFork: Step<(), (), ()>? = rootStep.asObservable().fork(workflow)
        _ = secondFork?
            .onStep { (_, _) -> Observable<((), ())> in
                return Observable.just(((), ()))
            }
            .commit()
        
        XCTAssertEqual(0, workflow.completeCallCount)
        
        _ = workflow.subscribe(())
        
        XCTAssertEqual(1, workflow.completeCallCount) // XCTAssertEqual failed: ("1") is not equal to ("2")
    }

[edrew08]

Is the problem that you are traveling down both forks? If you're forking a workflow, wouldn't you expect to either travel down the firstFork path or the secondFork path, but not both?

[jbrophy17]

Maybe I am misusing the fork, but my understanding is that both forks are committed, but only one is fully executed. Something like:

    func test_fork_pathing() {
        let workflow = TestWorkflow()
        
        let emptyTrueObservable = Observable.just(((), true))
        let rootStep = workflow
            .onStep { _ -> Observable<((), Bool)> in
                return emptyTrueObservable
        }
        
        let firstFork: Step<(), (), Bool>? = rootStep.asObservable().fork(workflow)
        _ = firstFork?
            .onStep { (_, value) -> Observable<((), ())> in
                if value {
                    return Observable.just(((), ()))
                }
                return Observable.empty()
            }
            .commit()
        
        let secondFork: Step<(), (), Bool>? = rootStep.asObservable().fork(workflow)
        _ = secondFork?
            .onStep { (_, value) -> Observable<((), ())> in
                if !value {
                    return Observable.just(((), ()))
                }
                return Observable.empty()
            }
            .commit()

        XCTAssertEqual(0, workflow.completeCallCount)

        _ = workflow.subscribe(())

        XCTAssertEqual(1, workflow.completeCallCount) //XCTAssertEqual failed: ("1") is not equal to ("2")
        
        
    }

One option is to return an error on the non-followed fork, but it's not really an error that the path isn't executed, it's expected behavior. But it does mean that didComplete is only called once.
Error example:

    func test_fork_pathing() {
        
        enum WorkflowError: Error {
            case deadend
        }
        
        let workflow = TestWorkflow()
        
        let emptyTrueObservable = Observable.just(((), true))
        let rootStep = workflow
            .onStep { _ -> Observable<((), Bool)> in
                return emptyTrueObservable
        }
        
        let firstFork: Step<(), (), Bool>? = rootStep.asObservable().fork(workflow)
        _ = firstFork?
            .onStep { (_, value) -> Observable<((), ())> in
                if value {
                    return Observable.just(((), ()))
                }
                return Observable.error(WorkflowError.deadend)
            }
            .commit()
        
        let secondFork: Step<(), (), Bool>? = rootStep.asObservable().fork(workflow)
        _ = secondFork?
            .onStep { (_, value) -> Observable<((), ())> in
                if !value {
                    return Observable.just(((), ()))
                }
                return Observable.error(WorkflowError.deadend)
            }
            .commit()

        XCTAssertEqual(0, workflow.completeCallCount)

        _ = workflow.subscribe(())

        XCTAssertEqual(1, workflow.completeCallCount) // success
    }

Or am I writing my forked Workflow completely wrong here?

[neakor]

This is a bug in the Rx chain inside Workflow. The fix is pretty simple. I'll put up a PR. Thanks for raising this.