refactor: streamline orchestration event handling and improve state management (#115)

Author: torosent

.github/workflows/dts-e2e-tests.yaml

@@ -1,7 +1,7 @@
 name: 🧪 DTS Emulator E2E Tests
 
 # This workflow runs E2E tests against the Durable Task Scheduler (DTS) emulator.
-# It mirrors the Python testing setup at durabletask-python for Azure-managed tests.
+# Tests are split across parallel jobs with separate task hubs for isolation.
 
 on:
   push:
@@ -20,6 +20,16 @@ jobs:
       fail-fast: false
       matrix:
         node-version: ["22.x", "24.x"]
+        test-group:
+          - name: "entity"
+            pattern: "test/e2e-azuremanaged/entity.spec.ts"
+          - name: "orchestration"
+            pattern: "test/e2e-azuremanaged/orchestration.spec.ts"
+          - name: "query-restart"
+            pattern: "test/e2e-azuremanaged/query-apis.spec.ts test/e2e-azuremanaged/restart.spec.ts"
+          - name: "retry-history-rewind"
+            pattern: "test/e2e-azuremanaged/retry-handler.spec.ts test/e2e-azuremanaged/retry-advanced.spec.ts test/e2e-azuremanaged/history.spec.ts test/e2e-azuremanaged/rewind.spec.ts"
+    name: "e2e (${{ matrix.test-group.name }}, node ${{ matrix.node-version }})"
     env:
       EMULATOR_VERSION: "latest"
     runs-on: ubuntu-latest
@@ -36,7 +46,7 @@ jobs:
           docker run --name dtsemulator -d -p 8080:8080 mcr.microsoft.com/dts/dts-emulator:$EMULATOR_VERSION
 
       - name: ⏳ Wait for container to be ready
-        run: sleep 10  # Adjust if your service needs more time to start
+        run: sleep 10
 
       - name: 🔧 Set environment variables
         run: |
@@ -50,7 +60,8 @@ jobs:
           registry-url: "https://registry.npmjs.org"
 
       - name: ⚙️ Install dependencies
-        run: npm install
+        run: npm ci
 
-      - name: ✅ Run E2E tests against DTS emulator
-        run: npm run test:e2e:azuremanaged:internal
+      - name: ✅ Run E2E tests — ${{ matrix.test-group.name }}
+        run: npx jest ${{ matrix.test-group.pattern }} --runInBand --detectOpenHandles
+        timeout-minutes: 15

.github/workflows/pr-validation.yaml

@@ -16,7 +16,7 @@ jobs:
     runs-on: ubuntu-latest
 
     env:
-      NODE_VER: 18.x
+      NODE_VER: 22.x
 
     steps:
       - name: 📥 Checkout code
@@ -29,7 +29,7 @@ jobs:
           registry-url: "https://registry.npmjs.org"
 
       - name: ⚙️ Install dependencies
-        run: npm install
+        run: npm ci
 
       - name: 🔍 Run linting
         run: npm run lint
@@ -56,7 +56,7 @@ jobs:
           registry-url: "https://registry.npmjs.org"
 
       - name: ⚙️ Install dependencies
-        run: npm install
+        run: npm ci
 
       # Install Go SDK for durabletask-go sidecar
       - name: 🔧 Install Go SDK

package-lock.json

@@ -16,7 +16,7 @@
     "test:e2e": "./scripts/test-e2e.sh",
     "test:e2e:internal": "jest tests/e2e --runInBand --detectOpenHandles",
     "test:e2e:one": "jest tests/e2e --runInBand --detectOpenHandles --testNamePattern",
-    "test:e2e:azuremanaged:internal": "jest test/e2e-azuremanaged --runInBand --detectOpenHandles",
+    "test:e2e:azuremanaged:internal": "jest test/e2e-azuremanaged --detectOpenHandles",
     "test:e2e:azuremanaged": "./scripts/test-e2e-azuremanaged.sh",
     "lint": "eslint .",
     "pretty": "prettier --list-different \"**/*.{ts,tsx,js,jsx,json,md}\"",

package.json

@@ -39,34 +39,17 @@ export function getCredentialFromAuthenticationType(
     }
 
     case "workloadidentity": {
-      const options: {
-        clientId?: string;
-        tenantId?: string;
-        tokenFilePath?: string;
-        additionallyAllowedTenants?: string[];
-      } = {};
-
       const clientId = connectionString.getClientId();
-      if (clientId) {
-        options.clientId = clientId;
-      }
-
       const tenantId = connectionString.getTenantId();
-      if (tenantId) {
-        options.tenantId = tenantId;
-      }
-
       const tokenFilePath = connectionString.getTokenFilePath();
-      if (tokenFilePath) {
-        options.tokenFilePath = tokenFilePath;
-      }
-
       const additionallyAllowedTenants = connectionString.getAdditionallyAllowedTenants();
-      if (additionallyAllowedTenants) {
-        options.additionallyAllowedTenants = additionallyAllowedTenants;
-      }
 
-      return new WorkloadIdentityCredential(options);
+      return new WorkloadIdentityCredential({
+        ...(clientId && { clientId }),
+        ...(tenantId && { tenantId }),
+        ...(tokenFilePath && { tokenFilePath }),
+        ...(additionallyAllowedTenants && { additionallyAllowedTenants }),
+      });
     }
 
     case "environment":

packages/durabletask-js-azuremanaged/src/credential-factory.ts

@@ -91,14 +91,14 @@ abstract class DurableTaskAzureManagedOptionsBase {
 
     // Add https:// prefix if no protocol is specified
     if (!endpoint.startsWith("http://") && !endpoint.startsWith("https://")) {
-      endpoint = "https://" + endpoint;
+      endpoint = `https://${endpoint}`;
     }
 
     try {
       const url = new URL(endpoint);
       let authority = url.hostname;
       if (url.port) {
-        authority += ":" + url.port;
+        authority = `${authority}:${url.port}`;
       }
       return authority;
     } catch {

packages/durabletask-js-azuremanaged/src/options.ts

@@ -30,7 +30,7 @@ export class Page<T> {
    * Returns true if there are more pages available.
    */
   get hasMoreResults(): boolean {
-    return this.continuationToken !== undefined && this.continuationToken !== "";
+    return !!this.continuationToken;
   }
 }
 

packages/durabletask-js/src/orchestration/page.ts

@@ -517,33 +517,27 @@ export function processActionsForTracing(
   }
 }
 
-/**
- * Emits a span for calling an entity from an orchestration (request/response).
- *
- * @param orchestrationSpan - The parent orchestration span.
- * @param operationName - The entity operation name.
- * @param targetInstanceId - The target entity instance ID.
- * @param taskId - The sequential task ID.
- */
-export function emitSpanForEntityCall(
+function emitSpanForEntityOperation(
   orchestrationSpan: Span,
   operationName: string,
+  taskType: string,
+  getSpanKind: (otel: any) => any,
   targetInstanceId?: string,
   taskId?: number,
 ): void {
   const otel = getOtelApi();
   const tracer = getTracer();
   if (!otel || !tracer) return;
 
-  const spanName = createSpanName(TaskType.CALL_ENTITY, operationName);
+  const spanName = createSpanName(taskType, operationName);
   const parentContext = otel.trace.setSpan(otel.context.active(), orchestrationSpan);
 
   const span = tracer.startSpan(
     spanName,
     {
-      kind: otel.SpanKind.CLIENT,
+      kind: getSpanKind(otel),
       attributes: {
-        [DurableTaskAttributes.TYPE]: TaskType.CALL_ENTITY,
+        [DurableTaskAttributes.TYPE]: taskType,
         [DurableTaskAttributes.ENTITY_OPERATION]: operationName,
         ...(targetInstanceId ? { [DurableTaskAttributes.ENTITY_INSTANCE_ID]: targetInstanceId } : {}),
         ...(taskId !== undefined ? { [DurableTaskAttributes.TASK_TASK_ID]: taskId } : {}),
@@ -555,6 +549,30 @@ export function emitSpanForEntityCall(
   span.end();
 }
 
+/**
+ * Emits a span for calling an entity from an orchestration (request/response).
+ *
+ * @param orchestrationSpan - The parent orchestration span.
+ * @param operationName - The entity operation name.
+ * @param targetInstanceId - The target entity instance ID.
+ * @param taskId - The sequential task ID.
+ */
+export function emitSpanForEntityCall(
+  orchestrationSpan: Span,
+  operationName: string,
+  targetInstanceId?: string,
+  taskId?: number,
+): void {
+  emitSpanForEntityOperation(
+    orchestrationSpan,
+    operationName,
+    TaskType.CALL_ENTITY,
+    (otel) => otel.SpanKind.CLIENT,
+    targetInstanceId,
+    taskId,
+  );
+}
+
 /**
  * Emits a span for signaling an entity from an orchestration (fire-and-forget).
  *
@@ -569,28 +587,14 @@ export function emitSpanForEntitySignal(
   targetInstanceId?: string,
   taskId?: number,
 ): void {
-  const otel = getOtelApi();
-  const tracer = getTracer();
-  if (!otel || !tracer) return;
-
-  const spanName = createSpanName(TaskType.SIGNAL_ENTITY, operationName);
-  const parentContext = otel.trace.setSpan(otel.context.active(), orchestrationSpan);
-
-  const span = tracer.startSpan(
-    spanName,
-    {
-      kind: otel.SpanKind.PRODUCER,
-      attributes: {
-        [DurableTaskAttributes.TYPE]: TaskType.SIGNAL_ENTITY,
-        [DurableTaskAttributes.ENTITY_OPERATION]: operationName,
-        ...(targetInstanceId ? { [DurableTaskAttributes.ENTITY_INSTANCE_ID]: targetInstanceId } : {}),
-        ...(taskId !== undefined ? { [DurableTaskAttributes.TASK_TASK_ID]: taskId } : {}),
-      },
-    },
-    parentContext,
+  emitSpanForEntityOperation(
+    orchestrationSpan,
+    operationName,
+    TaskType.SIGNAL_ENTITY,
+    (otel) => otel.SpanKind.PRODUCER,
+    targetInstanceId,
+    taskId,
   );
-
-  span.end();
 }
 
 /**