feat(testing): Use temporal.download for downloading test server (#864)

Author: bergundy

.github/workflows/ci.yml

@@ -30,6 +30,9 @@ jobs:
       - uses: actions/setup-node@v1
         with:
           node-version: ${{ matrix.node }}
+      - uses: Swatinem/rust-cache@v1
+        with:
+          working-directory: packages/core-bridge
       # Don't build during install phase since we're going to explicitly build
       - run: npm ci --ignore-scripts
       - run: npm run build
@@ -162,6 +165,9 @@ jobs:
       - name: Install gcc-aarch64-linux-gnu
         run: sudo apt update && sudo apt install -y gcc-aarch64-linux-gnu
         if: ${{ matrix.target == 'aarch64-unknown-linux-gnu' }}
+      - uses: Swatinem/rust-cache@v1
+        with:
+          working-directory: packages/core-bridge
       # Don't build during install phase since we're going to explicitly build
       - run: npm ci --ignore-scripts
       - name: Compile all non-rust code

.github/workflows/stress.yml

@@ -30,6 +30,9 @@ jobs:
       - uses: actions/setup-node@v1
         with:
           node-version: ${{ matrix.node }}
+      - uses: Swatinem/rust-cache@v1
+        with:
+          working-directory: packages/core-bridge
       # Don't build during install phase since we're going to explicitly build
       - run: npm ci --ignore-scripts
       - run: npm run build

packages/client/src/workflow-options.ts

@@ -64,30 +64,3 @@ export interface WorkflowSignalWithStartOptionsWithArgs<SignalArgs extends any[]
    */
   signalArgs: SignalArgs;
 }
-
-// export interface WorkflowOptionsWithDefaults<T extends Workflow> extends CommonWorkflowOptionsWithDefaults<T> {
-//   /**
-//    * If set to true, instructs the client to follow the chain of execution before returning a Workflow's result.
-//    *
-//    * Workflow execution is chained if the Workflow has a cron schedule or continues-as-new or configured to retry
-//    * after failure or timeout.
-//    *
-//    * @default true
-//    */
-//   followRuns: boolean;
-// }
-//
-// /**
-//  * Adds default values to `workflowId` and `workflowIdReusePolicy` to given workflow options.
-//  */
-// export function addDefaults<T extends Workflow>(
-//   opts: WithWorkflowArgs<T, WorkflowOptions>
-// ): WorkflowOptionsWithDefaults<T> {
-//   const { workflowId, args, ...rest } = opts;
-//   return {
-//     followRuns: true,
-//     args: args ?? [],
-//     workflowId: workflowId ?? uuid4(),
-//     ...rest,
-//   };
-// }

packages/common/src/failure.ts

@@ -16,6 +16,7 @@ export enum TimeoutType {
 }
 
 checkExtends<temporal.api.enums.v1.TimeoutType, TimeoutType>();
+checkExtends<TimeoutType, temporal.api.enums.v1.TimeoutType>();
 
 // Avoid importing the proto implementation to reduce workflow bundle size
 // Copied from temporal.api.enums.v1.RetryState
@@ -31,6 +32,7 @@ export enum RetryState {
 }
 
 checkExtends<temporal.api.enums.v1.RetryState, RetryState>();
+checkExtends<RetryState, temporal.api.enums.v1.RetryState>();
 
 export type WorkflowExecution = temporal.api.common.v1.IWorkflowExecution;
 

packages/internal-workflow-common/src/activity-options.ts

@@ -11,6 +11,7 @@ export enum ActivityCancellationType {
 }
 
 checkExtends<coresdk.workflow_commands.ActivityCancellationType, ActivityCancellationType>();
+checkExtends<ActivityCancellationType, coresdk.workflow_commands.ActivityCancellationType>();
 
 /**
  * Options for remote activity invocation

packages/internal-workflow-common/src/workflow-options.ts

@@ -36,9 +36,15 @@ export enum WorkflowIdReusePolicy {
    * The Workflow cannot be started.
    */
   WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE = 3,
+
+  /**
+   * Terminate the current workflow if one is already running.
+   */
+  WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING = 4,
 }
 
 checkExtends<temporal.api.enums.v1.WorkflowIdReusePolicy, WorkflowIdReusePolicy>();
+checkExtends<WorkflowIdReusePolicy, temporal.api.enums.v1.WorkflowIdReusePolicy>();
 
 export interface BaseWorkflowOptions {
   /**

packages/test/src/load/worker.ts

@@ -157,8 +157,13 @@ async function main() {
     });
 
     await withOptionalStatusServer(worker, statusPort, async () => {
-      await worker.run();
-      await connection.close();
+      const interval = setInterval(() => logger.info('worker status', worker.getStatus()), 5000);
+      try {
+        await worker.run();
+      } finally {
+        clearInterval(interval);
+        await connection.close();
+      }
     });
   });
 }

packages/testing/scripts/common.mjs

@@ -2,10 +2,29 @@ import stream from 'node:stream';
 import util from 'node:util';
 import zlib from 'node:zlib';
 import fs from 'node:fs';
+import os from 'node:os';
 import got from 'got';
 import tar from 'tar-stream';
 import unzipper from 'unzipper';
-import { outputPath, systemArch, systemPlatform } from './common.mjs';
+import { URL, fileURLToPath } from 'node:url';
+
+const platformMapping = { darwin: 'darwin', linux: 'linux', win32: 'windows' };
+const archAlias = { x64: 'amd64', arm64: 'arm64' };
+
+const platform = platformMapping[os.platform()];
+if (!platform) {
+  throw new Error(`Unsupported platform ${os.platform()}`);
+}
+
+const arch = archAlias[os.arch()];
+if (!arch) {
+  throw new Error(`Unsupported architecture ${os.arch()}`);
+}
+
+const ext = platform === 'windows' ? '.exe' : '';
+const outputPath = fileURLToPath(new URL(`../test-server${ext}`, import.meta.url));
+const pkgPath = fileURLToPath(new URL(`../package.json`, import.meta.url));
+const pkg = JSON.parse(fs.readFileSync(pkgPath));
 
 try {
   if (fs.statSync(outputPath).isFile) {
@@ -20,72 +39,44 @@ try {
 
 const pipeline = util.promisify(stream.pipeline);
 
-const defaultHeaders = {
-  'User-Agent': '@temporalio/testing installer',
-};
-
-const { GITHUB_TOKEN } = process.env;
-
-if (GITHUB_TOKEN) {
-  console.log(`Using GITHUB_TOKEN`);
-  defaultHeaders['Authorization'] = `Bearer ${GITHUB_TOKEN}`;
-}
-
-const latestReleaseRes = await got('https://api.github.com/repos/temporalio/sdk-java/releases/latest', {
+const defaultOptions = {
   headers: {
-    ...defaultHeaders,
-    Accept: 'application/vnd.github.v3+json',
+    'User-Agent': '@temporalio/testing installer',
   },
-}).json();
+};
 
-function findTestServerAsset(assets) {
-  for (const asset of assets) {
-    const m = asset.name.match(/^temporal-test-server_[^_]+_([^_]+)_([^.]+)\.(?:zip|tar.gz)$/);
-    if (m) {
-      const [_, assetPlatform, _assetArch] = m;
-      if (assetPlatform === systemPlatform) {
-        // TODO: assetArch === systemArch (no arm builds for test server yet)
-        return asset;
-      }
-    }
-  }
-  throw new Error(`No prebuilt test server for ${systemPlatform}-${systemArch}`);
-}
+const lookupOptions = {
+  ...defaultOptions,
+  searchParams: { 'sdk-name': 'typescript', 'sdk-version': pkg.version, platform, arch },
+};
 
-const asset = findTestServerAsset(latestReleaseRes.assets);
-console.log('Downloading test server', { asset: asset.name, outputPath });
+const lookupUrl = 'https://temporal.download/temporal-test-server/default';
+console.log('Looking up default test server', { lookupUrl, options: lookupOptions });
+const { archiveUrl, fileToExtract } = await got(
+  'https://temporal.download/temporal-test-server/default',
+  lookupOptions
+).json();
 
-if (asset.content_type === 'application/x-gzip' || asset.content_type === 'application/x-gtar') {
+console.log('Downloading test server', { archiveUrl, fileToExtract, outputPath });
+if (archiveUrl.endsWith('.tar.gz')) {
   const extract = tar.extract();
-  extract.on('entry', (_headers, stream, next) => {
-    stream.pipe(fs.createWriteStream(outputPath));
+  extract.on('entry', (headers, stream, next) => {
+    if (headers.name === fileToExtract) {
+      stream.pipe(fs.createWriteStream(outputPath));
+    }
     next();
   });
-  await pipeline(
-    got.stream(asset.browser_download_url, {
-      headers: {
-        ...defaultHeaders,
-      },
-    }),
-    zlib.createGunzip(),
-    extract
-  );
+  await pipeline(got.stream(archiveUrl, defaultOptions), zlib.createGunzip(), extract);
   await fs.promises.chmod(outputPath, 0o755);
-} else if (asset.content_type === 'application/zip') {
+} else if (archiveUrl.endsWith('.zip')) {
   got
-    .stream(asset.browser_download_url, {
-      headers: {
-        ...defaultHeaders,
-      },
-    })
+    .stream(archiveUrl, defaultOptions)
     .pipe(unzipper.Parse())
     .on('entry', (entry) => {
-      if (entry.type === 'File') {
+      if (entry.type === 'File' && entry.path === fileToExtract) {
         entry.pipe(fs.createWriteStream(outputPath));
       } else {
         entry.autodrain();
       }
     });
-} else {
-  throw new Error(`Unexpected content type for Test server download: ${asset.content_type}`);
 }

packages/testing/scripts/download-test-server.mjs

@@ -223,6 +223,7 @@ export enum ChildWorkflowCancellationType {
 }
 
 checkExtends<coresdk.child_workflow.ChildWorkflowCancellationType, ChildWorkflowCancellationType>();
+checkExtends<ChildWorkflowCancellationType, coresdk.child_workflow.ChildWorkflowCancellationType>();
 
 /**
  * How a Child Workflow reacts to the Parent Workflow reaching a Closed state.
@@ -254,6 +255,7 @@ export enum ParentClosePolicy {
 }
 
 checkExtends<coresdk.child_workflow.ParentClosePolicy, ParentClosePolicy>();
+checkExtends<ParentClosePolicy, coresdk.child_workflow.ParentClosePolicy>();
 
 export interface ChildWorkflowOptions extends CommonWorkflowOptions {
   /**