Skip to content

Commit 76aa4a4

Browse files
adrien2padrien2p
authored
fix(): workflows concurrency (#13645)
1 parent ca334b7 commit 76aa4a4

12 files changed

Lines changed: 196 additions & 86 deletions

File tree

‎.changeset/fair-mirrors-enjoy.md‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
---
2+
"@medusajs/medusa": patch
3+
"@medusajs/workflow-engine-inmemory": patch
4+
"@medusajs/workflow-engine-redis": patch
5+
"@medusajs/core-flows": patch
6+
"@medusajs/orchestration": patch
7+
"@medusajs/workflows-sdk": patch
8+
---
9+
10+
Fix/workflows concurrency

‎integration-tests/http/__tests__/cart/store/cart.spec.ts‎

Lines changed: 84 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -9,10 +9,15 @@ import {
99
PromotionStatus,
1010
PromotionType,
1111
} from "@medusajs/utils"
12-
import { createAdminUser, generatePublishableKey, generateStoreHeaders, } from "../../../../helpers/create-admin-user"
12+
import {
13+
createAdminUser,
14+
generatePublishableKey,
15+
generateStoreHeaders,
16+
} from "../../../../helpers/create-admin-user"
1317
import { setupTaxStructure } from "../../../../modules/__tests__/fixtures"
1418
import { createAuthenticatedCustomer } from "../../../../modules/helpers/create-authenticated-customer"
1519
import { medusaTshirtProduct } from "../../../__fixtures__/product"
20+
import { setTimeout } from "timers/promises"
1621

1722
jest.setTimeout(100000)
1823

@@ -150,10 +155,9 @@ medusaIntegrationTestRunner({
150155

151156
describe("GET /store/carts/[id]", () => {
152157
it("should return 404 when trying to fetch a cart that does not exist", async () => {
153-
const response = await api.get(
154-
`/store/carts/fake`,
155-
storeHeadersWithCustomer
156-
).catch((e) => e)
158+
const response = await api
159+
.get(`/store/carts/fake`, storeHeadersWithCustomer)
160+
.catch((e) => e)
157161

158162
expect(response.response.status).toEqual(404)
159163
})
@@ -1868,6 +1872,80 @@ medusaIntegrationTestRunner({
18681872
)
18691873
})
18701874

1875+
it("should successfully complete cart and fail on concurrent complete", async () => {
1876+
const paymentCollection = (
1877+
await api.post(
1878+
`/store/payment-collections`,
1879+
{ cart_id: cart.id },
1880+
storeHeaders
1881+
)
1882+
).data.payment_collection
1883+
1884+
await api.post(
1885+
`/store/payment-collections/${paymentCollection.id}/payment-sessions`,
1886+
{ provider_id: "pp_system_default" },
1887+
storeHeaders
1888+
)
1889+
1890+
await createCartCreditLinesWorkflow.run({
1891+
input: [
1892+
{
1893+
cart_id: cart.id,
1894+
amount: 100,
1895+
currency_code: "usd",
1896+
reference: "test",
1897+
reference_id: "test",
1898+
},
1899+
],
1900+
container: appContainer,
1901+
})
1902+
1903+
// Concurrently complete the cart
1904+
let completedCart: any[] = []
1905+
for (let i = 0; i < 5; i++) {
1906+
completedCart.push(
1907+
api
1908+
.post(`/store/carts/${cart.id}/complete`, {}, storeHeaders)
1909+
.catch((e) => e)
1910+
)
1911+
1912+
await setTimeout(25)
1913+
}
1914+
1915+
let all = await Promise.all(completedCart)
1916+
1917+
let success = all.filter((res) => res.status === 200)
1918+
let failure = all.filter((res) => res.status !== 200)
1919+
1920+
const successData = success[0].data.order
1921+
for (const res of success) {
1922+
expect(res.data.order).toEqual(successData)
1923+
}
1924+
1925+
expect(failure.length).toBeGreaterThan(0)
1926+
1927+
expect(successData).toEqual(
1928+
expect.objectContaining({
1929+
id: expect.any(String),
1930+
currency_code: "usd",
1931+
credit_lines: [
1932+
expect.objectContaining({
1933+
amount: 100,
1934+
reference: "test",
1935+
reference_id: "test",
1936+
}),
1937+
],
1938+
items: expect.arrayContaining([
1939+
expect.objectContaining({
1940+
unit_price: 1500,
1941+
compare_at_unit_price: null,
1942+
quantity: 1,
1943+
}),
1944+
]),
1945+
})
1946+
)
1947+
})
1948+
18711949
it("should successfully complete cart", async () => {
18721950
const paymentCollection = (
18731951
await api.post(
@@ -1883,7 +1961,7 @@ medusaIntegrationTestRunner({
18831961
storeHeaders
18841962
)
18851963

1886-
createCartCreditLinesWorkflow.run({
1964+
await createCartCreditLinesWorkflow.run({
18871965
input: [
18881966
{
18891967
cart_id: cart.id,

‎packages/core/core-flows/src/cart/workflows/complete-cart.ts‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -109,10 +109,13 @@ export const completeCartWorkflow = createWorkflow(
109109
entity: "order_cart",
110110
fields: ["cart_id", "order_id"],
111111
filters: { cart_id: input.id },
112+
options: {
113+
isList: false,
114+
},
112115
})
113116

114117
const orderId = transform({ orderCart }, ({ orderCart }) => {
115-
return orderCart.data[0]?.order_id
118+
return orderCart?.data?.order_id
116119
})
117120

118121
const cart = useRemoteQueryStep({
@@ -263,7 +266,7 @@ export const completeCartWorkflow = createWorkflow(
263266
const createdOrders = createOrdersStep([cartToOrder])
264267

265268
const createdOrder = transform({ createdOrders }, ({ createdOrders }) => {
266-
return createdOrders?.[0] ?? undefined
269+
return createdOrders[0]
267270
})
268271

269272
const reservationItemsData = transform(

‎packages/core/orchestration/src/__tests__/transaction/transaction-orchestrator.ts‎

Lines changed: 10 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -207,7 +207,6 @@ describe("Transaction Orchestrator", () => {
207207
},
208208
{
209209
action: "three",
210-
async: true,
211210
maxRetries: 0,
212211
next: {
213212
action: "five",
@@ -228,24 +227,14 @@ describe("Transaction Orchestrator", () => {
228227

229228
await strategy.resume(transaction)
230229

231-
expect(transaction.getErrors()).toHaveLength(2)
230+
expect(transaction.getErrors()).toHaveLength(1)
232231
expect(transaction.getErrors()).toEqual([
233232
{
234233
action: "three",
235-
error: {
234+
error: expect.objectContaining({
236235
message: "Step 3 failed",
237236
name: "Error",
238237
stack: expect.any(String),
239-
},
240-
handlerType: "invoke",
241-
},
242-
{
243-
action: "three",
244-
error: expect.objectContaining({
245-
message: expect.stringContaining(
246-
"Converting circular structure to JSON"
247-
),
248-
stack: expect.any(String),
249238
}),
250239
handlerType: "invoke",
251240
},
@@ -1052,6 +1041,8 @@ describe("Transaction Orchestrator", () => {
10521041

10531042
await strategy.resume(transaction)
10541043

1044+
await new Promise((resolve) => process.nextTick(resolve))
1045+
10551046
expect(mocks.one).toHaveBeenCalledTimes(1)
10561047
expect(mocks.two).toHaveBeenCalledTimes(0)
10571048
expect(transaction.getState()).toBe(TransactionState.INVOKING)
@@ -1148,6 +1139,8 @@ describe("Transaction Orchestrator", () => {
11481139

11491140
await strategy.resume(transaction)
11501141

1142+
await new Promise((resolve) => process.nextTick(resolve))
1143+
11511144
expect(mocks.one).toHaveBeenCalledTimes(1)
11521145
expect(mocks.compensateOne).toHaveBeenCalledTimes(0)
11531146
expect(mocks.two).toHaveBeenCalledTimes(0)
@@ -1171,6 +1164,8 @@ describe("Transaction Orchestrator", () => {
11711164
transaction,
11721165
})
11731166

1167+
await new Promise((resolve) => process.nextTick(resolve))
1168+
11741169
expect(resumedTransaction.getState()).toBe(TransactionState.COMPENSATING)
11751170
expect(mocks.compensateOne).toHaveBeenCalledTimes(1)
11761171

@@ -1263,6 +1258,7 @@ describe("Transaction Orchestrator", () => {
12631258
})
12641259

12651260
await strategy.resume(transaction)
1261+
await new Promise((resolve) => process.nextTick(resolve))
12661262

12671263
expect(mocks.one).toHaveBeenCalledTimes(1)
12681264
expect(mocks.compensateOne).toHaveBeenCalledTimes(1)
@@ -1335,6 +1331,7 @@ describe("Transaction Orchestrator", () => {
13351331
})
13361332

13371333
await strategy.resume(transaction)
1334+
await new Promise((resolve) => process.nextTick(resolve))
13381335

13391336
expect(transaction.getState()).toBe(TransactionState.DONE)
13401337
expect(mocks.one).toHaveBeenCalledTimes(1)

‎packages/core/orchestration/src/__tests__/workflow/global-workflow.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,8 @@ describe("WorkflowManager", () => {
116116
it("should continue an asyncronous transaction after reporting a successful step", async () => {
117117
const transaction = await flow.run("deliver-product", "t-id")
118118

119+
await new Promise((resolve) => process.nextTick(resolve))
120+
119121
expect(handlers.get("foo").invoke).toHaveBeenCalledTimes(1)
120122
expect(handlers.get("callExternal").invoke).toHaveBeenCalledTimes(1)
121123
expect(handlers.get("bar").invoke).toHaveBeenCalledTimes(0)
@@ -135,6 +137,8 @@ describe("WorkflowManager", () => {
135137
it("should revert an asyncronous transaction after reporting a failure step", async () => {
136138
const transaction = await flow.run("deliver-product", "t-id")
137139

140+
await new Promise((resolve) => process.nextTick(resolve))
141+
138142
expect(handlers.get("foo").invoke).toHaveBeenCalledTimes(1)
139143
expect(handlers.get("callExternal").invoke).toHaveBeenCalledTimes(1)
140144
expect(handlers.get("bar").invoke).toHaveBeenCalledTimes(0)

‎packages/core/orchestration/src/__tests__/workflow/local-workflow.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -158,6 +158,8 @@ describe("WorkflowManager", () => {
158158
const flow = new LocalWorkflow("deliver-product", container)
159159
const transaction = await flow.run("t-id")
160160

161+
await new Promise((resolve) => process.nextTick(resolve))
162+
161163
expect(handlers.get("foo").invoke).toHaveBeenCalledTimes(1)
162164
expect(handlers.get("callExternal").invoke).toHaveBeenCalledTimes(1)
163165
expect(handlers.get("bar").invoke).toHaveBeenCalledTimes(0)
@@ -177,6 +179,8 @@ describe("WorkflowManager", () => {
177179
const flow = new LocalWorkflow("deliver-product", container)
178180
const transaction = await flow.run("t-id")
179181

182+
await new Promise((resolve) => process.nextTick(resolve))
183+
180184
expect(handlers.get("foo").invoke).toHaveBeenCalledTimes(1)
181185
expect(handlers.get("callExternal").invoke).toHaveBeenCalledTimes(1)
182186
expect(handlers.get("bar").invoke).toHaveBeenCalledTimes(0)

‎packages/core/orchestration/src/transaction/transaction-orchestrator.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -951,8 +951,10 @@ export class TransactionOrchestrator extends EventEmitter {
951951
this.executeSyncStep(promise, transaction, step, nextSteps)
952952
)
953953
} else {
954-
// Execute async step in background and continue the execution of the transaction
955-
this.executeAsyncStep(promise, transaction, step, nextSteps)
954+
// Execute async step in background as part of the next event loop cycle and continue the execution of the transaction
955+
process.nextTick(() =>
956+
this.executeAsyncStep(promise, transaction, step, nextSteps)
957+
)
956958
hasAsyncSteps = true
957959
}
958960
}

‎packages/core/workflows-sdk/src/utils/composer/transform.ts‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,10 @@ export function transform(
167167
const ret = {
168168
__id: uniqId,
169169
__type: OrchestrationUtils.SymbolWorkflowStepTransformer,
170-
__temporary_storage_key: null as { key: string } | null,
170+
} as WorkflowData & {
171+
__id: string
172+
__type: string
173+
__temporary_storage_key: { key: string } | null
171174
}
172175

173176
const returnFn = async function (
@@ -176,6 +179,7 @@ export function transform(
176179
): Promise<any> {
177180
if ("transaction" in transactionContext) {
178181
const temporaryDataKey = `${transactionContext.transaction.modelId}_${transactionContext.transaction.transactionId}_${uniqId}`
182+
179183
ret.__temporary_storage_key ??= { key: temporaryDataKey }
180184

181185
if (
@@ -199,15 +203,14 @@ export function transform(
199203
const fn = functions[i]
200204
const arg = i === 0 ? stepValue : finalResult
201205

202-
finalResult = await fn.apply(fn, [arg, transactionContext])
206+
finalResult = fn.apply(fn, [arg, transactionContext])
207+
if (finalResult instanceof Promise) {
208+
finalResult = await finalResult
209+
}
203210
}
204211

205212
if ("transaction" in transactionContext) {
206213
const temporaryDataKey = ret.__temporary_storage_key!
207-
if (!temporaryDataKey) {
208-
return finalResult
209-
}
210-
211214
transactionContext.transaction.setTemporaryData(
212215
temporaryDataKey,
213216
finalResult
@@ -217,10 +220,11 @@ export function transform(
217220
return finalResult
218221
}
219222

220-
const proxyfiedRet = proxify<WorkflowData & { __resolver: any }>(
221-
ret as unknown as WorkflowData
222-
)
223+
const proxyfiedRet = proxify<
224+
WorkflowData & { __resolver: any; __temporary_storage_key: string | null }
225+
>(ret as unknown as WorkflowData)
223226
proxyfiedRet.__resolver = returnFn as any
227+
proxyfiedRet.__temporary_storage_key = null as string | null
224228

225229
return proxyfiedRet
226230
}

‎packages/medusa/src/api/store/carts/[id]/complete/route.ts‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
1-
import { completeCartWorkflow } from "@medusajs/core-flows"
1+
import { completeCartWorkflowId } from "@medusajs/core-flows"
22
import { prepareRetrieveQuery } from "@medusajs/framework"
33
import { MedusaRequest, MedusaResponse } from "@medusajs/framework/http"
44
import { HttpTypes } from "@medusajs/framework/types"
55
import {
66
ContainerRegistrationKeys,
77
MedusaError,
8+
Modules,
89
} from "@medusajs/framework/utils"
910
import { refetchCart } from "../../helpers"
1011
import { defaultStoreCartFields } from "../../query-config"
@@ -14,13 +15,21 @@ export const POST = async (
1415
res: MedusaResponse<HttpTypes.StoreCompleteCartResponse>
1516
) => {
1617
const cart_id = req.params.id
18+
const we = req.scope.resolve(Modules.WORKFLOW_ENGINE)
1719

18-
const { errors, result } = await completeCartWorkflow(req.scope).run({
20+
const { errors, result, transaction } = await we.run(completeCartWorkflowId, {
1921
input: { id: cart_id },
20-
context: { transactionId: cart_id },
22+
transactionId: cart_id,
2123
throwOnError: false,
2224
})
2325

26+
if (!transaction.hasFinished()) {
27+
throw new MedusaError(
28+
MedusaError.Types.CONFLICT,
29+
"Cart is already being completed by another request"
30+
)
31+
}
32+
2433
const query = req.scope.resolve(ContainerRegistrationKeys.QUERY)
2534

2635
// When an error occurs on the workflow, its potentially got to with cart validations, payments
@@ -47,7 +56,7 @@ export const POST = async (
4756
).remoteQueryConfig.fields
4857
)
4958

50-
if (!statusOKErrors.includes(error.type)) {
59+
if (!statusOKErrors.includes(error?.type)) {
5160
throw error
5261
}
5362

0 commit comments

Comments
 (0)