cloud.opencode.base.flow.persistence.memory.InMemoryFlowRepository

All Implemented Interfaces:: FlowRepository

public final class InMemoryFlowRepository extends Object implements FlowRepository

Thread-safe, in-memory FlowRepository for unit tests, integration tests, and short-lived demos. Not suitable for production: state is lost on JVM exit and there is no cross-process coordination. 线程安全的内存版 FlowRepository，用于单测 / 集测 / 短期演示。不适合生产：进程退出即丢数据，且无跨进程协调。

Implementation notes:

Mutable entity tables use ConcurrentHashMap.
Append-only history and vote lists use CopyOnWriteArrayList (small lists, infrequent writes).
updateInstance(FlowInstance) / updateTask(Task) implement optimistic locking via ConcurrentHashMap.compute(K, BiFunction): the new row is accepted only when the stored lockVersion equals the incoming lockVersion; on success the stored row is replaced with lockVersion + 1.
casIncrementAddsignDepth(String, int, int) uses compute as well for a single-shot CAS that respects both the version predicate and the addsign_depth < maxDepth predicate.
cancelOtherPendingTasks(String, String, String) iterates matching tasks and applies an optimistic-lock cancel update to each. It is best-effort with respect to concurrent writers: tasks updated between the snapshot read and the per-task CAS are simply skipped.

Since:: JDK 25, opencode-base-flow V1.0.5
Author:: Leon Soo

Field Summary

Fields

Modifier and Type

Field

Description

static final int

MAX_LOG_ENTRIES

Hard ceiling on the in-memory history / vote logs so a long-running test process or demo doesn't slowly OOM.
Constructor Summary

Constructors

Constructor

Description

InMemoryFlowRepository()

Constructs an empty in-memory repository.
Method Summary

Modifier and Type

Method

Description

void

appendHistory(HistoryRecord record)

Appends a new history row.

int

cancelAllTasksForStep(String instanceId, String stepId)

Bulk-transitions every task under (instanceId, stepId) whose current status is not TaskStatus.CANCELED into the TaskStatus.CANCELED state — including DONE / REJECTED rows from a prior approval round.

int

cancelOtherPendingTasks(String instanceId, String stepId, String exceptTaskId)

Bulk-cancels every other TaskStatus.PENDING task on the same step (excluding exceptTaskId), respecting optimistic-lock version.

int

casIncrementAddsignDepth(String instanceId, int expectedVersion, int maxDepth)

Compare-and-set: UPDATE addsign_depth = addsign_depth + 1, version = version + 1 WHERE id = ? AND version = ? AND addsign_depth < maxDepth.

Optional<FlowDefinition>

findActiveDefinition(String flowKey)

Returns the single FlowStatus.ACTIVE definition for the given business key, if any.

Optional<FlowDefinition>

findDefinitionById(String id)

Looks up a definition by its primary id.

List<FlowDefinition>

findDefinitionVersions(String flowKey)

Lists every persisted version of the given business key, in version ascending order.

List<HistoryRecord>

findHistoryByInstance(String instanceId)

Lists every history row for an instance in chronological order (earliest first).

Optional<HistoryRecord>

findHistoryByTaskId(String taskId)

Looks up a history row by its related task id (one-to-one for terminal task actions such as approve / reject).

List<HistoryRecord>

findHistoryByUser(String userId, int page, int size)

Paginated history query for the given user (matched on HistoryRecord.operatorId()).

Optional<FlowInstance>

findInstanceById(String id)

Looks up an instance by id.

List<FlowInstance>

findInstancesByInitiator(String userId, int page, int size)

Lists instances started by the given user, paginated.

Optional<Task>

findTaskById(String id)

Looks up a task by id.

List<Task>

findTasksByInstance(String instanceId)

Lists every task (any status) under the given instance.

List<Task>

findTasksByInstanceAndStatus(String instanceId, TaskStatus status)

Lists tasks that belong to the given instance and currently sit in the given status.

List<Task>

findTodoByUser(String userId, int page, int size)

Paginated todo-list query for the given user (status = TaskStatus.PENDING only).

List<VoteRecord>

findVotesByTask(String taskId)

Lists every vote recorded against the given task, in vote order.

void

recordVote(String taskId, String voterId, VoteOpinion opinion)

Records a single vote at a StepType.VOTE step.

void

runInTransaction(Runnable action)

Serialises the action under a reentrant lock and provides snapshot rollback if the action throws.

FlowDefinition

saveDefinition(FlowDefinition def)

Persists a new flow definition.

FlowInstance

saveInstance(FlowInstance inst)

Persists a new flow instance.

Task

saveTask(Task task)

Persists a new task.

void

updateDefinitionStatus(String id, FlowStatus newStatus)

Mutates the lifecycle status of a definition (ACTIVE / DEPRECATED).

int

updateInstance(FlowInstance inst)

Optimistic-lock UPDATE; returns the number of affected rows (0 indicates version conflict).

int

updateTask(Task task)

Optimistic-lock UPDATE; returns affected rows (0 = version conflict).

Methods inherited from class Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details
- MAX_LOG_ENTRIES
  public static final int MAX_LOG_ENTRIES
  
  Hard ceiling on the in-memory history / vote logs so a long-running test process or demo doesn't slowly OOM. When the cap is hit the repository fails-closed instead of silently dropping rows (which would corrupt the audit invariant). 单进程内 history / vote 日志总条数上限；上限到达后 fail-closed 拒写。
  
  See Also:
  
  Constant Field Values
Constructor Details
- InMemoryFlowRepository
  
  public InMemoryFlowRepository()
  
  Constructs an empty in-memory repository. 构造一个空的内存仓储。
  
  Since:
  
  JDK 25, opencode-base-flow V1.0.5
Method Details
- saveDefinition
  
  public FlowDefinition saveDefinition(FlowDefinition def)
  
  Description copied from interface: FlowRepository
  
  Persists a new flow definition. The returned record reflects the stored state (including any engine-assigned fields). 持久化一份新的流程定义。返回已存储的视图。
  
  Specified by:
  
  saveDefinition in interface FlowRepository
  
  Parameters:
  
  def - the definition to persist — 待持久化的定义
  
  Returns:
  
  the persisted definition — 已持久化的定义
- findDefinitionById
  
  public Optional<FlowDefinition> findDefinitionById(String id)
  
  Description copied from interface: FlowRepository
  
  Looks up a definition by its primary id. 根据主键 ID 查找定义。
  
  Specified by:
  
  findDefinitionById in interface FlowRepository
  
  Parameters:
  
  id - definition id — 定义 ID
  
  Returns:
  
  the definition wrapped in an Optional — 定义（Optional）
- findActiveDefinition
  
  public Optional<FlowDefinition> findActiveDefinition(String flowKey)
  
  Description copied from interface: FlowRepository
  
  Returns the single FlowStatus.ACTIVE definition for the given business key, if any. 返回指定业务 key 当前唯一启用的（FlowStatus.ACTIVE）定义。
  
  Specified by:
  
  findActiveDefinition in interface FlowRepository
  
  Parameters:
  
  flowKey - business flow key — 业务 key
  
  Returns:
  
  the active definition — 启用中的定义
- findDefinitionVersions
  
  public List<FlowDefinition> findDefinitionVersions(String flowKey)
  
  Description copied from interface: FlowRepository
  
  Lists every persisted version of the given business key, in version ascending order. 返回给定业务 key 所有版本（按版本号升序）。
  
  Specified by:
  
  findDefinitionVersions in interface FlowRepository
  
  Parameters:
  
  flowKey - business flow key — 业务 key
  
  Returns:
  
  all definition versions — 全部版本
- updateDefinitionStatus
  
  public void updateDefinitionStatus(String id, FlowStatus newStatus)
  
  Description copied from interface: FlowRepository
  
  Mutates the lifecycle status of a definition (ACTIVE / DEPRECATED). 修改某定义的生命周期状态（ACTIVE / DEPRECATED）。
  
  Specified by:
  
  updateDefinitionStatus in interface FlowRepository
  
  Parameters:
  
  id - definition id — 定义 ID
  
  newStatus - the new status — 新状态
- saveInstance
  
  public FlowInstance saveInstance(FlowInstance inst)
  
  Description copied from interface: FlowRepository
  
  Persists a new flow instance. 持久化新的流程实例。
  
  Specified by:
  
  saveInstance in interface FlowRepository
  
  Parameters:
  
  inst - the instance to persist — 待持久化的实例
  
  Returns:
  
  the persisted instance — 已持久化的实例
- findInstanceById
  
  public Optional<FlowInstance> findInstanceById(String id)
  
  Description copied from interface: FlowRepository
  
  Looks up an instance by id. 根据 ID 查找实例。
  
  Specified by:
  
  findInstanceById in interface FlowRepository
  
  Parameters:
  
  id - instance id — 实例 ID
  
  Returns:
  
  the instance wrapped in an Optional — 实例（Optional）
- findInstancesByInitiator
  
  public List<FlowInstance> findInstancesByInitiator(String userId, int page, int size)
  
  Description copied from interface: FlowRepository
  
  Lists instances started by the given user, paginated. 分页列出指定发起人启动的实例。
  
  Specified by:
  
  findInstancesByInitiator in interface FlowRepository
  
  Parameters:
  
  userId - initiator user id — 发起人 userId
  
  page - zero-based page index — 页码（0 起）
  
  size - page size — 每页大小
  
  Returns:
  
  instances on the requested page — 该页实例
- updateInstance
  
  public int updateInstance(FlowInstance inst)
  
  Description copied from interface: FlowRepository
  
  Optimistic-lock UPDATE; returns the number of affected rows (0 indicates version conflict). 乐观锁 UPDATE；返回 affected rows（0 表示版本冲突）。
  
  Specified by:
  
  updateInstance in interface FlowRepository
  
  Parameters:
  
  inst - the instance to update — 待更新的实例
  
  Returns:
  
  affected rows — 影响行数
- casIncrementAddsignDepth
  
  public int casIncrementAddsignDepth(String instanceId, int expectedVersion, int maxDepth)
  
  Description copied from interface: FlowRepository
  
  Compare-and-set: UPDATE addsign_depth = addsign_depth + 1, version = version + 1 WHERE id = ? AND version = ? AND addsign_depth < maxDepth. Returns affected rows (0 = conflict or depth exceeded). Callers must re-read the instance to distinguish the two cases. CAS：UPDATE addsign_depth + 1, version + 1 WHERE id = ? AND version = ? AND addsign_depth < maxDepth。返回 affected rows（0 表示冲突或超深）。调用方需复读 instance 区分两种情况。
  
  Specified by:
  
  casIncrementAddsignDepth in interface FlowRepository
  
  Parameters:
  
  instanceId - instance id — 实例 ID
  
  expectedVersion - expected current version — 期望的当前版本
  
  maxDepth - configured max addsign depth — 配置的加签链上限
  
  Returns:
  
  affected rows — 影响行数
- saveTask
  
  public Task saveTask(Task task)
  
  Description copied from interface: FlowRepository
  
  Persists a new task. 持久化新任务。
  
  Specified by:
  
  saveTask in interface FlowRepository
  
  Parameters:
  
  task - the task to persist — 待持久化的任务
  
  Returns:
  
  the persisted task — 已持久化的任务
- findTaskById
  
  public Optional<Task> findTaskById(String id)
  
  Description copied from interface: FlowRepository
  
  Looks up a task by id. 根据 ID 查找任务。
  
  Specified by:
  
  findTaskById in interface FlowRepository
  
  Parameters:
  
  id - task id — 任务 ID
  
  Returns:
  
  the task wrapped in an Optional — 任务（Optional）
- findTasksByInstanceAndStatus
  
  public List<Task> findTasksByInstanceAndStatus(String instanceId, TaskStatus status)
  
  Description copied from interface: FlowRepository
  
  Lists tasks that belong to the given instance and currently sit in the given status. 列出指定实例下、指定状态的任务。
  
  Specified by:
  
  findTasksByInstanceAndStatus in interface FlowRepository
  
  Parameters:
  
  instanceId - owning instance id — 所属实例 ID
  
  status - task status filter — 任务状态
  
  Returns:
  
  matching tasks — 符合条件的任务
- findTasksByInstance
  
  public List<Task> findTasksByInstance(String instanceId)
  
  Description copied from interface: FlowRepository
  
  Lists every task (any status) under the given instance. 列出指定实例下所有任务（不限状态）。
  
  Specified by:
  
  findTasksByInstance in interface FlowRepository
  
  Parameters:
  
  instanceId - owning instance id — 所属实例 ID
  
  Returns:
  
  all tasks — 所有任务
- findTodoByUser
  
  public List<Task> findTodoByUser(String userId, int page, int size)
  
  Description copied from interface: FlowRepository
  
  Paginated todo-list query for the given user (status = TaskStatus.PENDING only). 分页查询某用户的待办（仅 TaskStatus.PENDING）。
  
  Specified by:
  
  findTodoByUser in interface FlowRepository
  
  Parameters:
  
  userId - assignee user id — 承接人 userId
  
  page - zero-based page index — 页码（0 起）
  
  size - page size — 每页大小
  
  Returns:
  
  tasks on the requested page — 该页任务
- updateTask
  
  public int updateTask(Task task)
  
  Description copied from interface: FlowRepository
  
  Optimistic-lock UPDATE; returns affected rows (0 = version conflict). 乐观锁 UPDATE；返回 affected rows（0 表示版本冲突）。
  
  Specified by:
  
  updateTask in interface FlowRepository
  
  Parameters:
  
  task - the task to update — 待更新的任务
  
  Returns:
  
  affected rows — 影响行数
- cancelAllTasksForStep
  
  public int cancelAllTasksForStep(String instanceId, String stepId)
  
  Description copied from interface: FlowRepository
  
  Bulk-transitions every task under (instanceId, stepId) whose current status is not TaskStatus.CANCELED into the TaskStatus.CANCELED state — including DONE / REJECTED rows from a prior approval round. Used by RejectOperation when rolling back to a previously visited step so that StepEvaluator only ever observes tasks from the new batch. 批量将 (instanceId, stepId) 下所有非 CANCELED 状态的任务（含上一轮的 DONE / REJECTED）置为 TaskStatus.CANCELED。 RejectOperation 在回退到已访问节点时调用此方法，确保 StepEvaluator 只看见新一轮批次的任务。
  Invariant: each step has at most one active batch of tasks at any given time. Reject re-entry must enforce this by calling this method for every affected step before creating fresh PENDING rows.
  
  不变式：任意时刻每个 step 最多有一批活跃 task；reject 重入前必须对所有受影响 step 调用本方法。
  
  Specified by:
  
  cancelAllTasksForStep in interface FlowRepository
  
  Parameters:
  
  instanceId - owning instance id — 所属实例 ID
  
  stepId - step id — 节点 ID
  
  Returns:
  
  number of affected rows — 影响行数
- cancelOtherPendingTasks
  
  public int cancelOtherPendingTasks(String instanceId, String stepId, String exceptTaskId)
  
  Description copied from interface: FlowRepository
  
  Bulk-cancels every other TaskStatus.PENDING task on the same step (excluding exceptTaskId), respecting optimistic-lock version. Returns the number of tasks that were successfully canceled. 批量取消同节点其他 PENDING 任务（不含 exceptTaskId），遵守乐观锁；返回成功取消的条数。
  
  Specified by:
  
  cancelOtherPendingTasks in interface FlowRepository
  
  Parameters:
  
  instanceId - owning instance id — 所属实例 ID
  
  stepId - step id — 节点 ID
  
  exceptTaskId - the task id to keep — 保留的任务 ID
  
  Returns:
  
  number of canceled tasks — 成功取消条数
- appendHistory
  
  public void appendHistory(HistoryRecord record)
  
  Description copied from interface: FlowRepository
  
  Appends a new history row. History is append-only — once written, rows are never updated or deleted. 追加一条历史记录。历史表仅追加，写入后不可更新或删除。
  
  Specified by:
  
  appendHistory in interface FlowRepository
  
  Parameters:
  
  record - the history row to append — 待追加的历史
- findHistoryByInstance
  
  public List<HistoryRecord> findHistoryByInstance(String instanceId)
  
  Description copied from interface: FlowRepository
  
  Lists every history row for an instance in chronological order (earliest first). 按时间升序列出指定实例的所有历史记录。
  
  Specified by:
  
  findHistoryByInstance in interface FlowRepository
  
  Parameters:
  
  instanceId - owning instance id — 所属实例 ID
  
  Returns:
  
  chronological history — 时间升序的历史
- findHistoryByUser
  
  public List<HistoryRecord> findHistoryByUser(String userId, int page, int size)
  
  Description copied from interface: FlowRepository
  
  Paginated history query for the given user (matched on HistoryRecord.operatorId()). 分页查询某用户的历史记录（按 HistoryRecord.operatorId() 匹配）。
  
  Specified by:
  
  findHistoryByUser in interface FlowRepository
  
  Parameters:
  
  userId - operator user id — 操作人 userId
  
  page - zero-based page index — 页码（0 起）
  
  size - page size — 每页大小
  
  Returns:
  
  history rows on the requested page — 该页历史
- findHistoryByTaskId
  
  public Optional<HistoryRecord> findHistoryByTaskId(String taskId)
  
  Description copied from interface: FlowRepository
  
  Looks up a history row by its related task id (one-to-one for terminal task actions such as approve / reject). 按关联任务 ID 查找历史（approve / reject 等终态任务动作一对一）。
  
  Specified by:
  
  findHistoryByTaskId in interface FlowRepository
  
  Parameters:
  
  taskId - related task id — 关联任务 ID
  
  Returns:
  
  the history row wrapped in an Optional — 历史（Optional）
- recordVote
  
  public void recordVote(String taskId, String voterId, VoteOpinion opinion)
  
  Description copied from interface: FlowRepository
  
  Records a single vote at a StepType.VOTE step. Append-only. 在 StepType.VOTE 节点记录一票。追加写。
  
  Specified by:
  
  recordVote in interface FlowRepository
  
  Parameters:
  
  taskId - related task id — 关联任务 ID
  
  voterId - voter user id — 投票人
  
  opinion - vote opinion — 投票意见
- findVotesByTask
  
  public List<VoteRecord> findVotesByTask(String taskId)
  
  Description copied from interface: FlowRepository
  
  Lists every vote recorded against the given task, in vote order. 列出关联任务的所有投票（按投票顺序）。
  
  Specified by:
  
  findVotesByTask in interface FlowRepository
  
  Parameters:
  
  taskId - related task id — 关联任务 ID
  
  Returns:
  
  votes in submission order — 投票（按提交顺序）
- runInTransaction
  
  public void runInTransaction(Runnable action)
  
  Serialises the action under a reentrant lock and provides snapshot rollback if the action throws. Snapshot is taken on the outermost runInTransaction entry (nested calls join the outer scope and do not re-snapshot). Mutations to definitions, instances, tasks, historyLog, voteLog are reverted to the snapshot when the action throws, then the original exception propagates.
  Caveat: only mutations made via this repository are tracked. Concurrent writes from threads outside any runInTransaction call are not isolated; the lock only serialises tx callers.
  用可重入锁串行化动作并在抛异常时按快照回滚：仅最外层 tx 入口取快照，嵌套调用并入外层不重复快照；动作抛异常时还原 5 张表后再向上抛出。仅追踪经本仓储的写入；非 tx 路径上的并发写入不在隔离范围内。
  
  Specified by:
  
  runInTransaction in interface FlowRepository
  
  Parameters:
  
  action - the action to run — 待执行动作
  
  Since:
  
  JDK 25, opencode-base-flow V1.0.5

Class InMemoryFlowRepository

Field Summary

Constructor Summary

Method Summary

Methods inherited from class Object

Field Details

MAX_LOG_ENTRIES

Constructor Details

InMemoryFlowRepository

Method Details

saveDefinition

findDefinitionById

findActiveDefinition

findDefinitionVersions

updateDefinitionStatus

saveInstance

findInstanceById

findInstancesByInitiator

updateInstance

casIncrementAddsignDepth

saveTask

findTaskById

findTasksByInstanceAndStatus

findTasksByInstance

findTodoByUser

updateTask

cancelAllTasksForStep

cancelOtherPendingTasks

appendHistory

findHistoryByInstance

findHistoryByUser

findHistoryByTaskId

recordVote

findVotesByTask

runInTransaction