Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 26 additions & 20 deletions packages/workflows/src/context.ts
Original file line number Diff line number Diff line change
Expand Up @@ -565,29 +565,20 @@ export class WorkflowContextImpl implements WorkflowContextInterface {
}

/**
* Validate that all expected entries in the branch were visited.
* Throws HistoryDivergedError if there are unvisited entries.
* Validate that every direct entry in this scope was visited. Nested scopes
* validate their own entries when they execute.
*/
validateComplete(): void {
const prefix = locationToKey(this.storage, this.currentLocation);

for (const key of this.storage.history.entries.keys()) {
// Check if this key is under our current location prefix
// Handle root prefix (empty string) specially - all keys are under root
const isUnderPrefix =
prefix === ""
? true // Root: all keys are children
: key.startsWith(`${prefix}/`) || key === prefix;
for (const [key, entry] of this.storage.history.entries) {
const isDirectChild =
entry.location.length === this.currentLocation.length + 1 &&
isLocationPrefix(this.currentLocation, entry.location);

if (isUnderPrefix) {
if (!this.visitedKeys.has(key)) {
// Entry exists in history but wasn't visited
// This means workflow code may have changed
throw new HistoryDivergedError(
`Entry "${key}" exists in history but was not visited. ` +
`Workflow code may have changed. Use ctx.removed() to handle migrations.`,
);
}
if (isDirectChild && !this.visitedKeys.has(key)) {
throw new HistoryDivergedError(
`Entry "${key}" exists in history but was not visited. ` +
"Workflow code may have changed. Use ctx.removed() to handle migrations.",
);
}
}
}
Expand Down Expand Up @@ -791,6 +782,13 @@ export class WorkflowContextImpl implements WorkflowContextInterface {

// Check for duplicate name in current execution
this.checkDuplicateName(config.name);
const parentKey = locationToKey(this.storage, this.currentLocation);
const candidateKey = parentKey
? `${parentKey}/${config.name}`
: config.name;
if (!this.storage.history.entries.has(candidateKey)) {
this.validateComplete();
}

const location = appendName(
this.storage,
Expand Down Expand Up @@ -2558,6 +2556,14 @@ export class WorkflowContextImpl implements WorkflowContextInterface {

// Mark this entry as visited for validateComplete
this.markVisited(key);
if (originalType === "message") {
const generatedKeyPrefix = `${key}:`;
for (const existingKey of this.storage.history.entries.keys()) {
if (existingKey.startsWith(generatedKeyPrefix)) {
this.markVisited(existingKey);
}
}
}

this.stopRollbackIfMissing(existing);

Expand Down
7 changes: 6 additions & 1 deletion packages/workflows/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,7 @@ import {
import {
CriticalError,
EvictedError,
HistoryDivergedError,
MessageWaitError,
RollbackCheckpointError,
RollbackError,
Expand Down Expand Up @@ -993,6 +994,7 @@ async function executeWorkflow<TInput, TOutput>(

try {
const output = await workflowFn(ctx, effectiveInput);
ctx.validateComplete();

storage.state = "completed";
storage.output = output;
Expand Down Expand Up @@ -1035,7 +1037,10 @@ async function executeWorkflow<TInput, TOutput>(
);
}

if (error instanceof RollbackCheckpointError) {
if (
error instanceof HistoryDivergedError ||
error instanceof RollbackCheckpointError
) {
await setFailedState(storage, driver, error, historyNotifier);
if (onError && !isErrorReported(error)) {
await notifyError(onError, logger, {
Expand Down
23 changes: 23 additions & 0 deletions packages/workflows/tests/steps.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import {
EntryInProgressError,
HistoryDivergedError,
InMemoryDriver,
loadStorage,
RollbackError,
runWorkflow,
StepExhaustedError,
Expand Down Expand Up @@ -74,6 +75,28 @@ for (const mode of modes) {
.result;
expect(callCount).toBe(1);
});
it("should reject a renamed root step on replay", async () => {
const originalWorkflow = async (ctx: WorkflowContextInterface) => {
return await ctx.step("original-step-name", async () => "original");
};
const renamedWorkflow = async (ctx: WorkflowContextInterface) => {
return await ctx.step("renamed-step-name", async () => "changed");
};

await runWorkflow("wf-1", originalWorkflow, undefined, driver, { mode })
.result;

await expect(
runWorkflow("wf-1", renamedWorkflow, undefined, driver, { mode })
.result,
).rejects.toThrow(HistoryDivergedError);
const storage = await loadStorage(driver);
expect(storage.nameRegistry).toEqual(["original-step-name"]);
expect([...storage.history.entries.keys()]).toEqual([
"original-step-name",
]);
expect(storage.state).not.toBe("completed");
});

it("should replay void step on restart", async () => {
let callCount = 0;
Expand Down