Skip to content

Commit 984510d

Browse files
authored
Record a child enqueue's deduplication collision as the parent's step (#625)
Closes #602. When a workflow enqueued a child whose deduplication ID another workflow held, `DBOSQueueDuplicatedException` reached the parent with nothing recorded at the parent's step. A parent that caught it and took another path, then was recovered after the holder finished, enqueued the child for real on replay, and its later steps diverged from their checkpoints. #602's second problem, the child's insert and the parent's step committing separately, was already fixed by #616: `initChildWorkflowStatus` writes both in one transaction. ## Changes **1. The collision is the parent's step result.** When a child enqueue collides, the parent records `DBOSQueueDuplicatedException` as its step's result, with no child ID, then rethrows it. The record is fenced on the parent's execution: a parent that no longer owns its workflow is refused, and stops. If recording fails, that failure is thrown with the collision attached as suppressed; throwing the collision unrecorded would let the parent take a path its replay can't repeat. **2. A replay rethrows it.** The replay checks before a child start or enqueue (`executeWorkflow`, `enqueueWorkflowByName`, `enqueueDebounced`) go through one helper, `recordedChild`. `checkChildWorkflow` now returns the recorded step rather than only its child ID, so a recorded collision is thrown again instead of looking like an empty slot. **3. The recorded step must be this call's.** A replay accepts what is recorded at the slot only when it was recorded under the name of the workflow being started; otherwise the step is a different one and the replay throws `DBOSUnexpectedStepException`, as `checkStepResult` does for any other step. This applies to a recorded child as well as a recorded collision. Every release so far recorded a child under the workflow's name, so their histories still replay. - **Exemption:** releases before 1.2 recorded the debouncer workflow they started in a debounced enqueue's slot, under `Constants.DEBOUNCER_WORKFLOW_NAME`. `enqueueDebounced` accepts that name there, and returns the recorded child itself rather than leave it to `executeWorkflow`. The exemption is kept until 2.0, which can drop replay support for workflows that debounced before 1.2. - **Behaviour change:** a slot holding a different step is now reported as `DBOSUnexpectedStepException` when the replay reads it, rather than as `DBOSStepNondeterminismException` when the child's start writes it. **4. Debouncer's own replay of a collision stays.** It throws the collision again when the slot recorded nothing and the next slot is `DBOS.lookupDebouncer`. New histories record the collision, so the general replay handles them, but histories recorded before this change still need it. Matches Python (`_init_workflow` and `_check_recorded_enqueue` in `_core.py`) and TypeScript (`enqueue_workflow.ts`), which record the collision as the parent's step and rethrow it on replay. They also record a rejected workflow-ID reuse the same way; Java has no ID reuse policy yet. ## Tests `ChildEnqueueErrorTest`: - A parent that catches a deduplicated child enqueue records the collision as its step, and when recovered after the holder finishes, still falls back and enqueues no child. Fails without the recording. - A collision recorded under another step's name, and a child recorded under another step's name, each make the recovered parent throw `DBOSUnexpectedStepException`. Each fails without its name check. `DebouncerTest.replaysADebounceThatStartedADebouncerWorkflow`, which seeds the pre-1.2 slot, covers the exemption; it fails without it. Run locally: the queue, child-workflow, debouncer, ownership-fence, ownership and invocation tests. The full suite has not been run locally; this PR's CI run is that pass. 🤖 Generated with [Claude Code](https://claude.com/claude-code)
1 parent 110c8d7 commit 984510d

4 files changed

Lines changed: 352 additions & 19 deletions

File tree

‎transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1530,7 +1530,8 @@ public List<String> startQueuedPartitionedWorkflows(
15301530
ownerXid));
15311531
}
15321532

1533-
public Optional<String> checkChildWorkflow(String workflowUuid, int functionId) {
1533+
/** See {@link WorkflowDAO#checkChildWorkflow}. */
1534+
public Optional<StepResult> checkChildWorkflow(String workflowUuid, int functionId) {
15341535
return dbRetry(() -> WorkflowDAO.checkChildWorkflow(ctx, workflowUuid, functionId));
15351536
}
15361537

‎transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java‎

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2110,12 +2110,17 @@ public static <T> Result<T> awaitWorkflowResult(
21102110
}
21112111
}
21122112

2113-
public static Optional<String> checkChildWorkflow(
2113+
/**
2114+
* What a parent's step recorded for a child it started or enqueued: the child's ID, or the
2115+
* deduplication collision its enqueue failed with. Empty when nothing is recorded at that step.
2116+
*/
2117+
public static Optional<StepResult> checkChildWorkflow(
21142118
DbContext ctx, String workflowUuid, int functionId) throws SQLException {
21152119

21162120
final String sql =
21172121
"""
2118-
SELECT child_workflow_id FROM "%s".operation_outputs WHERE workflow_uuid = ? AND function_id = ?
2122+
SELECT function_name, child_workflow_id, error, serialization
2123+
FROM "%s".operation_outputs WHERE workflow_uuid = ? AND function_id = ?
21192124
"""
21202125
.formatted(ctx.schema());
21212126

@@ -2127,8 +2132,15 @@ public static Optional<String> checkChildWorkflow(
21272132

21282133
try (ResultSet rs = stmt.executeQuery()) {
21292134
if (rs.next()) {
2130-
String childWorkflowId = rs.getString("child_workflow_id");
2131-
return childWorkflowId != null ? Optional.of(childWorkflowId) : Optional.empty();
2135+
return Optional.of(
2136+
new StepResult(
2137+
workflowUuid,
2138+
functionId,
2139+
rs.getString("function_name"),
2140+
null,
2141+
SystemDatabase.errorOrNull(rs.getString("error")),
2142+
rs.getString("child_workflow_id"),
2143+
rs.getString("serialization")));
21322144
}
21332145
return Optional.empty();
21342146
}

‎transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java‎

Lines changed: 84 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -530,10 +530,11 @@ public Object debounceDelayedWorkflow(
530530
* that slot recorded -- which, for a debounce recorded before debounced workflows, is the
531531
* debouncer workflow, not {@code workflowId}.
532532
*
533-
* <p>An enqueue that collided records nothing in its slot, so a replay alone cannot tell it from
534-
* one that never ran. A collision is always followed by the {@code DBOS.lookupDebouncer} step in
535-
* the next slot, in this release and the ones before it, so when that step is recorded the replay
536-
* throws the collision again rather than enqueue for real: the key may be free by now, or held on
533+
* <p>An enqueue that collides records the collision in its slot, and a replay throws it again.
534+
* Releases before 1.3 recorded nothing there, so a replay alone cannot tell their collision from
535+
* an enqueue that never ran. A collision is always followed by the {@code DBOS.lookupDebouncer}
536+
* step in the next slot, so when that step is recorded and the slot is empty, the replay throws
537+
* the collision again rather than enqueue for real: the key may be free by now, or held on
537538
* another queue than the one it collided on.
538539
*
539540
* <p>The workflow takes {@code timeout}, timed from its dequeue, and no deadline: it may start
@@ -554,10 +555,16 @@ public <T, E extends Exception> WorkflowHandle<T, E> enqueueDebounced(
554555
@Nullable Map<String, Object> attributes) {
555556
var ctx = DBOSContextHolder.get();
556557
var parent = getParent(ctx);
557-
if (parent != null
558-
&& systemDatabase.checkChildWorkflow(parent.workflowId(), parent.functionId()).isEmpty()
559-
&& lookupDebouncerRecordedAfter(parent)) {
560-
throw new DBOSQueueDuplicatedException(workflowId, queueName, deduplicationId);
558+
if (parent != null) {
559+
// Returned here rather than left to executeWorkflow, which would not accept the debouncer
560+
// workflow a release before 1.2 recorded in this slot.
561+
var childId = recordedChild(parent, workflow.workflowName(), true);
562+
if (childId != null) {
563+
return retrieveWorkflow(childId);
564+
}
565+
if (lookupDebouncerRecordedAfter(parent)) {
566+
throw new DBOSQueueDuplicatedException(workflowId, queueName, deduplicationId);
567+
}
561568
}
562569
var options =
563570
debouncedOptions(
@@ -2166,9 +2173,9 @@ public <T, E extends Exception> WorkflowHandle<T, E> enqueueWorkflowByName(
21662173
ctx.getNextWorkflowId(childWorkflowId), () -> UUID.randomUUID().toString()));
21672174

21682175
if (parent != null) {
2169-
var childId = systemDatabase.checkChildWorkflow(parent.workflowId(), parent.functionId());
2170-
if (childId.isPresent()) {
2171-
return retrieveWorkflow(childId.get());
2176+
var childId = recordedChild(parent, options.workflowName(), false);
2177+
if (childId != null) {
2178+
return retrieveWorkflow(childId);
21722179
}
21732180
}
21742181

@@ -2373,6 +2380,42 @@ private void validateQueue(String queueName, String queuePartitionKey) {
23732380
}
23742381
}
23752382

2383+
/**
2384+
* The child the parent's current step recorded when it first ran, or null when it recorded none.
2385+
* When the step recorded a deduplication collision for this start or enqueue, throws it again, so
2386+
* a replay sees what the original call saw.
2387+
*
2388+
* <p>The step is this call's only if it was recorded under the same workflow name; otherwise it
2389+
* is another one, and the replay is not deterministic. One exception, for a debounced enqueue
2390+
* ({@code debounced}): releases before 1.2 recorded the debouncer workflow they started in its
2391+
* slot, under that workflow's name.
2392+
*/
2393+
private @Nullable String recordedChild(
2394+
WorkflowInfo parent, String workflowName, boolean debounced) {
2395+
var recorded = systemDatabase.checkChildWorkflow(parent.workflowId(), parent.functionId());
2396+
if (recorded.isEmpty()) {
2397+
return null;
2398+
}
2399+
var step = recorded.get();
2400+
var debouncerWorkflow =
2401+
debounced
2402+
&& step.error() == null
2403+
&& step.childWorkflowId() != null
2404+
&& Constants.DEBOUNCER_WORKFLOW_NAME.equals(step.stepName());
2405+
if (!debouncerWorkflow && !workflowName.equals(step.stepName())) {
2406+
throw new DBOSUnexpectedStepException(
2407+
parent.workflowId(), parent.functionId(), workflowName, step.stepName());
2408+
}
2409+
if (step.error() != null) {
2410+
var t = SerializationUtil.deserializeError(step.error(), step.serialization(), serializer);
2411+
if (t instanceof RuntimeException re) {
2412+
throw re;
2413+
}
2414+
throw new RuntimeException(t.getMessage(), t);
2415+
}
2416+
return step.childWorkflowId();
2417+
}
2418+
23762419
// execute or enqueue a worklow
23772420
private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
23782421
RegisteredWorkflow workflow, Object[] args, ExecutionOptions options, WorkflowInfo parent) {
@@ -2384,9 +2427,9 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
23842427
}
23852428

23862429
if (parent != null) {
2387-
var childId = systemDatabase.checkChildWorkflow(parent.workflowId(), parent.functionId());
2388-
if (childId.isPresent()) {
2389-
return retrieveWorkflow(childId.get());
2430+
var childId = recordedChild(parent, workflow.workflowName(), false);
2431+
if (childId != null) {
2432+
return retrieveWorkflow(childId);
23902433
}
23912434
}
23922435

@@ -2702,6 +2745,7 @@ public static void enqueueWorkflow(
27022745
RegisteredWorkflow.fullyQualifiedName(
27032746
workflowName, Objects.requireNonNullElse(className, ""), instanceName));
27042747

2748+
var startTime = System.currentTimeMillis();
27052749
try {
27062750
persistWorkflow(
27072751
systemDatabase,
@@ -2719,6 +2763,32 @@ public static void enqueueWorkflow(
27192763
} catch (DBOSQueueDuplicatedException e) {
27202764
logger.debug(
27212765
"Workflow queue {} reused deduplicationId {}", e.queueName(), e.deduplicationId());
2766+
// A parent records the collision as its step, so a replay throws it again even after the
2767+
// holder has finished and the enqueue would succeed. Only while the parent's execution still
2768+
// owns the parent: one that does not is refused here, and stops.
2769+
// A failure to record it is thrown instead, with the collision attached: thrown unrecorded,
2770+
// the collision would let the parent take a path its replay cannot repeat.
2771+
if (parent != null) {
2772+
// In the native format whatever serializer is configured: a replay must rebuild the exact
2773+
// type the parent and Debouncer catch, which a custom serializer need not preserve.
2774+
var error = SerializationUtil.serializeError(e, SerializationUtil.NATIVE, null);
2775+
try {
2776+
systemDatabase.recordStepResult(
2777+
new StepResult(
2778+
parent.workflowId(),
2779+
parent.functionId(),
2780+
workflowName,
2781+
null,
2782+
error.serializedValue(),
2783+
null,
2784+
error.serialization()),
2785+
startTime,
2786+
parent.ownerXid());
2787+
} catch (RuntimeException recordFailure) {
2788+
recordFailure.addSuppressed(e);
2789+
throw recordFailure;
2790+
}
2791+
}
27222792
throw e;
27232793
} catch (DBOSWorkflowExecutionConflictException e) {
27242794
// A parent that no longer owns its workflow was refused: the fence working, not a failure.

0 commit comments

Comments
 (0)