rx.delay frequently destroys tasks about aioreactive HOT 1 CLOSED

@curry_flipped(1)
def sleep(
    source: AsyncObservable[_TSource],
    seconds: float,
    sleep_on_start: bool = False,
) -> AsyncObservable[_TSource]:
    started = sleep_on_start

    @curry_flipped(1)
    async def _sleep(source, seconds: float):
        nonlocal started
        if started:
            await asyncio.sleep(seconds)
        else:
            started = True
        return source

    return pipe(
        source,
        rx.map_async(_sleep(seconds)),
    )

def setup(token: str):
    session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=None))

    return pipe(
        rx.interval(0, 2),
        rx.flat_map(
            lambda _: pipe(
                range(14),
                rx.from_iterable,
                rx_helpers.sleep(0.1),
                rx.map(lambda y: arrow.now().shift(days=y)),
                rx.map_async(
                    schedules(token, session),
                ),
            )
        ),
        rx.subscribe_async(rx.AsyncAnonymousObserver()),
    )