DX-1829: Add period and rate to flow control and deprecate ratePerSecond

package.json

@@ -103,7 +103,7 @@
   },
   "dependencies": {
     "@ai-sdk/openai": "^1.2.1",
-    "@upstash/qstash": "^2.7.22",
+    "@upstash/qstash": "^2.8.1",
     "ai": "^4.1.54",
     "zod": "^3.24.1"
   },

src/agents/adapters.test.ts

@@ -15,7 +15,7 @@ describe("wrapTools", () => {
     new WorkflowContext({
       headers: new Headers({}) as Headers,
       initialPayload: "mock",
-      qstashClient: new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token }),
+      qstashClient: new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token, enableTelemetry: false }),
       steps: [],
       url: WORKFLOW_ENDPOINT,
       workflowRunId,

src/context/auto-executor.test.ts

@@ -85,7 +85,7 @@ describe("auto-executor", () => {
 
   const getContext = (steps: Step[]) => {
     return new SpyWorkflowContext({
-      qstashClient: new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token }),
+      qstashClient: new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token, enableTelemetry: false }),
       workflowRunId,
       initialPayload,
       headers: new Headers({}) as Headers,

src/context/context.test.ts

@@ -16,7 +16,11 @@ import { upstash } from "@upstash/qstash";
 
 describe("context tests", () => {
   const token = nanoid();
-  const qstashClient = new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token });
+  const qstashClient = new Client({
+    baseUrl: MOCK_QSTASH_SERVER_URL,
+    token,
+    enableTelemetry: false,
+  });
   test("should raise when there are nested steps (with run)", () => {
     const context = new WorkflowContext({
       qstashClient,

src/qstash/headers.ts

@@ -251,11 +251,14 @@ function addPrefixToHeaders(headers: Record<string, string>, prefix: string) {
 
 export const prepareFlowControl = (flowControl: FlowControl) => {
   const parallelism = flowControl.parallelism?.toString();
-  const rate = flowControl.ratePerSecond?.toString();
+  const rate = (flowControl.rate ?? flowControl.ratePerSecond)?.toString();
+  const period =
+    typeof flowControl.period === "number" ? `${flowControl.period}s` : flowControl.period;
 
   const controlValue = [
     parallelism ? `parallelism=${parallelism}` : undefined,
     rate ? `rate=${rate}` : undefined,
+    period ? `period=${period}` : undefined,
   ].filter(Boolean);
 
   if (controlValue.length === 0) {

src/serve/authorization.test.ts

@@ -10,7 +10,11 @@ import { DisabledWorkflowContext } from "./authorization";
 
 describe("disabled workflow context", () => {
   const token = nanoid();
-  const qstashClient = new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token });
+  const qstashClient = new Client({
+    baseUrl: MOCK_QSTASH_SERVER_URL,
+    token,
+    enableTelemetry: false,
+  });
   const disabledContext = new DisabledWorkflowContext({
     qstashClient,
     workflowRunId: "wfr-foo",

src/serve/serve-many.test.ts

@@ -41,7 +41,11 @@ describe("serveMany", () => {
 
   describe("serve tests", () => {
     const token = nanoid();
-    const qstashClient = new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token });
+    const qstashClient = new Client({
+      baseUrl: MOCK_QSTASH_SERVER_URL,
+      token,
+      enableTelemetry: false,
+    });
 
     const workflowOne = createWorkflow(
       async (context: WorkflowContext<number>) => {
@@ -160,7 +164,7 @@ describe("serveMany", () => {
               "Upstash-Forward-Upstash-Workflow-Sdk-Version": ["1"],
               "Upstash-Telemetry-Framework": ["nextjs"],
               "Upstash-Telemetry-Runtime": ["[email protected]"],
-              "Upstash-Telemetry-Sdk": ["@upstash/[email protected].7"],
+              "Upstash-Telemetry-Sdk": ["@upstash/[email protected].13"],
               "Upstash-Workflow-Init": ["false"],
               "Upstash-Workflow-RunId": ["wfr_id"],
               "Upstash-Workflow-Runid": ["wfr_id"],
@@ -215,7 +219,7 @@ describe("serveMany", () => {
               "Upstash-Forward-Upstash-Workflow-Sdk-Version": ["1"],
               "Upstash-Telemetry-Framework": ["nextjs"],
               "Upstash-Telemetry-Runtime": ["[email protected]"],
-              "Upstash-Telemetry-Sdk": ["@upstash/[email protected].7"],
+              "Upstash-Telemetry-Sdk": ["@upstash/[email protected].13"],
               "Upstash-Workflow-Init": ["false"],
               "Upstash-Workflow-RunId": ["wfr_id"],
               "Upstash-Workflow-Runid": ["wfr_id"],
@@ -289,7 +293,7 @@ describe("serveMany", () => {
                 "upstash-retries": "0",
                 "upstash-telemetry-framework": "nextjs",
                 "upstash-telemetry-runtime": "[email protected]",
-                "upstash-telemetry-sdk": "@upstash/[email protected].7",
+                "upstash-telemetry-sdk": "@upstash/[email protected].13",
                 "upstash-workflow-calltype": "toCallback",
                 "upstash-workflow-init": "false",
                 "upstash-workflow-runid": "wfr_id",

src/serve/serve.test.ts

@@ -1,6 +1,6 @@
 /* eslint-disable @typescript-eslint/no-unused-vars */
 /* eslint-disable @typescript-eslint/require-await */
-import { describe, expect, spyOn, test } from "bun:test";
+import { describe, expect, jest, spyOn, test } from "bun:test";
 import { serve } from ".";
 import {
   driveWorkflow,
@@ -32,7 +32,7 @@ const someWork = (input: string) => {
 const workflowRunId = `wfr${nanoid()}`;
 const token = nanoid();
 
-const qstashClient = new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token });
+const qstashClient = new Client({ baseUrl: MOCK_QSTASH_SERVER_URL, token, enableTelemetry: false });
 
 describe("serve", () => {
   test("should send create workflow request in initial request", async () => {
@@ -409,7 +409,7 @@ describe("serve", () => {
                 "upstash-workflow-url": WORKFLOW_ENDPOINT,
                 "upstash-telemetry-framework": "unknown",
                 "upstash-telemetry-runtime": "unknown",
-                "upstash-telemetry-sdk": "@upstash/[email protected].7",
+                "upstash-telemetry-sdk": "@upstash/[email protected].13",
               },
               body: '{"stepId":3,"stepName":"step 3","stepType":"Run","out":"\\"combined results: result 1,result 2\\"","concurrent":1}',
             },
@@ -458,7 +458,7 @@ describe("serve", () => {
                 "upstash-workflow-url": WORKFLOW_ENDPOINT,
                 "upstash-telemetry-framework": "unknown",
                 "upstash-telemetry-runtime": "unknown",
-                "upstash-telemetry-sdk": "@upstash/[email protected].7",
+                "upstash-telemetry-sdk": "@upstash/[email protected].13",
               },
               body: '{"stepId":1,"stepName":"sleep-step","stepType":"SleepFor","sleepFor":1,"concurrent":1}',
             },
@@ -511,7 +511,7 @@ describe("serve", () => {
                 "upstash-failure-callback-workflow-url": "https://requestcatcher.com/api",
                 "upstash-telemetry-framework": "unknown",
                 "upstash-telemetry-runtime": "unknown",
-                "upstash-telemetry-sdk": "@upstash/[email protected].7",
+                "upstash-telemetry-sdk": "@upstash/[email protected].13",
               },
               body: '{"stepId":1,"stepName":"sleep-step","stepType":"SleepFor","sleepFor":1,"concurrent":1}',
             },
@@ -566,7 +566,7 @@ describe("serve", () => {
                 "upstash-failure-callback-workflow-url": "https://requestcatcher.com/api",
                 "upstash-telemetry-framework": "unknown",
                 "upstash-telemetry-runtime": "unknown",
-                "upstash-telemetry-sdk": "@upstash/[email protected].7",
+                "upstash-telemetry-sdk": "@upstash/[email protected].13",
               },
               body: '{"stepId":1,"stepName":"sleep-step","stepType":"SleepFor","sleepFor":1,"concurrent":1}',
             },
@@ -661,7 +661,7 @@ describe("serve", () => {
                 "upstash-retries": "4",
                 "upstash-telemetry-framework": "unknown",
                 "upstash-telemetry-runtime": "unknown",
-                "upstash-telemetry-sdk": "@upstash/[email protected].7",
+                "upstash-telemetry-sdk": "@upstash/[email protected].13",
                 "upstash-timeout": "10",
                 "upstash-workflow-calltype": "toCallback",
                 "upstash-workflow-init": "false",
@@ -875,7 +875,7 @@ describe("serve", () => {
             "Upstash-Workflow-Url": [WORKFLOW_ENDPOINT],
             "Upstash-Telemetry-Framework": ["unknown"],
             "Upstash-Telemetry-Runtime": ["unknown"],
-            "Upstash-Telemetry-Sdk": ["@upstash/[email protected].7"],
+            "Upstash-Telemetry-Sdk": ["@upstash/[email protected].13"],
           },
           timeoutUrl: WORKFLOW_ENDPOINT,
           url: WORKFLOW_ENDPOINT,
@@ -1017,6 +1017,10 @@ describe("serve", () => {
 
   describe("incorrect url will throw", () => {
     const qstashClient = new Client({ token: process.env.QSTASH_TOKEN! });
+    qstashClient.batch = jest
+      .fn()
+      .mockReturnValue([{ deduplicatedId: false, messageId: "some-message-id" }]);
+    qstashClient.publish = jest.fn({ deduplicatedId: false, messageId: "some-message-id" });
     const client = new WorkflowClient({ token: process.env.QSTASH_TOKEN! });
 
     test("allow http://", async () => {
@@ -1111,6 +1115,7 @@ describe("serve", () => {
       headers: {
         [header]: headerValue,
       },
+      enableTelemetry: false,
     });
 
     const { handler: endpoint } = serve(
@@ -1158,7 +1163,7 @@ describe("serve", () => {
               "upstash-forward-test-header": headerValue,
               "upstash-telemetry-framework": "unknown",
               "upstash-telemetry-runtime": "unknown",
-              "upstash-telemetry-sdk": "@upstash/[email protected].7",
+              "upstash-telemetry-sdk": "@upstash/[email protected].13",
             },
             body: '{"stepId":1,"stepName":"sleep-step","stepType":"SleepFor","sleepFor":1,"concurrent":1}',
           },

src/workflow-requests.test.ts

@@ -525,8 +525,9 @@ describe("Workflow Requests", () => {
         undefined,
         {
           key: "call-flow-key",
-          ratePerSecond: 5,
+          rate: 5,
           parallelism: 6,
+          period: 30,
         }
       );
       const { headers } = lazyStep.getHeaders({
@@ -539,8 +540,9 @@ describe("Workflow Requests", () => {
           initialPayload: undefined,
           flowControl: {
             key: "regular-flow-key",
-            ratePerSecond: 3,
+            rate: 3,
             parallelism: 4,
+            period: "1m",
           },
         }),
         invokeCount: 3,
@@ -572,9 +574,9 @@ describe("Workflow Requests", () => {
         "content-type": "application/json",
         // flow control:
         "Upstash-Callback-Flow-Control-Key": "regular-flow-key",
-        "Upstash-Callback-Flow-Control-Value": "parallelism=4, rate=3",
+        "Upstash-Callback-Flow-Control-Value": "parallelism=4, rate=3, period=1m",
         "Upstash-Flow-Control-Key": "call-flow-key",
-        "Upstash-Flow-Control-Value": "parallelism=6, rate=5",
+        "Upstash-Flow-Control-Value": "parallelism=6, rate=5, period=30s",
       });
     });