Test

Prompt

Напиши на чистому Python (тільки стандартна бібліотека: `asyncio`, `time`, `heapq`, `typing`, `unittest` тощо — без зовнішніх пакетів) високопродуктивний асинхронний розподілений планувальник задач із підтримкою DAG-залежностей (Directed Acyclic Graph), пріоритетів, обмеження паралелізму та механізму відкату (Saga / Compensation pattern). ### Вимоги до реалізації: 1. **Граф задач (DAG) та виявлення циклів:** - Кожна задача `Task` має: `id: str`, `priority: int` (менше число = вищий пріоритет), `dependencies: list[str]`, `timeout: float`, `max_retries: int`, `action: Callable[[dict], Awaitable[Any]]`, `compensate: Optional[Callable[[dict], Awaitable[None]]]`. - Перед запуском планувальник зобов'язаний перевірити граф на наявність циклів або відсутніх залежностей і викинути кастомний виняток `InvalidDAGError` із зазначенням конкретного циклу. 2. **Динамічне виконання та пріоритезація:** - Паралельно може виконуватися не більше `max_concurrency` задач одночасно. - Якщо кілька задач одночасно готові до запуску (всі їхні залежності успішно виконані), першою має стартувати задача з найвищим пріоритетом (`priority`). - Результати виконаних батьківських задач передаються у дочірню задачу у вигляді словника `{parent_id: result}`. 3. **Таймаути, ретраї та атомарний відкат (Saga Rollback):** - Якщо задача перевищує свій `timeout` або кидає виняток, планувальник робить до `max_retries` повторних спроб з експоненційною затримкою (`0.05 * (2 ** attempt)`). - **Критична умова:** Якщо задача остаточно провалилася після всіх спроб, планувальник має: 1) Негайно скасувати всі інші задачі, що зараз виконуються (`asyncio.Task.cancel()`). 2) Виконати функції `compensate` для всіх задач, які вже встигли успішно завершитися, **строго у зворотному топологічному порядку** (спочатку дочірні, потім батьківські), передаючи в `compensate` той самий словник результатів батьків, що й при запуску. 3) Кинути виняток `WorkflowFailedError` з інформацією про першопричину збою та статус відкату. 4. **Вбудований набір тестів (обов'язково в кінці файлу):** Напиши блок `if __name__ == "__main__":`, який автоматично проганяє 4 жорсткі тести і виводить `[PASS]` або `[FAIL]` з таймінгами: - **Тест 1:** Діамантовий DAG (A -> B, C -> D) із перевіркою передачі даних і суворого дотримання `max_concurrency=2`. - **Тест 2:** Перевірка пріоритетів — коли розблоковуються 3 задачі одночасно при `max_concurrency=1`, вони мають виконатися строго в порядку пріоритету, а не в порядку додавання. - **Тест 3:** Виявлення прихованого циклу в графі з 6 вузлів. - **Тест 4 (найважчий):** Сценарій збою на фінальному вузлі після успішного виконання ланцюжка з 4 залежних вузлів — перевірити, що паралельна гілка скасувалася, а `compensate` викликалися у строго зворотному порядку залежностей. ### Правила виводу: - Видай **тільки повний, готовий до запуску код** в одному блоці без скорочень (`...` чи `# TODO` заборонені). - Код повинен запускатися з першого разу без жодних ручних правок.