JobFlow/ТЗ at main · RomanSnitko/JobFlow · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
# JobFlow

JobFlow — это платформа распределенной обработки задач.

Можно просто вгружать сотни тысяч задач одновременно,
запустить несколько серверов воркеров (которые сами по себе работают на корутинах),
а они уже будут из redis разбирать таски и выполнять асинхронно.


Features:

    Если сервер упадет, после перезагрузки он продолжит выполнять задачи (благодаря Postgres).
    Можно запустить 1 сервер API и 10 серверов-воркеров, которые будут разбирать общую очередь.
    Можно запланировать задачу на «через час» (delayed tasks).


## План ТЗ:


Модуль 1: сущности данных

    Task.hpp

    Описать структуру task: UUID, тип (email/video),
    статус (pending/running/done/failed),
    данные (JSON), время запуска.

    binds: Используется всеми модулями.


Модуль 2: слой данных

    PostgresDAO.*

    методы для sql-запросов:

        InsertTask: сохранить новую задачу.
        UpdateStatus: изменить статус на running или done.
        GetReadyTasks: найти задачи, у которых статус pending и время запуска уже наступило.

    binds: Общается напрямую с userver::storages::postgres.


Модуль 3: Task API

    TaskSubmitHandler.*, TaskStatusHandler.*

    what:

        Submit: принять JSON от клиента -> превратить в модель Task ->
        вызвать PostgresDAO::InsertTask.
        Status: принять ID -> найти в базе ->
        вернуть клиенту JSON со статусом.

    binds: Вызывает методы из PostgresDAO.


Модуль 4: работа с очередью в redis

    RedisQueue.*

    what: Написать методы для работы с Redis:

        Push(taskId): положить ID задачи в список.
        Pop(): забрать ID задачи (блокирующая оперция).

    binds: Использует userver::storages::redis.


Модуль 5: планировщик

    TaskScheduler.*

    what: Создать фоновый процесс (userver::utils::PeriodicTask).

    logic: Раз в секунду вызывает PostgresDAO::GetReadyTasks.
    Если нашли ID -> вызываем RedisQueue::Push -> обновляем статус в БД на queued.

    binds: Соединяет Postgres и Redis.


Модуль 6: воркер

    TaskWorker.*

    what: inf цикл в отдельной корутине.

    logic: вызвать RedisQueue::Pop() (взять задачу) -Ю получить ID ->
    UpdateStatus(running) -> выполняет «бизнес-логику»
    (пока просто sleep и вывод в лог (потом можно просто направлять на реальные задачи)) ->
    UpdateStatus(done)

    binds: Самый шлюший актив модуль, использует и Redis, и Postgres.


Путь задачи:

    Ingestion: Пользователь делает POST /v1/tasks/submit. Handler проверяет JSON и говорит PostgresDAO:
    сохранить. В базе появляется строка со статусом pending.

    Detection: Scheduler (каждую секунду обновляется) видит эту строку. Он говорит:
    пора работать суки. Он меняет статус в базе на queued и кидает короткий ID задачи в Redis.

    Dispatch: Worker висит на Redis и ждет. Как только в Redis упал ID, Воркер его подхватывает.

    Execution: Воркер говорит PostgresDAO: Твою мать, волтер, я взял задачу, ставь статус running.
    Затем он делает саму работу.

    Completion: Работа закончена (в последствии можно перенаправлять на реальные задачи,
    время работы остается ~такое же).
    Воркер говорит PostgresDAO: Ставь статус done.

API -> Postgres -> Scheduler -> Redis -> Worker