async def METHOD_NAME(actor_pool):
    pool, cluster_api, session_id, slots_ref, queue_ref = actor_pool
    speculation_conf = {
        "enabled": True,
        "interval": 1000,
        "threshold": 0.2,
        "min_task_runtime": 0.01,
        "multiplier": 1.5,
        "max_concurrent_run": 2,
    }
    speculative_scheduler = SpeculativeScheduler(queue_ref, slots_ref, speculation_conf)
    await speculative_scheduler.start()
    await speculative_scheduler._speculative_execution()
    total_subtasks = 5
    subtasks = [
        Subtask(str(i), retryable=False, logic_key=f"logic_key1", logic_parallelism=5)
        for i in range(total_subtasks)
    ]
    subtask_infos = [
        SubtaskScheduleInfo(subtask, max_reschedules=3) for subtask in subtasks
    ]
    # add unfinished subtasks
    for subtask_info in subtask_infos:
        speculative_scheduler.add_subtask(subtask_info)
    await speculative_scheduler._speculative_execution()
    assert len(speculative_scheduler._grouped_finished_subtasks.values()) == 0
    # finished some subtasks
    for subtask_info in subtask_infos[:-1]:
        speculative_scheduler.finish_subtask(subtask_info)
    assert (
        len(next(iter(speculative_scheduler._grouped_finished_subtasks.values())))
        == total_subtasks - 1
    )
    assert (
        len(next(iter(speculative_scheduler._grouped_unfinished_subtasks.values())))
        == 1
    )
    await speculative_scheduler._speculative_execution()
    subtask_infos[-1].subtask.retryable = True
    # pretend subtask has been running on a band.
    subtask_infos[-1].band_futures[("addr0", "numa-0")] = asyncio.ensure_future(
        asyncio.sleep(1)
    )
    await speculative_scheduler._speculative_execution()
    submitted = await queue_ref.get_subtasks()
    # assert stale subtasks resubmitted
    assert subtask_infos[-1].subtask in submitted
    await speculative_scheduler._speculative_execution()
    # if resubmitted subtasks not running, don't resubmitted again.
    assert 1 == len(await queue_ref.get_subtasks())
    # pretend subtask has been running on a band.
    subtask_infos[-1].band_futures[("addr1", "numa-0")] = asyncio.ensure_future(
        asyncio.sleep(1)
    )
    await speculative_scheduler._speculative_execution()
    # stale subtasks resubmitted again
    assert 2 == len(await queue_ref.get_subtasks())
    # pretend subtask has been running on another band.
    subtask_infos[-1].band_futures[("addr2", "numa-0")] = asyncio.ensure_future(
        asyncio.sleep(1)
    )
    # speculative run reached max limit `max_concurrent_run`, i.e. 2
    await speculative_scheduler._speculative_execution()
    # assert raise queue_ref raise NoAvailableBand
    speculative_scheduler._subtask_speculation_max_concurrent_run += 1
    await speculative_scheduler._speculative_execution()
    assert isinstance((await queue_ref.get_exceptions())[0], NoAvailableBand)
    # finish subtasks
    speculative_scheduler.finish_subtask(subtask_infos[-1])
    assert len(speculative_scheduler._grouped_unfinished_subtasks) == 0
    await speculative_scheduler.stop()