def METHOD_NAME(instance: DagsterInstance, empty_workspace_context):
    assert instance.run_storage.get_cursor_values({FAILURE_KEY, SUCCESS_KEY}) == {}
    daemon = TestEventLogConsumerDaemon()
    list(daemon.run_iteration(empty_workspace_context))
    assert instance.run_storage.get_cursor_values({FAILURE_KEY, SUCCESS_KEY}) == {
        FAILURE_KEY: str(0),
        SUCCESS_KEY: str(0),
    }
    run1 = create_run_for_test(instance, "foo")
    run2 = create_run_for_test(instance, "foo")
    instance.report_run_failed(run1)
    instance.report_run_failed(run2)
    list(daemon.run_iteration(empty_workspace_context))
    assert len(daemon.run_records) == 2
    cursors = instance.run_storage.get_cursor_values({FAILURE_KEY, SUCCESS_KEY})
    list(daemon.run_iteration(empty_workspace_context))
    assert instance.run_storage.get_cursor_values({FAILURE_KEY, SUCCESS_KEY}) == cursors
    for _ in range(5):
        instance.report_engine_event("foo", run1)
        instance.report_engine_event("foo", run2)
    list(daemon.run_iteration(empty_workspace_context))
    assert instance.run_storage.get_cursor_values({FAILURE_KEY, SUCCESS_KEY}) == {
        FAILURE_KEY: str(int(cursors[FAILURE_KEY]) + 10),
        SUCCESS_KEY: str(int(cursors[SUCCESS_KEY]) + 10),
    }
    run3 = create_run_for_test(instance, "foo")
    run4 = create_run_for_test(instance, "foo")
    instance.report_run_failed(run3)
    instance.report_run_failed(run4)
    list(daemon.run_iteration(empty_workspace_context))
    assert len(daemon.run_records) == 2