operators.retry(n) interferes with non-erroring repeats

[matiboy]

Describe the bug
When using e.g. operators.retry(3) the operator will stop the emission after 3 repeats even though no error had been raised.

To Reproduce

I feel that the below test should pass, but it fails as mentioned in the comments

def test_retry_with_count_when_complete():
    scheduler = TestScheduler()
    xs = scheduler.create_cold_observable(
        on_next(90, 42), on_completed(200)
    )

    result = scheduler.start(
        lambda: xs.pipe(
            operators.retry(2), # <-- expecting this to only react when an error is thrown
            operators.repeat() # <-- when it reaches this, retry has done its job and resub-ing to source via repeat should reset the counter
        )
    )
    assert result.messages == [
        on_next(290, 42),
        on_next(490, 42),
        on_next(690, 42), # <-- expecting this and next line, but in current state, ...
        on_next(890, 42), # <-- ... these two will never be emitted as retry(2) stops source from repeating
    ]

Expected behavior
retry should not interact with the sequence unless an error is emitted. Currently it limits the number of calls to n regardless of the source observable behavior.

I understand that it may be my understanding of retry(n) which is flawed, but it seems to me like a bug. Logically retry(3) should look at the on_completed as (quoting docs):

or until it successfully terminates

So when src completes, retry(3) has done its job, and it's only when we resubscribe via repeat() that it should reset to try 3 times imo.

Code or Screenshots

See reproduction test above

Additional context

• RxPY version: 4.0.4
• Python version : 3.11