[//]: # ([![Crates.io][crates-badge]][crates-url])
[![MIT licensed][mit-badge]][mit-url]
[![Build Status][actions-badge]][actions-url]

[//]: # ([crates-badge]: https://img.shields.io/crates/v/tokio.svg)

[//]: # ([crates-url]: https://crates.io/crates/tokio)

[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg

[mit-url]: https://github.com/qiaoruntao/mscheduler/blob/master/LICENSE

[actions-badge]: https://github.com/qiaoruntao/mscheduler/actions/workflows/ci.yml/badge.svg

[actions-url]: https://github.com/qiaoruntao/mscheduler/actions?query=branch%3Amaster

## 功能点
1. 任务的发布和执行
2. 多个worker同时运行任务
3. 任务失败后重试
4. worker ping超时后其他worker可以抢占任务
5. 

## 核心模型
Task
1. TaskState
   1. 整体任务状态参数
   2. Vec\<TaskWorkerState\>
2. TaskOption
3. \<T\>参数

TaskConsumer
1. Arc\<Func\> 因为需要spawn到其他线程内执行
2. 

任务中worker state的状态转换
1. INIT 刚发送
2. RUNNING 被worker占领并执行中
3. SUCCESS worker执行成功
4. FAIL worker执行失败

## 功能设计
### 任务发送
不变
### 任务消费

**如何发现任务**

1. 启动时主动查询一批任务
2. 通过change stream 订阅一批发生变化的任务
3. 任务列表空时主动查询一批任务

**查询什么任务**

1. INIT=》需要抢占
2. RUNNING=》等待超时后抢占
3. 其他worker执行失败,但是符合抢占条件=》走复杂判断逻辑

**按什么优先级查询任务**

目前没有特殊要求, 按任意key顺序查询

**如何判断任务是否可抢占**
1. 任务参数限制不可执行
   1. specific_worker_ids限制不可执行
2. 其他worker状态限制
   1. ping_expire_time没到, 不可抢占
   2. 其他worker info占据了并发执行数量
      1. max_unexpected_retries超出限制, 不执行. 约束: max_unexpected_retries\<failed worker info cnt
      2. concurrent_worker_cnt限制并行执行数量. 约束: concurrent_worker_cnt\<running worker cnt &&  concurrent_worker_cnt\<success worker info cnt

**抢占的流程**
1. 启动前使用change stream监听. 启动后查询一批任务, 查询条件: 任务参数允许worker执行 && 任务没有完全成功或失败(根据info cnt数量确定).
2. 查询后读取worker info状态, 用于判断抢占时间点
3. 遍历这些任务, 计算任务可抢占的时间点. 设置定时任务在这些时间点进行抢占. PriorityQueue实现, 有最大长度限制.
4. 如果可抢占任务为空, 那么重新查询一批任务直到没有数据返回. 之后完全可以通过change stream监听

**执行的流程**
1. 通过tokio::spawn隔离报错
2. 按option里的设置持续ping
3. 任务失败/成功后更新任务状态

**任务失败的报错**
没有特殊需求, 保持原样

**模块划分**
1. stream监听. 始终存在
2. 批量读取任务. 读取完成后可退出, 直到stream监听报错为止(存在时间gap, 无法判断是否存在新更新的任务). 所以这个由stream监听线程启动
3. 任务抢占. 监听和批量读取后通过该线程维护PriorityQueue, 并执行任务抢占
4. 任务执行. 任务抢占后spawn任务并执行. 执行过程中维护执行handler的状态. 任务完成后负责更新任务状态

## 测试用例
1. 一个worker, 持续运行, 

| Feature                    |      Bull       | Agenda |
|:---------------------------|:---------------:|:------:|
| Backend                    |      redis      | mongo  |
| Priorities                 |        ✓        |   ✓    |
| Concurrency                |        ✓        |   ✓    |
| Delayed jobs               |        ✓        |   ✓    |
| Global events              |        ✓        |   ✓    |
| Rate Limiter               |        ✓        |        |
| Pause/Resume               |        ✓        |   ✓    |
| Sandboxed worker           |        ✓        |   ✓    |
| Repeatable jobs            |        ✓        |   ✓    |
| Atomic ops                 |        ✓        |   ~    |
| Persistence                |        ✓        |   ✓    |
| UI                         |        ✓        |   ✓    |
| REST API                   |                 |   ✓    |
| Central (Scalable) Queue   |                 |   ✓    |
| Supports long running jobs |                 |   ✓    |
| Optimized for              | Jobs / Messages |  Jobs  |
| Gracefuly stop             |                 |        |
| Remote stop                |                 |        |
| Multiple worker            |                 |        |

## 流程

1. 发布任务
    1. 幂等

## worker状态

1. 执行中
   worker占用任务成功后立即进入执行中状态, 并设置下次刷新时间
   如果距离上次刷新时间超过worker_timeout_ms, 那么认为是超时状态
2. 失败
   worker主动设置执行状态为fail. 如果任务失败时worker不在列表中, 状态设置会失败
3. 成功
   worker主动设置执行状态为success. 如果任务成功时worker不在列表中, 状态设置会失败
4. 超时
   根据worker_timeout_ms推导得到的状态

## 任务状态

任务状态由当前worker列表决定

1. 未启动
   列表为空
2. 执行中
   列表中存在执行中worker
3. 成功
   列表中只有success的worker
4. 失败
   列表中只有fail的worker

## 提供功能

1. 发布任务(幂等)
   查找key对应的没有运行中的任务
   没有找到=>setOnInsert
   找到了=>set nothing, 按配置决定是否清除失败/成功的worker
2. 占用任务(多worker)
   查找条件: 判断是否被指定worker_id, 任务状态不是成功, 任务没有被当前worker处理过(不能在worker列表中), 可以接受更多的worker
   排序条件: priority最高, 优先被指定worker_id的
   没有找到=>不做操作
   找到了=>增加自己的worker对象, 并且过滤worker列表(清除超时的运行中worker)
3. 维持任务
   查找条件: key相同, worker id相同,
   找到了=> 更新对应的超时时间
   没找到=> 结束当前任务
4. 任务执行成功
   尝试更新任务状态为成功
5. 任务执行失败
   主动返回任务失败, 尝试更新任务状态为失败
6. 任务执行异常
   按option中设置重试, 并更新重试次数, 如果重试次数超过限制那么更新为任务失败. 重试过程只在本地发生, 所有重试次数失败后更新为失败状态

## 配置更新后的影响

1. specific_worker_ids变动
   如果更新后不允许当前worker执行, 那么结束任务
2. ping_interval_ms变动
   下次ping时生效
3. 其他参数变动
   不影响正在执行中的任务, 变动需要在重新占用任务时体现

## 具体实现

### 发布任务

直接发送就行

### 消费任务

核心问题: 下一次什么时候去占用任务

1. 启动时计算下一次时间next_try_time
2. 使用change_stream实时更新next_try_time

## TODO

-[ ] clean_success 暂不实现, maintenance
-[ ] clean_failed 暂不实现, maintenance
-[ ] detect compatibility of collection data
-[x] auto worker id
-[ ] 错误处理