Skip to content
Merged
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
148 changes: 145 additions & 3 deletions packages/durabletask-js/src/worker/orchestration-executor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,17 @@ export class OrchestrationExecutor {
// Event names are case-insensitive
const eventName = event.getEventraised()?.getName()?.toLowerCase();

// On the classic (Azure Storage / DurableTask.Core) backend, an entity call's response is
// delivered as an EVENTRAISED whose name equals the entity call's requestId (a lowercase GUID)
// and whose input is a DTFx ResponseMessage JSON, rather than as an ENTITYOPERATIONCOMPLETED.
// Route it back to the pending entity call so the callEntity task resolves/rejects. On DTS this
// branch never fires (responses arrive as ENTITYOPERATIONCOMPLETED and the pending call is
// removed by handleEntityOperationCompleted before any EVENTRAISED could match).
if (eventName && ctx._entityFeature.pendingEntityCalls.has(eventName)) {
await this.handleEntityResponseFromEventRaised(ctx, eventName, event);
return;
}

if (!ctx._isReplaying) {
WorkerLogs.orchestrationEventRaised(this._logger, ctx._instanceId, eventName!);
}
Expand Down Expand Up @@ -462,16 +473,147 @@ export class OrchestrationExecutor {
}
}

/**
* Handles an entity operation response that arrived as an EVENTRAISED event on the classic
* (Azure Storage / DurableTask.Core) backend. In that code path the Durable Functions gRPC shim
* wraps entity responses in a DTFx ResponseMessage JSON keyed by the call's requestId, rather than
* emitting an ENTITYOPERATIONCOMPLETED proto event. This decodes the wrapper and completes (or
* fails) the pending callEntity task. Mirrors the Java SDK's handleEntityResponseFromEventRaised.
*/
private async handleEntityResponseFromEventRaised(
ctx: RuntimeOrchestrationContext,
requestId: string,
event: pb.HistoryEvent,
): Promise<void> {
const pendingCall = ctx._entityFeature.pendingEntityCalls.get(requestId);
if (!pendingCall) {
return;
}

// Remove from pending calls and recover any critical-section lock before completing.
ctx._entityFeature.pendingEntityCalls.delete(requestId);
ctx._entityFeature.recoverLockAfterCall(pendingCall.entityId);

const decoded = this.decodeEntityResponseMessage(event.getEventraised()?.getInput()?.getValue());

if (decoded.isFailure) {
const exception = new EntityOperationFailedException(pendingCall.entityId, pendingCall.operationName, {
errorType: decoded.errorType,
errorMessage: decoded.errorMessage,
stackTrace: decoded.stackTrace,
});
Comment thread
Copilot marked this conversation as resolved.
pendingCall.task.failWithError(exception);
} else {
pendingCall.task.complete(decoded.result);
}

await ctx.resume();
}

/**
* Decodes a DTFx ResponseMessage JSON wrapper (as delivered via EVENTRAISED on the classic
* backend) into either a success result or a failure. Mirrors the Java SDK's decoding, with one
* deliberate enhancement (see below).
*
* The wrapper shape is:
* - `result` — the serialized operation result (double-encoded string; may be null/absent).
* - `exceptionType` — present ⇒ failure. Misleading name: it carries the error message string
* (the C# property is ErrorMessage but its [DataMember(Name = "exceptionType")]
* overrides the JSON key). Omitted when null.
* - `failureDetails` — optional structured `{ ErrorType, ErrorMessage, StackTrace, ... }` (PascalCase).
*
* NOTE (intentionally beyond the Java SDK): the Java SDK builds `new FailureDetails(errorType,
* errorMessage, null, false)`, dropping the stack trace. Over gRPC the classic backend uses
* DurableTask.Core native entities, whose ResponseMessage.FailureDetails carries a real StackTrace,
* so we propagate it into `EntityOperationFailedException.failureDetails.stackTrace`. Do not "fix"
* this back to null for Java parity — the extra fidelity is the point.
*/
private decodeEntityResponseMessage(
rawInput: string | undefined,
):
| { isFailure: false; result: any }
| { isFailure: true; errorType: string; errorMessage: string; stackTrace?: string } {
if (rawInput === undefined || rawInput === "") {
return { isFailure: false, result: undefined };
}

let parsed: any;
try {
parsed = JSON.parse(rawInput);
} catch {
// Not JSON at all — treat the raw string as the result value.
return { isFailure: false, result: rawInput };
}

const isObject = parsed !== null && typeof parsed === "object" && !Array.isArray(parsed);
if (!isObject || !Object.prototype.hasOwnProperty.call(parsed, "result")) {
// Not a recognized ResponseMessage wrapper — the parsed value itself is the result.
// (Mirrors Java's fallback to direct deserialization for a possible future raw-result format.)
return { isFailure: false, result: parsed };
}

const exceptionType = parsed.exceptionType;
const failureDetails = parsed.failureDetails;
const hasExceptionType = exceptionType !== undefined && exceptionType !== null;
const hasFailureDetails = failureDetails !== undefined && failureDetails !== null;

if (hasExceptionType || hasFailureDetails) {
// The "exceptionType" JSON field actually carries the error message (misleading name).
let errorMessage = hasExceptionType ? String(exceptionType) : "Entity operation failed";
let errorType = "unknown";
let stackTrace: string | undefined;

if (hasFailureDetails && typeof failureDetails === "object") {
if (failureDetails.ErrorType !== undefined && failureDetails.ErrorType !== null) {
errorType = String(failureDetails.ErrorType);
}
if (failureDetails.ErrorMessage !== undefined && failureDetails.ErrorMessage !== null) {
errorMessage = String(failureDetails.ErrorMessage);
}
if (failureDetails.StackTrace !== undefined && failureDetails.StackTrace !== null) {
stackTrace = String(failureDetails.StackTrace);
}
}

return { isFailure: true, errorType, errorMessage, stackTrace };
}
Comment thread
Copilot marked this conversation as resolved.

// Success — extract the inner (double-encoded) result value.
const resultNode = parsed.result;
if (resultNode === null || resultNode === undefined) {
return { isFailure: false, result: undefined };
}

const innerResult = typeof resultNode === "string" ? resultNode : JSON.stringify(resultNode);
try {
return { isFailure: false, result: JSON.parse(innerResult) };
} catch {
// Defensive: if the inner payload isn't valid JSON, use the raw string.
return { isFailure: false, result: innerResult };
}
}

private async handleEventSent(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise<void> {
// This history event confirms that a sendEvent action was successfully processed by the sidecar.
const eventId = event.getEventid();
const action = ctx._pendingActions[eventId];

const isSendEventAction = action?.hasSendevent();

if (!action) {
throw getNonDeterminismError(eventId, getName(ctx.sendEvent));
} else if (!isSendEventAction) {
}

// On the classic (Azure Storage / DurableTask.Core) backend, entity operations do not use
// the DTS entity protocol. The Durable Functions gRPC shim translates a sendEntityMessage
// action (call / signal / lock / unlock) into a classic send-event, so its confirmation
// arrives here as an EVENTSENT rather than as ENTITYOPERATIONCALLED/SIGNALED. Treat that as
// a valid confirmation: remove the pending action so it isn't re-sent. The matching entity
// response (for a call) is delivered later as an EVENTRAISED and routed by handleEventRaised.
if (action.hasSendentitymessage()) {
delete ctx._pendingActions[eventId];
return;
}

if (!action.hasSendevent()) {
const expectedMethodName = getName(ctx.sendEvent);
throw new NonDeterminismError(
`A previous execution called ${expectedMethodName} with ID=${eventId}, but the current execution is instead trying to call a different method as part of rebuilding its history.`,
Expand Down
Loading
Loading