feat: escalate overdue approval tasks via processDue

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
0264408
2026-09-16 09:23:30 +08:00
co-authored by Cursor
parent 850583f326
commit 30047f3612
32 changed files with 736 additions and 42 deletions
+3 -2
View File
@@ -10,7 +10,7 @@
|---|---|
| `ordo-api` | 公共模型与 `OrdoEngine` 端口 |
| `ordo-core` | 运行时(`DefaultOrdoEngine` / `InMemoryOrdoEngine`) |
| `ordo-storage-jdbc` | JDBC 存储 + Flyway 迁移(V1–V5) |
| `ordo-storage-jdbc` | JDBC 存储 + Flyway 迁移(V1–V6) |
| `ordo-spring-boot-starter` | Spring Boot 自动装配(JDBC + Flyway) |
| `ordo-example` | 内存引擎示例 |
@@ -23,11 +23,12 @@
- ACTION 步骤:事务提交后调用宿主 `ActionHandler`
- 发起人撤回:`WITHDRAWN`,待办任务 `SKIPPED`
- 任务转派:当前办理人 `reassign`,审计 `TASK_REASSIGNED`
- 到期升级:步骤 `due` + `processDue`,审计 `TASK_ESCALATED`
- 分页查询:任务 / 实例 / 流程定义
- 审计时间线:`ProcessEvent` + `queryHistory`
- 扩展点:`AssigneeResolver`、`RoutingCondition`、`ActionHandler`、`OrdoEventListener`
开发计划:到期升级、`CANCELLED`、定义不可变多版本、MySQL 方言、可选 REST + 目录 SPI。设计器为独立产品(不进本仓库),待 REST、目录与多版本定义之后。多租户 **暂不在计划中**。见 [docs/roadmap.md](docs/roadmap.md)。
开发计划:`CANCELLED`、定义不可变多版本、MySQL 方言、可选 REST + 目录 SPI。设计器为独立产品(不进本仓库),待 REST、目录与多版本定义之后。多租户 **暂不在计划中**。见 [docs/roadmap.md](docs/roadmap.md)。
详细用法(定义 JSON、扩展点、异常、查询、ACTION/审计语义)见 **[docs/usage.md](docs/usage.md)**。对外行为变更时同步更新该文档。
+2 -5
View File
@@ -7,22 +7,19 @@
## 已完成
- ANY/ALL 会签/或签
- JDBC 存储(PostgreSQL)+ Flyway V1–V5
- JDBC 存储(PostgreSQL)+ Flyway V1–V6
- 条件路由 `StepTransition` + `RoutingCondition`
- ACTION + `ActionHandler`;执行记录持久化
- 发起人撤回 `WITHDRAWN`
- 分页查询:任务 / 实例 / 定义
- 审计 `ProcessEvent` / `queryHistory`;`OrdoEventListener`
- 任务转派 `reassign` / `TASK_REASSIGNED`
- 到期升级 `processDue` / `TASK_ESCALATED`
## 开发计划(确定要做)
下列能力已纳入计划,尚未实现。实现顺序可按依赖调整,但范围本身不从计划中拿掉。
### 到期升级
目前无到期时间与定时器。计划支持 SLA/超时:到期后升级(改办理人、通知或进入指定步骤),即「到期升级」。
### CANCELLED
目前终态只有 `APPROVED` / `REJECTED` / `WITHDRAWN`(撤回仅发起人)。计划增加管理员/系统取消语义 `CANCELLED`,与撤回区分。
+11 -6
View File
@@ -12,7 +12,7 @@ Ordo 是嵌入宿主进程的审批引擎,入口是 `OrdoEngine`。
产品边界:不做业务表单、用户体系、多租户;不内置设计器 UI。业务字段放在 `ProcessContext`(不可变 `Map<String, Object>`)。当前也**没有** REST;HTTP 仍由宿主自建。计划中的可选 REST 与独立设计器见 [roadmap.md](roadmap.md)。
开发计划(尚未提供,见 [roadmap.md](roadmap.md)):到期升级、`CANCELLED`、定义不可变多版本、MySQL 方言、可选 REST + 目录 SPI。
开发计划(尚未提供,见 [roadmap.md](roadmap.md)):`CANCELLED`、定义不可变多版本、MySQL 方言、可选 REST + 目录 SPI。
## 2. 模块与接入
@@ -62,6 +62,8 @@ ordo:
enabled: true
definitions:
location: classpath*:ordo/*.json # 启动时对每个 JSON 调用 replace
due:
poll-ms: 0 # >0 时轮询 processDue;默认不调度
```
启动加载使用 `replace`:无 `RUNNING` 实例则整图替换;有运行中实例则保留库里的定义。
@@ -169,6 +171,7 @@ new ProcessDefinition("leave-request-routed", "Leave request",
| `kind` | `APPROVAL`(默认)或 `ACTION`。 |
| `candidates` / `policy` | 仅审批步。`policy` 默认 `ANY`。审批步至少一名候选人,禁止重复。 |
| `action` | 仅 ACTION 步,对应 `ActionHandler.execute` 的 key。审批步禁止带 `action`。 |
| `due` | 仅审批步。可选。`after` 为 ISO-8601 时长;`then` 为 `reassign` / `notify` / `goto`。进入该步时任务 `dueAt = now + after`。 |
| `from` / `to` / `when` / `priority` | `to` 为 `null` 或 JSON `null` 表示结束。`when` 空则无条件。`priority` 默认 0,**越小越先匹配**。 |
## 4. 实例与任务
@@ -201,7 +204,8 @@ ordo.withdraw(instance.id(), "alice", "计划有变");
### 4.2 权限
- `approve` / `reject` / `reassign`:`actor` 必须等于该任务当前 `assignee`,否则 `UnauthorizedTaskOperationException`。
- `reassign`:仅 `PENDING` 任务;同一任务 id,办理人改为 `newAssignee`,不推进步骤。`newAssignee` 不可空白、不可等于当前 `assignee`,且同一步不能已有该人的 `PENDING` 任务,否则 `IllegalArgumentException`。不经过 `AssigneeResolver`。
- `reassign`:仅 `PENDING` 任务;同一任务 id,办理人改为 `newAssignee`,不推进步骤。`newAssignee` 不可空白、不可等于当前 `assignee`,且同一步不能已有该人的 `PENDING` 任务,否则 `IllegalArgumentException`。不经过 `AssigneeResolver`。人工转派不改 `dueAt`。
- `processDue(limit)`:认领 `dueAt <= now` 的 PENDING 任务(`limit > 0`),按步上 `due.then` 执行:`reassign` 换办理人(`to` 走 `AssigneeResolver`)、`notify` 可选 `ActionHandler`、`goto` 跳过当前步 PENDING 并进入 `to` 步骤(`to` 必须是步骤 id)。每种策略对一张任务最多成功一次(清空 `dueAt`)。引擎无后台线程;Spring 下 `ordo.due.poll-ms > 0` 才轮询。
- 任务非 `PENDING`:`TaskAlreadyCompletedException`。
- `withdraw`:仅 `initiator`,否则 `UnauthorizedInstanceOperationException`;实例非 `RUNNING`:`InstanceAlreadyCompletedException`。
@@ -275,10 +279,10 @@ public class MailActions implements ActionHandler {
`ProcessEventType`:
- `INSTANCE_STARTED` / `INSTANCE_APPROVED` / `INSTANCE_REJECTED` / `INSTANCE_WITHDRAWN`
- `TASK_CREATED` / `TASK_APPROVED` / `TASK_REJECTED` / `TASK_SKIPPED` / `TASK_REASSIGNED`
- `TASK_CREATED` / `TASK_APPROVED` / `TASK_REJECTED` / `TASK_SKIPPED` / `TASK_REASSIGNED` / `TASK_ESCALATED`
- `ACTION_SUCCEEDED` / `ACTION_FAILED`
字段:`id`、`instanceId`、可选 `taskId`/`stepId`/`actor`/`detail`、`occurredAt`。系统完成类事件 `actor` 可为空。`TASK_REASSIGNED` 的 `actor` 为转出人,`detail` 为转入人。
字段:`id`、`instanceId`、可选 `taskId`/`stepId`/`actor`/`detail`、`occurredAt`。系统完成类事件 `actor` 可为空。`TASK_REASSIGNED` 的 `actor` 为转出人,`detail` 为转入人。`TASK_ESCALATED` 的 `actor` 为空,`detail` 为新办理人 / notify 的 action key / goto 目标步 id。
`OrdoEventListener.onEvent(ProcessEvent)` 在**事务提交之后**按事件顺序调用(含本轮 ACTION 结果)。单个 listener 抛错只打日志,不影响流程和其他 listener。
@@ -312,6 +316,7 @@ ACTION 成功事件发生在提交之后,因此排在同轮事务内写入的
| `start` | 发起;可选 `ProcessContext` |
| `approve` / `reject` | 办理当前 PENDING 任务 |
| `reassign` | 当前办理人把 PENDING 任务转给他人 |
| `processDue` | 认领并处理已到期 PENDING 任务 |
| `withdraw` | 发起人撤回 |
| `find*` | 按 id / 待办索引读取 |
| `queryTasks` / `queryInstances` / `queryDefinitions` | 分页列表 |
@@ -338,12 +343,12 @@ ACTION 成功事件发生在提交之后,因此排在同轮事务内写入的
## 12. 存储
Flyway 脚本在 `ordo-storage-jdbc` 的 `db/migration`(V1–V5)。表包括定义/步骤/候选人/转移、实例、任务、`ordo_process_event`、`ordo_action_execution`。
Flyway 脚本在 `ordo-storage-jdbc` 的 `db/migration`(V1–V6)。表包括定义/步骤/候选人/转移、实例、任务、`ordo_process_event`、`ordo_action_execution`。
多 JVM 共享同一库时,多步写入走 `TransactionExecutor`,完成任务/实例用条件更新(仍 PENDING / 仍 RUNNING 才改),避免双花。
## 13. 未提供能力
开发计划中(见 [roadmap.md](roadmap.md)):到期升级、`CANCELLED`、定义不可变多版本、MySQL 方言、可选 REST + 目录 SPI。独立设计器不进本仓库,等 REST、目录与多版本定义之后再做。
开发计划中(见 [roadmap.md](roadmap.md)):`CANCELLED`、定义不可变多版本、MySQL 方言、可选 REST + 目录 SPI。独立设计器不进本仓库,等 REST、目录与多版本定义之后再做。
暂不在计划中:多租户、设计器 UI。当前 REST 由宿主自建。
@@ -10,7 +10,7 @@ import java.util.Set;
* action steps have an {@code actionKey} invoked by the host {@link ActionHandler}.
*/
public record ApprovalStep(String id, String name, List<String> candidates, ApprovalPolicy policy, StepKind kind,
String actionKey) {
String actionKey, StepDue due) {
public ApprovalStep {
requireText(id, "step id");
requireText(name, "step name");
@@ -24,6 +24,9 @@ public record ApprovalStep(String id, String name, List<String> candidates, Appr
}
requireText(actionKey, "action key");
actionKey = actionKey.strip();
if (due != null) {
throw new IllegalArgumentException("an action step must not have due");
}
} else {
if (actionKey != null && !actionKey.isBlank()) {
throw new IllegalArgumentException("an approval step must not have an action key");
@@ -43,7 +46,12 @@ public record ApprovalStep(String id, String name, List<String> candidates, Appr
}
public ApprovalStep(String id, String name, List<String> candidates, ApprovalPolicy policy) {
this(id, name, candidates, policy, StepKind.APPROVAL, null);
this(id, name, candidates, policy, StepKind.APPROVAL, null, null);
}
public ApprovalStep(String id, String name, List<String> candidates, ApprovalPolicy policy, StepKind kind,
String actionKey) {
this(id, name, candidates, policy, kind, actionKey, null);
}
/** Convenience factory for the common case of a single, fixed approver. */
@@ -3,5 +3,6 @@ package com.jetlumen.ordo.api;
import java.time.Instant;
public record ApprovalTask(String id, String instanceId, String stepId, String name, String assignee,
TaskStatus status, Instant createdAt, Instant completedAt, TaskAction action) {
TaskStatus status, Instant createdAt, Instant completedAt, TaskAction action,
Instant dueAt) {
}
@@ -0,0 +1,8 @@
package com.jetlumen.ordo.api;
/** What happens when an approval step's due time is reached. */
public enum DueThen {
REASSIGN,
NOTIFY,
GOTO
}
@@ -30,6 +30,10 @@ public interface OrdoEngine {
}
ApprovalTask reject(String taskId, String actor, String comment);
ApprovalTask reassign(String taskId, String actor, String newAssignee);
/** Claims and processes up to {@code limit} overdue pending tasks. {@code limit} must be positive. */
int processDue(int limit);
default ProcessInstance withdraw(String instanceId, String actor) {
return withdraw(instanceId, actor, null);
}
@@ -43,6 +43,9 @@ public record ProcessDefinition(String id, String name, List<ApprovalStep> steps
if (!stepsWithOutgoing.contains(step.id())) {
throw new IllegalArgumentException("step " + step.id() + " has no outgoing transition");
}
if (step.due() != null && step.due().then() == DueThen.GOTO && !ids.contains(step.due().to())) {
throw new IllegalArgumentException("unknown due toStepId: " + step.due().to());
}
}
transitions = transitions.stream()
.sorted(Comparator.comparing(StepTransition::fromStepId).thenComparingInt(StepTransition::priority))
@@ -6,6 +6,8 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.io.InputStream;
import java.time.DateTimeException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
@@ -47,7 +49,8 @@ public final class ProcessDefinitionParser {
ApprovalPolicy policy = step.policy() == null ? ApprovalPolicy.ANY : step.policy();
StepKind kind = step.kind() == null ? StepKind.APPROVAL : step.kind();
List<String> candidates = step.candidates() == null ? List.of() : step.candidates();
steps.add(new ApprovalStep(step.id(), step.name(), candidates, policy, kind, step.action()));
steps.add(new ApprovalStep(step.id(), step.name(), candidates, policy, kind, step.action(),
parseDue(step.due())));
}
rotateStartStep(steps, document.startStep());
List<StepTransition> transitions = new ArrayList<>();
@@ -84,6 +87,28 @@ public final class ProcessDefinitionParser {
steps.add(0, start);
}
private static StepDue parseDue(DueDocument due) {
if (due == null) {
return null;
}
if (due.after() == null || due.after().isBlank() || due.then() == null || due.then().isBlank()) {
throw new IllegalArgumentException("due after and then must not be blank");
}
Duration after;
try {
after = Duration.parse(due.after().strip());
} catch (DateTimeException e) {
throw new IllegalArgumentException("invalid due after: " + due.after(), e);
}
DueThen then;
try {
then = DueThen.valueOf(due.then().strip().toUpperCase());
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("invalid due then: " + due.then(), e);
}
return new StepDue(after, then, due.to(), due.action());
}
private record DefinitionDocument(
String id,
String name,
@@ -98,7 +123,15 @@ public final class ProcessDefinitionParser {
List<String> candidates,
ApprovalPolicy policy,
StepKind kind,
@JsonProperty("action") String action) {
@JsonProperty("action") String action,
DueDocument due) {
}
private record DueDocument(
String after,
String then,
String to,
String action) {
}
private record TransitionDocument(
@@ -8,6 +8,7 @@ public enum ProcessEventType {
TASK_REJECTED,
TASK_SKIPPED,
TASK_REASSIGNED,
TASK_ESCALATED,
INSTANCE_APPROVED,
INSTANCE_REJECTED,
INSTANCE_WITHDRAWN,
@@ -0,0 +1,56 @@
package com.jetlumen.ordo.api;
import java.time.Duration;
import java.util.Objects;
/** Relative SLA on an approval step. */
public record StepDue(Duration after, DueThen then, String to, String action) {
public StepDue {
Objects.requireNonNull(after, "after must not be null");
if (after.isZero() || after.isNegative()) {
throw new IllegalArgumentException("due after must be positive");
}
Objects.requireNonNull(then, "due then must not be null");
to = blankToNull(to);
action = blankToNull(action);
switch (then) {
case REASSIGN -> {
if (to == null) {
throw new IllegalArgumentException("reassign due must have to");
}
if (action != null) {
throw new IllegalArgumentException("reassign due must not have action");
}
}
case NOTIFY -> {
if (to != null) {
throw new IllegalArgumentException("notify due must not have to");
}
}
case GOTO -> {
if (to == null) {
throw new IllegalArgumentException("goto due must have to");
}
if (action != null) {
throw new IllegalArgumentException("goto due must not have action");
}
}
}
}
public static StepDue reassign(Duration after, String to) {
return new StepDue(after, DueThen.REASSIGN, to, null);
}
public static StepDue notify(Duration after, String action) {
return new StepDue(after, DueThen.NOTIFY, null, action);
}
public static StepDue gotoStep(Duration after, String stepId) {
return new StepDue(after, DueThen.GOTO, stepId, null);
}
private static String blankToNull(String value) {
return value == null || value.isBlank() ? null : value.strip();
}
}
@@ -5,6 +5,7 @@ import com.jetlumen.ordo.api.query.Page;
import com.jetlumen.ordo.api.query.PageRequest;
import com.jetlumen.ordo.api.query.TaskQuery;
import java.time.Instant;
import java.util.List;
import java.util.Optional;
@@ -40,6 +41,10 @@ public interface ApprovalTaskRepository {
*/
boolean reassignIfPending(String taskId, String expectedAssignee, String newAssignee);
List<ApprovalTask> findDuePending(Instant now, int limit);
boolean claimIfDue(String taskId, String expectedAssignee, Instant now);
/** Paginated, filterable query; results are ordered newest-first (created_at desc). */
Page<ApprovalTask> query(TaskQuery query, PageRequest pageRequest);
}
@@ -2,6 +2,7 @@ package com.jetlumen.ordo.api;
import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -174,4 +175,55 @@ class ProcessDefinitionParserTest {
assertEquals(StepKind.APPROVAL, definition.steps().get(0).kind());
assertEquals(ApprovalStep.action("notify", "Notify HR", "leave-approved-mail"), definition.steps().get(1));
}
@Test
void parsesDueOnApprovalSteps() {
String json = """
{
"id": "leave",
"name": "Leave request",
"steps": [
{
"id": "manager",
"name": "Manager approval",
"candidates": ["maria"],
"due": { "after": "PT48H", "then": "reassign", "to": "director" }
}
],
"transitions": [
{ "from": "manager", "to": null }
]
}
""";
ProcessDefinition definition = ProcessDefinitionParser.fromJson(json);
assertEquals(StepDue.reassign(java.time.Duration.parse("PT48H"), "director"), definition.steps().get(0).due());
}
@Test
void rejectsActionStepDueAndInvalidDueThen() {
String actionDue = """
{
"id": "leave",
"name": "Leave request",
"steps": [
{ "id": "mail", "name": "Mail", "kind": "ACTION", "action": "go",
"due": { "after": "PT1H", "then": "notify" } }
],
"transitions": [ { "from": "mail", "to": null } ]
}
""";
assertThrows(IllegalArgumentException.class, () -> ProcessDefinitionParser.fromJson(actionDue));
String badThen = """
{
"id": "leave",
"name": "Leave request",
"steps": [
{ "id": "manager", "name": "Manager", "candidates": ["maria"],
"due": { "after": "PT1H", "then": "explode" } }
],
"transitions": [ { "from": "manager", "to": null } ]
}
""";
assertThrows(IllegalArgumentException.class, () -> ProcessDefinitionParser.fromJson(badThen));
}
}
@@ -118,4 +118,19 @@ class ProcessDefinitionTest {
assertEquals(StepKind.ACTION, definition.steps().get(0).kind());
assertEquals("leave-approved-mail", definition.steps().get(0).actionKey());
}
@Test
void rejectsActionStepsWithDue() {
assertThrows(IllegalArgumentException.class,
() -> new ApprovalStep("mail", "Mail", List.of(), ApprovalPolicy.ANY, StepKind.ACTION, "notify",
StepDue.notify(java.time.Duration.ofHours(1), "overdue")));
}
@Test
void rejectsGotoDueToAnUnknownStep() {
assertThrows(IllegalArgumentException.class, () -> new ProcessDefinition("leave", "Leave request", List.of(
new ApprovalStep("manager", "Manager approval", List.of("maria"), ApprovalPolicy.ANY, StepKind.APPROVAL,
null, StepDue.gotoStep(java.time.Duration.ofHours(1), "missing"))),
List.of(StepTransition.end("manager"))));
}
}
@@ -16,6 +16,7 @@ import com.jetlumen.ordo.api.ProcessEventType;
import com.jetlumen.ordo.api.ProcessInstance;
import com.jetlumen.ordo.api.ProcessStatus;
import com.jetlumen.ordo.api.RoutingCondition;
import com.jetlumen.ordo.api.StepDue;
import com.jetlumen.ordo.api.StepKind;
import com.jetlumen.ordo.api.StepTransition;
import com.jetlumen.ordo.api.TaskAction;
@@ -195,7 +196,7 @@ public final class DefaultOrdoEngine implements OrdoEngine {
}
Instant now = clock.instant();
ApprovalTask updated = new ApprovalTask(task.id(), task.instanceId(), task.stepId(), task.name(),
newAssignee, task.status(), task.createdAt(), task.completedAt(), task.action());
newAssignee, task.status(), task.createdAt(), task.completedAt(), task.action(), task.dueAt());
record(events, task.instanceId(), task.id(), task.stepId(), ProcessEventType.TASK_REASSIGNED, actor,
newAssignee, now);
return updated;
@@ -228,7 +229,8 @@ public final class DefaultOrdoEngine implements OrdoEngine {
TaskAction action = new TaskAction(actor, comment, now);
for (ApprovalTask pending : taskRepository.findPendingByInstanceId(instanceId)) {
ApprovalTask skipped = new ApprovalTask(pending.id(), pending.instanceId(), pending.stepId(),
pending.name(), pending.assignee(), TaskStatus.SKIPPED, pending.createdAt(), now, action);
pending.name(), pending.assignee(), TaskStatus.SKIPPED, pending.createdAt(), now, action,
pending.dueAt());
if (taskRepository.completeIfPending(skipped)) {
record(events, instanceId, pending.id(), pending.stepId(), ProcessEventType.TASK_SKIPPED, actor,
comment, now);
@@ -301,12 +303,106 @@ public final class DefaultOrdoEngine implements OrdoEngine {
return actionExecutionRepository.query(instanceId, pageRequest);
}
@Override
public synchronized int processDue(int limit) {
if (limit <= 0) {
throw new IllegalArgumentException("limit must be positive");
}
List<PendingAction> queued = new ArrayList<>();
List<ProcessEvent> events = new ArrayList<>();
Integer processed = transactionExecutor.execute(() -> {
Instant now = clock.instant();
int count = 0;
for (ApprovalTask overdue : taskRepository.findDuePending(now, limit)) {
if (escalateDueTask(overdue, now, queued, events)) {
count++;
}
}
return count;
});
finishCommittedWork(queued, events);
return processed;
}
private boolean escalateDueTask(ApprovalTask overdue, Instant now, List<PendingAction> queued,
List<ProcessEvent> events) {
if (!taskRepository.claimIfDue(overdue.id(), overdue.assignee(), now)) {
return false;
}
ProcessInstance instance = requireInstance(overdue.instanceId());
if (instance.status() != ProcessStatus.RUNNING) {
return false;
}
ProcessDefinition definition = requireDefinition(instance.definitionId());
ApprovalStep step = requireStep(definition, overdue.stepId());
StepDue due = step.due();
if (due == null) {
record(events, overdue.instanceId(), overdue.id(), overdue.stepId(), ProcessEventType.TASK_ESCALATED, null,
null, now);
return true;
}
return switch (due.then()) {
case REASSIGN -> escalateReassign(overdue, instance, step, due, now, events);
case NOTIFY -> escalateNotify(overdue, instance, due, now, queued, events);
case GOTO -> escalateGoto(overdue, instance, definition, due, now, queued, events);
};
}
private boolean escalateReassign(ApprovalTask overdue, ProcessInstance instance, ApprovalStep step, StepDue due,
Instant now, List<ProcessEvent> events) {
String newAssignee = assigneeResolver.resolve(due.to(), step, instance.context());
requireText(newAssignee, "resolved assignee");
boolean duplicatePending = taskRepository.findByInstanceIdAndStepId(overdue.instanceId(), overdue.stepId())
.stream()
.anyMatch(other -> !other.id().equals(overdue.id())
&& other.status() == TaskStatus.PENDING
&& other.assignee().equals(newAssignee));
if (duplicatePending || overdue.assignee().equals(newAssignee)) {
LOG.log(Level.WARNING, "due reassign skipped for task: " + overdue.id());
record(events, overdue.instanceId(), overdue.id(), overdue.stepId(), ProcessEventType.TASK_ESCALATED, null,
newAssignee, now);
return true;
}
if (!taskRepository.reassignIfPending(overdue.id(), overdue.assignee(), newAssignee)) {
return false;
}
record(events, overdue.instanceId(), overdue.id(), overdue.stepId(), ProcessEventType.TASK_ESCALATED, null,
newAssignee, now);
return true;
}
private boolean escalateNotify(ApprovalTask overdue, ProcessInstance instance, StepDue due, Instant now,
List<PendingAction> queued, List<ProcessEvent> events) {
if (due.action() != null) {
String executionId = nextId();
actionExecutionRepository.insert(new ActionExecution(executionId, instance.id(), overdue.stepId(),
due.action(), ActionExecutionStatus.PENDING, null,
now.plusMillis(eventSequence.getAndIncrement()), null));
queued.add(new PendingAction(executionId, due.action(), instance.id(), overdue.stepId(),
instance.context()));
}
record(events, overdue.instanceId(), overdue.id(), overdue.stepId(), ProcessEventType.TASK_ESCALATED, null,
due.action(), now);
return true;
}
private boolean escalateGoto(ApprovalTask overdue, ProcessInstance instance, ProcessDefinition definition,
StepDue due, Instant now, List<PendingAction> queued, List<ProcessEvent> events) {
List<ApprovalTask> siblings = taskRepository.findByInstanceIdAndStepId(instance.id(), overdue.stepId());
skipPendingSiblings(siblings, "", null, now, events);
enterStep(instance, definition, requireStep(definition, due.to()), now, queued, events);
record(events, overdue.instanceId(), overdue.id(), overdue.stepId(), ProcessEventType.TASK_ESCALATED, null,
due.to(), now);
return true;
}
private void createStepTasks(ProcessInstance instance, ApprovalStep step, Instant now, List<ProcessEvent> events) {
for (String candidate : step.candidates()) {
String assignee = assigneeResolver.resolve(candidate, step, instance.context());
requireText(assignee, "resolved assignee");
Instant dueAt = step.due() == null ? null : now.plus(step.due().after());
ApprovalTask task = new ApprovalTask(nextId(), instance.id(), step.id(), step.name(), assignee,
TaskStatus.PENDING, now, null, null);
TaskStatus.PENDING, now, null, null, dueAt);
taskRepository.save(task);
record(events, instance.id(), task.id(), step.id(), ProcessEventType.TASK_CREATED, assignee, null, now);
}
@@ -328,7 +424,7 @@ public final class DefaultOrdoEngine implements OrdoEngine {
private ApprovalTask completeTask(ApprovalTask task, TaskStatus status, TaskAction action,
List<ProcessEvent> events) {
ApprovalTask completed = new ApprovalTask(task.id(), task.instanceId(), task.stepId(), task.name(),
task.assignee(), status, task.createdAt(), action.operatedAt(), action);
task.assignee(), status, task.createdAt(), action.operatedAt(), action, task.dueAt());
if (!taskRepository.completeIfPending(completed)) {
throw new TaskAlreadyCompletedException(task.id());
}
@@ -480,7 +576,8 @@ public final class DefaultOrdoEngine implements OrdoEngine {
continue;
}
ApprovalTask skipped = new ApprovalTask(sibling.id(), sibling.instanceId(), sibling.stepId(),
sibling.name(), sibling.assignee(), TaskStatus.SKIPPED, sibling.createdAt(), now, null);
sibling.name(), sibling.assignee(), TaskStatus.SKIPPED, sibling.createdAt(), now, null,
sibling.dueAt());
// Best-effort: if another concurrent decision already completed this sibling, leave it as-is.
if (taskRepository.completeIfPending(skipped)) {
record(events, sibling.instanceId(), sibling.id(), sibling.stepId(), ProcessEventType.TASK_SKIPPED,
@@ -159,4 +159,9 @@ public final class InMemoryOrdoEngine implements OrdoEngine {
public Page<ActionExecution> queryActionExecutions(String instanceId, PageRequest pageRequest) {
return delegate.queryActionExecutions(instanceId, pageRequest);
}
@Override
public int processDue(int limit) {
return delegate.processDue(limit);
}
}
@@ -9,10 +9,12 @@ import com.jetlumen.ordo.api.query.TaskQuery;
import com.jetlumen.ordo.api.repository.ApprovalTaskRepository;
import com.jetlumen.ordo.api.repository.ProcessInstanceRepository;
import java.time.Instant;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
/** Development-only in-memory implementation of the task storage port. */
@@ -92,7 +94,37 @@ public final class InMemoryApprovalTaskRepository implements ApprovalTaskReposit
return false;
}
tasks.put(taskId, new ApprovalTask(current.id(), current.instanceId(), current.stepId(), current.name(),
newAssignee, current.status(), current.createdAt(), current.completedAt(), current.action()));
newAssignee, current.status(), current.createdAt(), current.completedAt(), current.action(),
current.dueAt()));
return true;
}
@Override
public synchronized List<ApprovalTask> findDuePending(Instant now, int limit) {
Objects.requireNonNull(now, "now must not be null");
if (limit <= 0) {
throw new IllegalArgumentException("limit must be positive");
}
return tasks.values().stream()
.filter(task -> task.status() == TaskStatus.PENDING)
.filter(task -> task.dueAt() != null && !task.dueAt().isAfter(now))
.sorted(Comparator.comparing(ApprovalTask::dueAt).thenComparing(ApprovalTask::id))
.limit(limit)
.toList();
}
@Override
public synchronized boolean claimIfDue(String taskId, String expectedAssignee, Instant now) {
ApprovalTask current = tasks.get(taskId);
if (current == null || current.status() != TaskStatus.PENDING
|| !current.assignee().equals(expectedAssignee)
|| current.dueAt() == null
|| current.dueAt().isAfter(now)) {
return false;
}
tasks.put(taskId, new ApprovalTask(current.id(), current.instanceId(), current.stepId(), current.name(),
current.assignee(), current.status(), current.createdAt(), current.completedAt(), current.action(),
null));
return true;
}
@@ -13,6 +13,8 @@ import com.jetlumen.ordo.api.ProcessEventType;
import com.jetlumen.ordo.api.ProcessInstance;
import com.jetlumen.ordo.api.ProcessStatus;
import com.jetlumen.ordo.api.RoutingCondition;
import com.jetlumen.ordo.api.StepDue;
import com.jetlumen.ordo.api.StepKind;
import com.jetlumen.ordo.api.StepTransition;
import com.jetlumen.ordo.api.TaskStatus;
import com.jetlumen.ordo.api.exception.DefinitionAlreadyExistsException;
@@ -33,6 +35,9 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
@@ -677,4 +682,103 @@ class InMemoryOrdoEngineTest {
assertTrue(types.contains(ProcessEventType.ACTION_FAILED));
assertTrue(types.contains(ProcessEventType.INSTANCE_APPROVED));
}
@Test
void processDueReassignsAfterTheStepDueElapses() {
MutableClock clock = new MutableClock(Instant.parse("2026-01-15T09:00:00Z"));
InMemoryOrdoEngine dueEngine = new InMemoryOrdoEngine(clock);
dueEngine.register(ProcessDefinition.linear("leave-due", "Leave request", List.of(
new ApprovalStep("manager", "Manager approval", List.of("maria"), ApprovalPolicy.ANY, StepKind.APPROVAL,
null, StepDue.reassign(java.time.Duration.ofHours(1), "diana")))));
var instance = dueEngine.start("leave-due", "alice");
ApprovalTask task = dueEngine.findPendingTasksByInstanceId(instance.id()).get(0);
assertEquals("maria", task.assignee());
assertEquals(0, dueEngine.processDue(10));
clock.set(Instant.parse("2026-01-15T10:00:00Z"));
assertEquals(1, dueEngine.processDue(10));
ApprovalTask escalated = dueEngine.findTask(task.id()).orElseThrow();
assertEquals("diana", escalated.assignee());
assertNull(escalated.dueAt());
assertEquals(0, dueEngine.processDue(10));
dueEngine.approve(task.id(), "diana");
assertEquals(ProcessStatus.APPROVED, dueEngine.findInstance(instance.id()).orElseThrow().status());
assertTrue(dueEngine.queryHistory(instance.id(), new PageRequest(0, 20)).content().stream()
.anyMatch(event -> event.type() == ProcessEventType.TASK_ESCALATED && "diana".equals(event.detail())));
}
@Test
void processDueNotifiesWithoutChangingAssignee() {
MutableClock clock = new MutableClock(Instant.parse("2026-01-15T09:00:00Z"));
List<String> actions = new java.util.ArrayList<>();
InMemoryOrdoEngine dueEngine = new InMemoryOrdoEngine(clock, AssigneeResolver.direct(),
RoutingCondition.always(), (key, context) -> actions.add(key));
dueEngine.register(ProcessDefinition.linear("leave-notify", "Leave request", List.of(
new ApprovalStep("manager", "Manager approval", List.of("maria"), ApprovalPolicy.ANY, StepKind.APPROVAL,
null, StepDue.notify(java.time.Duration.ofMinutes(30), "overdue-mail")))));
var instance = dueEngine.start("leave-notify", "alice");
clock.set(Instant.parse("2026-01-15T09:30:00Z"));
assertEquals(1, dueEngine.processDue(10));
assertEquals("maria", dueEngine.findPendingTasksByInstanceId(instance.id()).get(0).assignee());
assertEquals(List.of("overdue-mail"), actions);
assertEquals(TaskStatus.PENDING, dueEngine.findPendingTasksByInstanceId(instance.id()).get(0).status());
}
@Test
void processDueGotoSkipsTheStepAndEntersTheTarget() {
MutableClock clock = new MutableClock(Instant.parse("2026-01-15T09:00:00Z"));
InMemoryOrdoEngine dueEngine = new InMemoryOrdoEngine(clock);
dueEngine.register(new ProcessDefinition("leave-goto", "Leave request", List.of(
new ApprovalStep("manager", "Manager approval", List.of("maria"), ApprovalPolicy.ANY, StepKind.APPROVAL,
null, StepDue.gotoStep(java.time.Duration.ofHours(1), "hr")),
ApprovalStep.single("hr", "HR approval", "henry")),
List.of(
StepTransition.always("manager", "hr"),
StepTransition.end("hr"))));
var instance = dueEngine.start("leave-goto", "alice");
clock.set(Instant.parse("2026-01-15T10:00:00Z"));
assertEquals(1, dueEngine.processDue(10));
assertEquals("hr", dueEngine.findPendingTasksByInstanceId(instance.id()).get(0).stepId());
assertEquals("henry", dueEngine.findPendingTasksByInstanceId(instance.id()).get(0).assignee());
assertEquals(TaskStatus.SKIPPED, dueEngine.findTasks(instance.id()).stream()
.filter(task -> task.stepId().equals("manager"))
.findFirst()
.orElseThrow()
.status());
}
@Test
void processDueRejectsNonPositiveLimit() {
assertThrows(IllegalArgumentException.class, () -> engine.processDue(0));
}
private static final class MutableClock extends Clock {
private Instant instant;
private MutableClock(Instant instant) {
this.instant = instant;
}
void set(Instant instant) {
this.instant = instant;
}
@Override
public ZoneId getZone() {
return ZoneOffset.UTC;
}
@Override
public Clock withZone(ZoneId zone) {
throw new UnsupportedOperationException();
}
@Override
public Instant instant() {
return instant;
}
}
}
@@ -87,13 +87,28 @@ class InMemoryApprovalTaskRepositoryTest {
assertEquals("diana", repository.findById("task-1").orElseThrow().assignee());
ApprovalTask completed = new ApprovalTask(pending.id(), pending.instanceId(), pending.stepId(), pending.name(),
"diana", TaskStatus.APPROVED, pending.createdAt(), CREATED_AT.plusSeconds(1), null);
"diana", TaskStatus.APPROVED, pending.createdAt(), CREATED_AT.plusSeconds(1), null, null);
assertTrue(repository.completeIfPending(completed));
assertFalse(repository.reassignIfPending("task-1", "diana", "henry"));
assertFalse(repository.reassignIfPending("missing", "maria", "diana"));
}
@Test
void claimIfDueClearsDueAtOnlyForMatchingPendingTasks() {
InMemoryApprovalTaskRepository repository = new InMemoryApprovalTaskRepository();
Instant dueAt = CREATED_AT.plusSeconds(60);
ApprovalTask pending = new ApprovalTask("task-1", "inst-1", "step", "Step", "maria", TaskStatus.PENDING,
CREATED_AT, null, null, dueAt);
repository.save(pending);
assertTrue(repository.findDuePending(dueAt, 10).contains(pending));
assertTrue(repository.claimIfDue("task-1", "maria", dueAt));
assertEquals(null, repository.findById("task-1").orElseThrow().dueAt());
assertTrue(repository.findDuePending(dueAt, 10).isEmpty());
assertFalse(repository.claimIfDue("task-1", "maria", dueAt));
}
private static ApprovalTask task(String id, String instanceId, String assignee, TaskStatus status, Instant createdAt) {
return new ApprovalTask(id, instanceId, "step", "Step", assignee, status, createdAt, null, null);
return new ApprovalTask(id, instanceId, "step", "Step", assignee, status, createdAt, null, null, null);
}
}
@@ -0,0 +1,68 @@
package com.jetlumen.ordo.spring;
import com.jetlumen.ordo.api.OrdoEngine;
import org.springframework.context.SmartLifecycle;
import java.lang.System.Logger;
import java.lang.System.Logger.Level;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/** Optional poller that calls {@link OrdoEngine#processDue(int)} when {@code ordo.due.poll-ms} is positive. */
public final class OrdoDuePoller implements SmartLifecycle {
private static final Logger LOG = System.getLogger("ordo");
private static final int BATCH_SIZE = 100;
private final OrdoEngine ordoEngine;
private final long pollMs;
private ScheduledExecutorService executor;
private volatile boolean running;
public OrdoDuePoller(OrdoEngine ordoEngine, long pollMs) {
this.ordoEngine = Objects.requireNonNull(ordoEngine, "ordoEngine must not be null");
this.pollMs = pollMs;
}
@Override
public boolean isAutoStartup() {
return pollMs > 0;
}
@Override
public void start() {
if (pollMs <= 0 || running) {
return;
}
executor = Executors.newSingleThreadScheduledExecutor(runnable -> {
Thread thread = new Thread(runnable, "ordo-due-poll");
thread.setDaemon(true);
return thread;
});
executor.scheduleWithFixedDelay(this::tick, pollMs, pollMs, TimeUnit.MILLISECONDS);
running = true;
}
private void tick() {
try {
ordoEngine.processDue(BATCH_SIZE);
} catch (RuntimeException e) {
LOG.log(Level.WARNING, "processDue failed", e);
}
}
@Override
public void stop() {
running = false;
if (executor != null) {
executor.shutdownNow();
executor = null;
}
}
@Override
public boolean isRunning() {
return running;
}
}
@@ -146,4 +146,10 @@ public class OrdoJdbcAutoConfiguration {
public OrdoDefinitionLoader ordoDefinitionLoader(OrdoEngine ordoEngine, OrdoProperties ordoProperties) {
return new OrdoDefinitionLoader(ordoEngine, ordoProperties.getDefinitions().getLocation());
}
@Bean
@ConditionalOnMissingBean
public OrdoDuePoller ordoDuePoller(OrdoEngine ordoEngine, OrdoProperties ordoProperties) {
return new OrdoDuePoller(ordoEngine, ordoProperties.getDue().getPollMs());
}
}
@@ -10,6 +10,7 @@ public class OrdoProperties {
private boolean enabled = true;
private final Definitions definitions = new Definitions();
private final Due due = new Due();
public boolean isEnabled() {
return enabled;
@@ -23,6 +24,10 @@ public class OrdoProperties {
return definitions;
}
public Due getDue() {
return due;
}
public static class Definitions {
private String location = "classpath*:ordo/*.json";
@@ -34,4 +39,17 @@ public class OrdoProperties {
this.location = location;
}
}
public static class Due {
/** Poll interval in milliseconds. {@code 0} disables scheduling. */
private long pollMs;
public long getPollMs() {
return pollMs;
}
public void setPollMs(long pollMs) {
this.pollMs = pollMs;
}
}
}
@@ -13,6 +13,7 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
@@ -21,10 +22,10 @@ import java.util.Optional;
/** JDBC implementation of the task storage port and its pending-task indexes. */
public final class JdbcApprovalTaskRepository implements ApprovalTaskRepository {
private static final String TASK_COLUMNS =
"id, instance_id, step_id, task_name, assignee, status, created_at, completed_at, action_actor, action_comment, action_at";
"id, instance_id, step_id, task_name, assignee, status, created_at, completed_at, action_actor, action_comment, action_at, due_at";
private static final String INSERT_TASK =
"INSERT INTO ordo_approval_task (id, instance_id, step_id, task_name, assignee, status, created_at)"
+ " VALUES (?, ?, ?, ?, ?, ?, ?)";
"INSERT INTO ordo_approval_task (id, instance_id, step_id, task_name, assignee, status, created_at, due_at)"
+ " VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
private static final String COMPLETE_IF_PENDING =
"UPDATE ordo_approval_task SET status = ?, completed_at = ?, action_actor = ?, action_comment = ?, action_at = ?"
+ " WHERE id = ? AND status = 'PENDING' AND assignee = ?";
@@ -43,9 +44,15 @@ public final class JdbcApprovalTaskRepository implements ApprovalTaskRepository
private static final String SELECT_PENDING_BY_INSTANCE =
"SELECT " + TASK_COLUMNS + " FROM ordo_approval_task WHERE status = 'PENDING' AND instance_id = ?"
+ " ORDER BY created_at, id";
private static final String CLAIM_IF_DUE =
"UPDATE ordo_approval_task SET due_at = NULL WHERE id = ? AND status = 'PENDING' AND assignee = ?"
+ " AND due_at IS NOT NULL AND due_at <= ?";
private static final String SELECT_DUE_PENDING =
"SELECT " + TASK_COLUMNS + " FROM ordo_approval_task WHERE status = 'PENDING' AND due_at IS NOT NULL"
+ " AND due_at <= ? ORDER BY due_at, id LIMIT ?";
private static final String TASK_COLUMNS_QUALIFIED =
"t.id, t.instance_id, t.step_id, t.task_name, t.assignee, t.status, t.created_at, t.completed_at,"
+ " t.action_actor, t.action_comment, t.action_at";
+ " t.action_actor, t.action_comment, t.action_at, t.due_at";
private final JdbcConnectionProvider connectionProvider;
@@ -139,6 +146,48 @@ public final class JdbcApprovalTaskRepository implements ApprovalTaskRepository
}
}
@Override
public List<ApprovalTask> findDuePending(Instant now, int limit) {
Objects.requireNonNull(now, "now must not be null");
if (limit <= 0) {
throw new IllegalArgumentException("limit must be positive");
}
Connection connection = connectionProvider.getConnection();
try (PreparedStatement select = connection.prepareStatement(SELECT_DUE_PENDING)) {
select.setTimestamp(1, Timestamp.from(now));
select.setInt(2, limit);
try (ResultSet resultSet = select.executeQuery()) {
List<ApprovalTask> tasks = new ArrayList<>();
while (resultSet.next()) {
tasks.add(ApprovalTaskMapper.read(resultSet));
}
return tasks;
}
} catch (SQLException e) {
throw new JdbcStorageException("failed to query due tasks", e);
} finally {
connectionProvider.close(connection);
}
}
@Override
public boolean claimIfDue(String taskId, String expectedAssignee, Instant now) {
Objects.requireNonNull(taskId, "taskId must not be null");
Objects.requireNonNull(expectedAssignee, "expectedAssignee must not be null");
Objects.requireNonNull(now, "now must not be null");
Connection connection = connectionProvider.getConnection();
try (PreparedStatement update = connection.prepareStatement(CLAIM_IF_DUE)) {
update.setString(1, taskId);
update.setString(2, expectedAssignee);
update.setTimestamp(3, Timestamp.from(now));
return update.executeUpdate() == 1;
} catch (SQLException e) {
throw new JdbcStorageException("failed to claim due task: " + taskId, e);
} finally {
connectionProvider.close(connection);
}
}
@Override
public Page<ApprovalTask> query(TaskQuery query, PageRequest pageRequest) {
Objects.requireNonNull(query, "query must not be null");
@@ -27,8 +27,8 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
private static final String INSERT_DEFINITION =
"INSERT INTO ordo_process_definition (id, name) VALUES (?, ?)";
private static final String INSERT_STEP =
"INSERT INTO ordo_approval_step (definition_id, step_id, step_name, policy, step_order, kind, action_key)"
+ " VALUES (?, ?, ?, ?, ?, ?, ?)";
"INSERT INTO ordo_approval_step (definition_id, step_id, step_name, policy, step_order, kind, action_key,"
+ " due_after, due_then, due_to, due_action) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String INSERT_CANDIDATE =
"INSERT INTO ordo_step_candidate (definition_id, step_id, candidate, candidate_order) VALUES (?, ?, ?, ?)";
private static final String INSERT_TRANSITION =
@@ -44,7 +44,8 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
private static final String SELECT_DEFINITION =
"SELECT id, name FROM ordo_process_definition WHERE id = ?";
private static final String SELECT_STEPS =
"SELECT step_id, step_name, policy, kind, action_key FROM ordo_approval_step"
"SELECT step_id, step_name, policy, kind, action_key, due_after, due_then, due_to, due_action"
+ " FROM ordo_approval_step"
+ " WHERE definition_id = ? ORDER BY step_order";
private static final String SELECT_CANDIDATES =
"SELECT step_id, candidate FROM ordo_step_candidate WHERE definition_id = ? ORDER BY step_id, candidate_order";
@@ -176,7 +177,7 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
for (StepRow row : stepRows) {
List<String> candidates = candidatesByStep.getOrDefault(row.stepId(), List.of());
steps.add(new ApprovalStep(row.stepId(), row.stepName(), candidates, row.policy(), row.kind(),
row.actionKey()));
row.actionKey(), ApprovalStepMapper.toDue(row)));
}
List<StepTransition> transitions = new ArrayList<>();
try (PreparedStatement selectTransitions = connection.prepareStatement(SELECT_TRANSITIONS)) {
@@ -249,6 +250,17 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
insertStep.setInt(5, stepOrder++);
insertStep.setString(6, step.kind().name());
insertStep.setString(7, step.actionKey());
if (step.due() == null) {
insertStep.setString(8, null);
insertStep.setString(9, null);
insertStep.setString(10, null);
insertStep.setString(11, null);
} else {
insertStep.setString(8, step.due().after().toString());
insertStep.setString(9, step.due().then().name());
insertStep.setString(10, step.due().to());
insertStep.setString(11, step.due().action());
}
insertStep.executeUpdate();
}
int candidateOrder = 0;
@@ -1,10 +1,13 @@
package com.jetlumen.ordo.storage.jdbc.mapper;
import com.jetlumen.ordo.api.ApprovalPolicy;
import com.jetlumen.ordo.api.DueThen;
import com.jetlumen.ordo.api.StepDue;
import com.jetlumen.ordo.api.StepKind;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Duration;
/**
* Maps rows of {@code ordo_approval_step}. A step's candidate list lives in the separate
@@ -19,9 +22,22 @@ public final class ApprovalStepMapper {
return new StepRow(resultSet.getString("step_id"), resultSet.getString("step_name"),
ApprovalPolicy.valueOf(resultSet.getString("policy")),
StepKind.valueOf(resultSet.getString("kind")),
resultSet.getString("action_key"));
resultSet.getString("action_key"),
resultSet.getString("due_after"),
resultSet.getString("due_then"),
resultSet.getString("due_to"),
resultSet.getString("due_action"));
}
public record StepRow(String stepId, String stepName, ApprovalPolicy policy, StepKind kind, String actionKey) {
public static StepDue toDue(StepRow row) {
if (row.dueAfter() == null || row.dueAfter().isBlank()) {
return null;
}
return new StepDue(Duration.parse(row.dueAfter()), DueThen.valueOf(row.dueThen()), row.dueTo(),
row.dueAction());
}
public record StepRow(String stepId, String stepName, ApprovalPolicy policy, StepKind kind, String actionKey,
String dueAfter, String dueThen, String dueTo, String dueAction) {
}
}
@@ -23,6 +23,11 @@ public final class ApprovalTaskMapper {
statement.setString(5, task.assignee());
statement.setString(6, task.status().name());
statement.setTimestamp(7, Timestamp.from(task.createdAt()));
if (task.dueAt() == null) {
statement.setNull(8, Types.TIMESTAMP);
} else {
statement.setTimestamp(8, Timestamp.from(task.dueAt()));
}
}
public static void bindComplete(PreparedStatement statement, ApprovalTask completedTask) throws SQLException {
@@ -49,6 +54,7 @@ public final class ApprovalTaskMapper {
TaskAction action = actionActor == null ? null
: new TaskAction(actionActor, resultSet.getString("action_comment"), actionAt.toInstant());
Timestamp completedAt = resultSet.getTimestamp("completed_at");
Timestamp dueAt = resultSet.getTimestamp("due_at");
return new ApprovalTask(
resultSet.getString("id"),
resultSet.getString("instance_id"),
@@ -58,6 +64,7 @@ public final class ApprovalTaskMapper {
TaskStatus.valueOf(resultSet.getString("status")),
resultSet.getTimestamp("created_at").toInstant(),
completedAt == null ? null : completedAt.toInstant(),
action);
action,
dueAt == null ? null : dueAt.toInstant());
}
}
@@ -0,0 +1,8 @@
-- SLA / due escalation: task due_at plus step due policy columns.
ALTER TABLE ordo_approval_task ADD COLUMN due_at TIMESTAMP;
CREATE INDEX idx_approval_task_status_due ON ordo_approval_task (status, due_at);
ALTER TABLE ordo_approval_step ADD COLUMN due_after VARCHAR(32);
ALTER TABLE ordo_approval_step ADD COLUMN due_then VARCHAR(16);
ALTER TABLE ordo_approval_step ADD COLUMN due_to VARCHAR(255);
ALTER TABLE ordo_approval_step ADD COLUMN due_action VARCHAR(255);
@@ -23,6 +23,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
class JdbcApprovalTaskRepositoryTest {
@@ -218,6 +219,22 @@ class JdbcApprovalTaskRepositoryTest {
}
}
@Test
void claimIfDueClearsDueAtAndIgnoresTasksThatAreNotDue() {
Instant dueAt = CREATED_AT.plusSeconds(60);
ApprovalTask pending = new ApprovalTask("task-due", "inst-1", "manager", "manager approval", "maria",
TaskStatus.PENDING, CREATED_AT, null, null, dueAt);
repository.save(pending);
assertTrue(repository.findDuePending(CREATED_AT.plusSeconds(30), 10).isEmpty());
List<ApprovalTask> due = repository.findDuePending(dueAt, 10);
assertEquals(1, due.size());
assertEquals("task-due", due.get(0).id());
assertTrue(repository.claimIfDue("task-due", "maria", dueAt));
assertNull(repository.findById("task-due").orElseThrow().dueAt());
assertFalse(repository.claimIfDue("task-due", "maria", dueAt));
}
private void attempt(CountDownLatch start, CountDownLatch done, AtomicInteger wins, ApprovalTask completed) {
try {
start.await();
@@ -234,12 +251,12 @@ class JdbcApprovalTaskRepositoryTest {
private static ApprovalTask pendingTask(String id, String instanceId, String stepId, String assignee,
Instant createdAt) {
return new ApprovalTask(id, instanceId, stepId, stepId + " approval", assignee,
TaskStatus.PENDING, createdAt, null, null);
TaskStatus.PENDING, createdAt, null, null, null);
}
private static ApprovalTask completedTask(ApprovalTask pending, String comment) {
return new ApprovalTask(pending.id(), pending.instanceId(), pending.stepId(), pending.name(),
pending.assignee(), TaskStatus.APPROVED, pending.createdAt(), COMPLETED_AT,
new TaskAction(pending.assignee(), comment, COMPLETED_AT));
new TaskAction(pending.assignee(), comment, COMPLETED_AT), pending.dueAt());
}
}
@@ -2,6 +2,7 @@ package com.jetlumen.ordo.storage.jdbc;
import com.jetlumen.ordo.api.ActionExecutionStatus;
import com.jetlumen.ordo.api.ActionHandler;
import com.jetlumen.ordo.api.ApprovalPolicy;
import com.jetlumen.ordo.api.ApprovalStep;
import com.jetlumen.ordo.api.ApprovalTask;
import com.jetlumen.ordo.api.AssigneeResolver;
@@ -14,6 +15,8 @@ import com.jetlumen.ordo.api.ProcessInstance;
import com.jetlumen.ordo.api.ProcessStatus;
import com.jetlumen.ordo.api.query.PageRequest;
import com.jetlumen.ordo.api.RoutingCondition;
import com.jetlumen.ordo.api.StepKind;
import com.jetlumen.ordo.api.StepDue;
import com.jetlumen.ordo.api.StepTransition;
import com.jetlumen.ordo.api.TaskStatus;
import com.jetlumen.ordo.api.exception.DefinitionInUseException;
@@ -118,6 +121,42 @@ class JdbcOrdoEngineIntegrationTest {
assertEquals(task.id(), reassignedEvent.taskId());
}
@Test
void processDueReassignsAnOverdueTask() {
ProcessDefinition dueDefinition = ProcessDefinition.linear("leave-due", "Leave request", List.of(
new ApprovalStep("manager", "Manager approval", List.of("maria"), ApprovalPolicy.ANY,
StepKind.APPROVAL, null,
StepDue.reassign(java.time.Duration.ofHours(1), "diana"))));
OrdoEngine startEngine = new DefaultOrdoEngine(Clock.fixed(NOW, ZoneOffset.UTC), AssigneeResolver.direct(),
RoutingCondition.always(), ActionHandler.noop(),
new JdbcTransactionExecutor(connectionProvider),
new JdbcProcessDefinitionRepository(connectionProvider),
new JdbcProcessInstanceRepository(connectionProvider),
new JdbcApprovalTaskRepository(connectionProvider),
new JdbcProcessHistoryRepository(connectionProvider),
new JdbcActionExecutionRepository(connectionProvider),
List.of());
startEngine.register(dueDefinition);
ProcessInstance instance = startEngine.start("leave-due", "alice");
assertEquals(0, startEngine.processDue(10));
OrdoEngine laterEngine = new DefaultOrdoEngine(Clock.fixed(NOW.plusSeconds(3601), ZoneOffset.UTC),
AssigneeResolver.direct(), RoutingCondition.always(), ActionHandler.noop(),
new JdbcTransactionExecutor(connectionProvider),
new JdbcProcessDefinitionRepository(connectionProvider),
new JdbcProcessInstanceRepository(connectionProvider),
new JdbcApprovalTaskRepository(connectionProvider),
new JdbcProcessHistoryRepository(connectionProvider),
new JdbcActionExecutionRepository(connectionProvider),
List.of());
assertEquals(1, laterEngine.processDue(10));
ApprovalTask escalated = laterEngine.findPendingTasksByInstanceId(instance.id()).get(0);
assertEquals("diana", escalated.assignee());
assertNull(escalated.dueAt());
assertTrue(laterEngine.queryHistory(instance.id(), new PageRequest(0, 20)).content().stream()
.anyMatch(event -> event.type() == ProcessEventType.TASK_ESCALATED));
}
@Test
void rollsBackTheWholeApprovalWhenTheNextStepCannotBeCreated() {
OrdoEngine failingEngine = newEngine((candidate, step, context) -> {
@@ -174,7 +174,7 @@ class JdbcPostgresIntegrationTest {
JdbcApprovalTaskRepository taskRepository = new JdbcApprovalTaskRepository(connectionProvider);
ApprovalTask pending = new ApprovalTask("task-race-pg", "inst-race-pg", "manager", "Manager approval",
"maria", TaskStatus.PENDING, NOW, null, null);
"maria", TaskStatus.PENDING, NOW, null, null, null);
taskRepository.save(pending);
Instant completedAt = NOW.plusSeconds(30);
@@ -272,6 +272,6 @@ class JdbcPostgresIntegrationTest {
private static ApprovalTask completedTask(ApprovalTask pending, String comment, Instant completedAt) {
return new ApprovalTask(pending.id(), pending.instanceId(), pending.stepId(), pending.name(),
pending.assignee(), TaskStatus.APPROVED, pending.createdAt(), completedAt,
new TaskAction(pending.assignee(), comment, completedAt));
new TaskAction(pending.assignee(), comment, completedAt), pending.dueAt());
}
}
@@ -1,7 +1,9 @@
package com.jetlumen.ordo.storage.jdbc;
import com.jetlumen.ordo.api.ApprovalPolicy;
import com.jetlumen.ordo.api.ApprovalStep;
import com.jetlumen.ordo.api.ProcessDefinition;
import com.jetlumen.ordo.api.StepDue;
import com.jetlumen.ordo.api.StepKind;
import com.jetlumen.ordo.api.StepTransition;
import com.jetlumen.ordo.api.query.Page;
@@ -110,4 +112,13 @@ class JdbcProcessDefinitionRepositoryTest {
assertEquals(List.of("c-def"), pageTwo.content().stream().map(ProcessDefinition::id).toList());
assertFalse(pageTwo.hasNext());
}
@Test
void insertsAndReadsBackStepDue() {
ProcessDefinition definition = ProcessDefinition.linear("leave-due", "Leave request", List.of(
new ApprovalStep("manager", "Manager approval", List.of("maria"), ApprovalPolicy.ANY,
StepKind.APPROVAL, null, StepDue.reassign(java.time.Duration.parse("PT48H"), "director"))));
assertTrue(repository.insertIfAbsent(definition));
assertEquals(definition, repository.findById("leave-due").orElseThrow());
}
}
@@ -18,7 +18,8 @@ final class JdbcTestSupport {
"/db/migration/V2__add_step_candidates_and_policy.sql",
"/db/migration/V3__add_step_transitions.sql",
"/db/migration/V4__add_step_kind_and_action_key.sql",
"/db/migration/V5__add_process_event_and_action_execution.sql"
"/db/migration/V5__add_process_event_and_action_execution.sql",
"/db/migration/V6__add_step_due.sql"
};
private static final String[] SCHEMA_SQL = loadSchemas();