feat: persist process history events, listeners, and ACTION executions

Give hosts an append-only timeline, post-commit OrdoEventListener hooks, and durable ACTION results without blocking the approval flow.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
0264408
2026-09-15 10:14:44 +08:00
co-authored by Cursor
parent 77d1e5198b
commit f3220fd0a0
26 changed files with 1063 additions and 67 deletions
@@ -0,0 +1,94 @@
package com.jetlumen.ordo.storage.jdbc;
import com.jetlumen.ordo.api.ActionExecution;
import com.jetlumen.ordo.api.ActionExecutionStatus;
import com.jetlumen.ordo.api.query.Page;
import com.jetlumen.ordo.api.query.PageRequest;
import com.jetlumen.ordo.api.repository.ActionExecutionRepository;
import com.jetlumen.ordo.storage.jdbc.mapper.ActionExecutionMapper;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
/** JDBC implementation of the ACTION execution port. */
public final class JdbcActionExecutionRepository implements ActionExecutionRepository {
private static final String COLUMNS =
"id, instance_id, step_id, action_key, status, error_message, started_at, finished_at";
private static final String INSERT =
"INSERT INTO ordo_action_execution (id, instance_id, step_id, action_key, status, started_at)"
+ " VALUES (?, ?, ?, ?, ?, ?)";
private static final String COMPLETE_IF_PENDING =
"UPDATE ordo_action_execution SET status = ?, error_message = ?, finished_at = ?"
+ " WHERE id = ? AND status = 'PENDING'";
private final JdbcConnectionProvider connectionProvider;
public JdbcActionExecutionRepository(JdbcConnectionProvider connectionProvider) {
this.connectionProvider = Objects.requireNonNull(connectionProvider, "connectionProvider must not be null");
}
@Override
public void insert(ActionExecution execution) {
Objects.requireNonNull(execution, "execution must not be null");
Connection connection = connectionProvider.getConnection();
try (PreparedStatement insert = connection.prepareStatement(INSERT)) {
ActionExecutionMapper.bindInsert(insert, execution);
insert.executeUpdate();
} catch (SQLException e) {
throw new JdbcStorageException("failed to insert action execution: " + execution.id(), e);
} finally {
connectionProvider.close(connection);
}
}
@Override
public boolean complete(String executionId, ActionExecutionStatus status, String errorMessage, Instant finishedAt) {
Objects.requireNonNull(executionId, "executionId must not be null");
Objects.requireNonNull(status, "status must not be null");
Objects.requireNonNull(finishedAt, "finishedAt must not be null");
Connection connection = connectionProvider.getConnection();
try (PreparedStatement update = connection.prepareStatement(COMPLETE_IF_PENDING)) {
ActionExecutionMapper.bindComplete(update, executionId, status, errorMessage, finishedAt);
return update.executeUpdate() == 1;
} catch (SQLException e) {
throw new JdbcStorageException("failed to complete action execution: " + executionId, e);
} finally {
connectionProvider.close(connection);
}
}
@Override
public Page<ActionExecution> query(String instanceId, PageRequest pageRequest) {
Objects.requireNonNull(instanceId, "instanceId must not be null");
Objects.requireNonNull(pageRequest, "pageRequest must not be null");
Connection connection = connectionProvider.getConnection();
try {
long total = PageSupport.count(connection,
"SELECT COUNT(*) FROM ordo_action_execution WHERE instance_id = ?", List.of(instanceId));
String sql = "SELECT " + COLUMNS + " FROM ordo_action_execution WHERE instance_id = ?"
+ " ORDER BY started_at ASC, id ASC LIMIT ? OFFSET ?";
List<ActionExecution> content = new ArrayList<>();
try (PreparedStatement select = connection.prepareStatement(sql)) {
select.setString(1, instanceId);
select.setInt(2, pageRequest.size());
select.setInt(3, pageRequest.offset());
try (ResultSet resultSet = select.executeQuery()) {
while (resultSet.next()) {
content.add(ActionExecutionMapper.read(resultSet));
}
}
}
return new Page<>(content, total, pageRequest.page(), pageRequest.size());
} catch (SQLException e) {
throw new JdbcStorageException("failed to query action executions for instance: " + instanceId, e);
} finally {
connectionProvider.close(connection);
}
}
}
@@ -0,0 +1,73 @@
package com.jetlumen.ordo.storage.jdbc;
import com.jetlumen.ordo.api.ProcessEvent;
import com.jetlumen.ordo.api.query.Page;
import com.jetlumen.ordo.api.query.PageRequest;
import com.jetlumen.ordo.api.repository.ProcessHistoryRepository;
import com.jetlumen.ordo.storage.jdbc.mapper.ProcessEventMapper;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
/** JDBC implementation of the process history port. */
public final class JdbcProcessHistoryRepository implements ProcessHistoryRepository {
private static final String COLUMNS =
"id, instance_id, task_id, step_id, event_type, actor, detail, occurred_at";
private static final String INSERT =
"INSERT INTO ordo_process_event (id, instance_id, task_id, step_id, event_type, actor, detail, occurred_at)"
+ " VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
private final JdbcConnectionProvider connectionProvider;
public JdbcProcessHistoryRepository(JdbcConnectionProvider connectionProvider) {
this.connectionProvider = Objects.requireNonNull(connectionProvider, "connectionProvider must not be null");
}
@Override
public void append(ProcessEvent event) {
Objects.requireNonNull(event, "event must not be null");
Connection connection = connectionProvider.getConnection();
try (PreparedStatement insert = connection.prepareStatement(INSERT)) {
ProcessEventMapper.bindInsert(insert, event);
insert.executeUpdate();
} catch (SQLException e) {
throw new JdbcStorageException("failed to append process event: " + event.id(), e);
} finally {
connectionProvider.close(connection);
}
}
@Override
public Page<ProcessEvent> query(String instanceId, PageRequest pageRequest) {
Objects.requireNonNull(instanceId, "instanceId must not be null");
Objects.requireNonNull(pageRequest, "pageRequest must not be null");
Connection connection = connectionProvider.getConnection();
try {
long total = PageSupport.count(connection,
"SELECT COUNT(*) FROM ordo_process_event WHERE instance_id = ?", List.of(instanceId));
String sql = "SELECT " + COLUMNS + " FROM ordo_process_event WHERE instance_id = ?"
+ " ORDER BY occurred_at ASC, id ASC LIMIT ? OFFSET ?";
List<ProcessEvent> content = new ArrayList<>();
try (PreparedStatement select = connection.prepareStatement(sql)) {
select.setString(1, instanceId);
select.setInt(2, pageRequest.size());
select.setInt(3, pageRequest.offset());
try (ResultSet resultSet = select.executeQuery()) {
while (resultSet.next()) {
content.add(ProcessEventMapper.read(resultSet));
}
}
}
return new Page<>(content, total, pageRequest.page(), pageRequest.size());
} catch (SQLException e) {
throw new JdbcStorageException("failed to query process events for instance: " + instanceId, e);
} finally {
connectionProvider.close(connection);
}
}
}
@@ -0,0 +1,50 @@
package com.jetlumen.ordo.storage.jdbc.mapper;
import com.jetlumen.ordo.api.ActionExecution;
import com.jetlumen.ordo.api.ActionExecutionStatus;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
/** Maps rows of {@code ordo_action_execution} to {@link ActionExecution} objects and back. */
public final class ActionExecutionMapper {
private ActionExecutionMapper() {
}
public static void bindInsert(PreparedStatement statement, ActionExecution execution) throws SQLException {
statement.setString(1, execution.id());
statement.setString(2, execution.instanceId());
statement.setString(3, execution.stepId());
statement.setString(4, execution.actionKey());
statement.setString(5, execution.status().name());
statement.setTimestamp(6, Timestamp.from(execution.startedAt()));
}
public static ActionExecution read(ResultSet resultSet) throws SQLException {
Timestamp finishedAt = resultSet.getTimestamp("finished_at");
return new ActionExecution(
resultSet.getString("id"),
resultSet.getString("instance_id"),
resultSet.getString("step_id"),
resultSet.getString("action_key"),
ActionExecutionStatus.valueOf(resultSet.getString("status")),
resultSet.getString("error_message"),
resultSet.getTimestamp("started_at").toInstant(),
finishedAt == null ? null : finishedAt.toInstant());
}
public static void bindComplete(PreparedStatement statement, String executionId, ActionExecutionStatus status,
String errorMessage, java.time.Instant finishedAt) throws SQLException {
statement.setString(1, status.name());
if (errorMessage == null || errorMessage.isBlank()) {
statement.setNull(2, Types.VARCHAR);
} else {
statement.setString(2, errorMessage);
}
statement.setTimestamp(3, Timestamp.from(finishedAt));
statement.setString(4, executionId);
}
}
@@ -0,0 +1,47 @@
package com.jetlumen.ordo.storage.jdbc.mapper;
import com.jetlumen.ordo.api.ProcessEvent;
import com.jetlumen.ordo.api.ProcessEventType;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
/** Maps rows of {@code ordo_process_event} to {@link ProcessEvent} objects and back. */
public final class ProcessEventMapper {
private ProcessEventMapper() {
}
public static void bindInsert(PreparedStatement statement, ProcessEvent event) throws SQLException {
statement.setString(1, event.id());
statement.setString(2, event.instanceId());
setNullableString(statement, 3, event.taskId());
setNullableString(statement, 4, event.stepId());
statement.setString(5, event.type().name());
setNullableString(statement, 6, event.actor());
setNullableString(statement, 7, event.detail());
statement.setTimestamp(8, Timestamp.from(event.occurredAt()));
}
public static ProcessEvent read(ResultSet resultSet) throws SQLException {
return new ProcessEvent(
resultSet.getString("id"),
resultSet.getString("instance_id"),
resultSet.getString("task_id"),
resultSet.getString("step_id"),
ProcessEventType.valueOf(resultSet.getString("event_type")),
resultSet.getString("actor"),
resultSet.getString("detail"),
resultSet.getTimestamp("occurred_at").toInstant());
}
private static void setNullableString(PreparedStatement statement, int index, String value) throws SQLException {
if (value == null) {
statement.setNull(index, Types.VARCHAR);
} else {
statement.setString(index, value);
}
}
}
@@ -0,0 +1,29 @@
-- Process history events and ACTION-step execution records.
CREATE TABLE ordo_process_event (
id VARCHAR(36) PRIMARY KEY,
instance_id VARCHAR(36) NOT NULL,
task_id VARCHAR(36),
step_id VARCHAR(64),
event_type VARCHAR(32) NOT NULL,
actor VARCHAR(255),
detail TEXT,
occurred_at TIMESTAMP NOT NULL,
CONSTRAINT fk_process_event_instance FOREIGN KEY (instance_id) REFERENCES ordo_process_instance (id)
);
CREATE INDEX idx_process_event_instance_time ON ordo_process_event (instance_id, occurred_at, id);
CREATE TABLE ordo_action_execution (
id VARCHAR(36) PRIMARY KEY,
instance_id VARCHAR(36) NOT NULL,
step_id VARCHAR(64) NOT NULL,
action_key VARCHAR(255) NOT NULL,
status VARCHAR(32) NOT NULL,
error_message TEXT,
started_at TIMESTAMP NOT NULL,
finished_at TIMESTAMP,
CONSTRAINT fk_action_execution_instance FOREIGN KEY (instance_id) REFERENCES ordo_process_instance (id)
);
CREATE INDEX idx_action_execution_instance ON ordo_action_execution (instance_id);
@@ -0,0 +1,66 @@
package com.jetlumen.ordo.storage.jdbc;
import com.jetlumen.ordo.api.ActionExecution;
import com.jetlumen.ordo.api.ActionExecutionStatus;
import com.jetlumen.ordo.api.ApprovalStep;
import com.jetlumen.ordo.api.ProcessContext;
import com.jetlumen.ordo.api.ProcessDefinition;
import com.jetlumen.ordo.api.ProcessInstance;
import com.jetlumen.ordo.api.ProcessStatus;
import com.jetlumen.ordo.api.query.Page;
import com.jetlumen.ordo.api.query.PageRequest;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
class JdbcActionExecutionRepositoryTest {
private static final Instant T0 = Instant.parse("2026-03-01T08:00:00Z");
private JdbcActionExecutionRepository repository;
@BeforeEach
void setUp() {
JdbcConnectionProvider connectionProvider = new JdbcConnectionProvider(JdbcTestSupport.newDataSource());
repository = new JdbcActionExecutionRepository(connectionProvider);
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(ProcessDefinition.linear("leave",
"Leave request", List.of(ApprovalStep.single("manager", "Manager approval", "maria"))));
new JdbcProcessInstanceRepository(connectionProvider).insert(new ProcessInstance("inst-1", "leave", "alice",
ProcessStatus.RUNNING, T0, null, ProcessContext.empty()));
}
@Test
void roundTripsPendingThenCompletedExecution() {
ActionExecution pending = new ActionExecution("ex-1", "inst-1", "notify", "leave-mail",
ActionExecutionStatus.PENDING, null, T0, null);
repository.insert(pending);
assertTrue(repository.complete("ex-1", ActionExecutionStatus.SUCCESS, null, T0.plusSeconds(2)));
Page<ActionExecution> page = repository.query("inst-1", new PageRequest(0, 10));
assertEquals(1, page.totalElements());
ActionExecution stored = page.content().getFirst();
assertEquals(ActionExecutionStatus.SUCCESS, stored.status());
assertEquals(T0.plusSeconds(2), stored.finishedAt());
assertFalse(repository.complete("ex-1", ActionExecutionStatus.FAILED, "nope", T0.plusSeconds(3)));
}
@Test
void completeFailsForUnknownExecution() {
assertFalse(repository.complete("missing", ActionExecutionStatus.FAILED, "x", T0));
}
@Test
void queryOrdersOldestFirst() {
repository.insert(new ActionExecution("ex-1", "inst-1", "a", "first", ActionExecutionStatus.PENDING, null,
T0, null));
repository.insert(new ActionExecution("ex-2", "inst-1", "b", "second", ActionExecutionStatus.PENDING, null,
T0.plusSeconds(1), null));
Page<ActionExecution> page = repository.query("inst-1", new PageRequest(0, 10));
assertEquals(List.of("ex-1", "ex-2"), page.content().stream().map(ActionExecution::id).toList());
}
}
@@ -1,5 +1,6 @@
package com.jetlumen.ordo.storage.jdbc;
import com.jetlumen.ordo.api.ActionExecutionStatus;
import com.jetlumen.ordo.api.ActionHandler;
import com.jetlumen.ordo.api.ApprovalStep;
import com.jetlumen.ordo.api.ApprovalTask;
@@ -7,8 +8,11 @@ import com.jetlumen.ordo.api.AssigneeResolver;
import com.jetlumen.ordo.api.OrdoEngine;
import com.jetlumen.ordo.api.ProcessContext;
import com.jetlumen.ordo.api.ProcessDefinition;
import com.jetlumen.ordo.api.ProcessEvent;
import com.jetlumen.ordo.api.ProcessEventType;
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.StepTransition;
import com.jetlumen.ordo.api.TaskStatus;
@@ -124,7 +128,10 @@ class JdbcOrdoEngineIntegrationTest {
new JdbcTransactionExecutor(connectionProvider),
new JdbcProcessDefinitionRepository(connectionProvider),
new JdbcProcessInstanceRepository(connectionProvider),
new JdbcApprovalTaskRepository(connectionProvider));
new JdbcApprovalTaskRepository(connectionProvider),
new JdbcProcessHistoryRepository(connectionProvider),
new JdbcActionExecutionRepository(connectionProvider),
List.of());
failingEngine.register(new ProcessDefinition("leave-noroute", "Leave request",
List.of(ApprovalStep.single("manager", "Manager approval", "maria")),
List.of(StepTransition.endWhen("manager", "never", 0))));
@@ -137,6 +144,9 @@ class JdbcOrdoEngineIntegrationTest {
assertEquals(TaskStatus.PENDING, storedTask.status());
assertNull(storedTask.action());
assertEquals(ProcessStatus.RUNNING, failingEngine.findInstance(instance.id()).orElseThrow().status());
assertEquals(List.of(ProcessEventType.INSTANCE_STARTED, ProcessEventType.TASK_CREATED),
failingEngine.queryHistory(instance.id(), new PageRequest(0, 20)).content().stream()
.map(ProcessEvent::type).toList());
}
@Test
@@ -196,12 +206,55 @@ class JdbcOrdoEngineIntegrationTest {
assertThrows(InstanceAlreadyCompletedException.class, () -> engine.withdraw(instance.id(), "alice"));
}
@Test
void persistHistoryAndActionExecutions() {
OrdoEngine actionEngine = new DefaultOrdoEngine(Clock.fixed(NOW, ZoneOffset.UTC), AssigneeResolver.direct(),
RoutingCondition.always(),
(key, context) -> {
if ("boom".equals(key)) {
throw new IllegalStateException("mail failed");
}
},
new JdbcTransactionExecutor(connectionProvider),
new JdbcProcessDefinitionRepository(connectionProvider),
new JdbcProcessInstanceRepository(connectionProvider),
new JdbcApprovalTaskRepository(connectionProvider),
new JdbcProcessHistoryRepository(connectionProvider),
new JdbcActionExecutionRepository(connectionProvider),
List.of());
actionEngine.register(new ProcessDefinition("leave-action", "Leave request", List.of(
ApprovalStep.single("manager", "Manager approval", "maria"),
ApprovalStep.action("notify", "Notify", "ok-mail"),
ApprovalStep.action("fail", "Fail", "boom")),
List.of(
StepTransition.always("manager", "notify"),
StepTransition.always("notify", "fail"),
StepTransition.end("fail"))));
ProcessInstance instance = actionEngine.start("leave-action", "alice");
actionEngine.approve(actionEngine.findPendingTasksByInstanceId(instance.id()).getFirst().id(), "maria");
assertEquals(ProcessStatus.APPROVED, actionEngine.findInstance(instance.id()).orElseThrow().status());
assertEquals(ActionExecutionStatus.SUCCESS,
actionEngine.queryActionExecutions(instance.id(), new PageRequest(0, 10)).content().getFirst().status());
assertEquals(ActionExecutionStatus.FAILED,
actionEngine.queryActionExecutions(instance.id(), new PageRequest(0, 10)).content().get(1).status());
List<ProcessEventType> types = actionEngine.queryHistory(instance.id(), new PageRequest(0, 50)).content()
.stream().map(ProcessEvent::type).toList();
assertTrue(types.contains(ProcessEventType.ACTION_SUCCEEDED));
assertTrue(types.contains(ProcessEventType.ACTION_FAILED));
assertTrue(types.contains(ProcessEventType.INSTANCE_APPROVED));
}
private OrdoEngine newEngine(AssigneeResolver assigneeResolver) {
return new DefaultOrdoEngine(Clock.fixed(NOW, ZoneOffset.UTC), assigneeResolver, RoutingCondition.always(),
ActionHandler.noop(),
new JdbcTransactionExecutor(connectionProvider),
new JdbcProcessDefinitionRepository(connectionProvider),
new JdbcProcessInstanceRepository(connectionProvider),
new JdbcApprovalTaskRepository(connectionProvider));
new JdbcApprovalTaskRepository(connectionProvider),
new JdbcProcessHistoryRepository(connectionProvider),
new JdbcActionExecutionRepository(connectionProvider),
List.of());
}
}
@@ -224,7 +224,10 @@ class JdbcPostgresIntegrationTest {
new JdbcTransactionExecutor(connectionProvider),
new JdbcProcessDefinitionRepository(connectionProvider),
new JdbcProcessInstanceRepository(connectionProvider),
new JdbcApprovalTaskRepository(connectionProvider));
new JdbcApprovalTaskRepository(connectionProvider),
new JdbcProcessHistoryRepository(connectionProvider),
new JdbcActionExecutionRepository(connectionProvider),
List.of());
}
private static PGSimpleDataSource newDataSource(String currentSchema) {
@@ -0,0 +1,60 @@
package com.jetlumen.ordo.storage.jdbc;
import com.jetlumen.ordo.api.ApprovalStep;
import com.jetlumen.ordo.api.ProcessContext;
import com.jetlumen.ordo.api.ProcessDefinition;
import com.jetlumen.ordo.api.ProcessEvent;
import com.jetlumen.ordo.api.ProcessEventType;
import com.jetlumen.ordo.api.ProcessInstance;
import com.jetlumen.ordo.api.ProcessStatus;
import com.jetlumen.ordo.api.query.Page;
import com.jetlumen.ordo.api.query.PageRequest;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
class JdbcProcessHistoryRepositoryTest {
private static final Instant T0 = Instant.parse("2026-03-01T08:00:00Z");
private JdbcProcessHistoryRepository repository;
@BeforeEach
void setUp() {
JdbcConnectionProvider connectionProvider = new JdbcConnectionProvider(JdbcTestSupport.newDataSource());
repository = new JdbcProcessHistoryRepository(connectionProvider);
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(ProcessDefinition.linear("leave",
"Leave request", List.of(ApprovalStep.single("manager", "Manager approval", "maria"))));
new JdbcProcessInstanceRepository(connectionProvider).insert(new ProcessInstance("inst-1", "leave", "alice",
ProcessStatus.RUNNING, T0, null, ProcessContext.empty()));
new JdbcProcessInstanceRepository(connectionProvider).insert(new ProcessInstance("inst-2", "leave", "bob",
ProcessStatus.RUNNING, T0, null, ProcessContext.empty()));
}
@Test
void appendsAndQueriesOldestFirstForOneInstance() {
ProcessEvent started = event("e1", "inst-1", ProcessEventType.INSTANCE_STARTED, T0);
ProcessEvent created = event("e2", "inst-1", ProcessEventType.TASK_CREATED, T0.plusSeconds(1));
ProcessEvent other = event("e3", "inst-2", ProcessEventType.INSTANCE_STARTED, T0);
repository.append(started);
repository.append(created);
repository.append(other);
Page<ProcessEvent> page = repository.query("inst-1", new PageRequest(0, 10));
assertEquals(2, page.totalElements());
assertEquals(List.of(started, created), page.content());
Page<ProcessEvent> first = repository.query("inst-1", new PageRequest(0, 1));
assertEquals(1, first.content().size());
assertEquals("e1", first.content().getFirst().id());
assertTrue(first.hasNext());
}
private static ProcessEvent event(String id, String instanceId, ProcessEventType type, Instant at) {
return new ProcessEvent(id, instanceId, null, null, type, "alice", null, at);
}
}
@@ -17,7 +17,8 @@ final class JdbcTestSupport {
"/db/migration/V1__create_ordo_tables.sql",
"/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/V4__add_step_kind_and_action_key.sql",
"/db/migration/V5__add_process_event_and_action_execution.sql"
};
private static final String[] SCHEMA_SQL = loadSchemas();