fix(engine): Always create cleaner job (#13557)
RESOLVES CORE-1206 **What** Instead of removing cleaner repeatable job and risk to remove it while other instances are still up, we always create it, since the id of a job is unique and we give one to the job, if already present it wont get added
This commit is contained in:
1 parent
e5a82518f1
commit
cb716856b6
3 files changed
+11
-11
No files matched your search
@@ -0,0 +1,6 @@
|
|||||||
|
---
|
||||||
|
"@medusajs/workflow-engine-redis": patch
|
||||||
|
"@medusajs/workflow-engine-inmemory": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
fix(engine): Always create cleaner job
|
||||||
+3
-1
@@ -29,6 +29,8 @@ import { WorkflowOrchestratorService } from "@services"
|
|||||||
import { type CronExpression, parseExpression } from "cron-parser"
|
import { type CronExpression, parseExpression } from "cron-parser"
|
||||||
import { WorkflowExecution } from "../models/workflow-execution"
|
import { WorkflowExecution } from "../models/workflow-execution"
|
||||||
|
|
||||||
|
const THIRTY_MINUTES_IN_MS = 1000 * 60 * 30
|
||||||
|
|
||||||
function calculateDelayFromExpression(expression: CronExpression): number {
|
function calculateDelayFromExpression(expression: CronExpression): number {
|
||||||
const nextTime = expression.next().getTime()
|
const nextTime = expression.next().getTime()
|
||||||
const now = Date.now()
|
const now = Date.now()
|
||||||
@@ -127,7 +129,7 @@ export class InMemoryDistributedTransactionStorage
|
|||||||
try {
|
try {
|
||||||
await this.clearExpiredExecutions()
|
await this.clearExpiredExecutions()
|
||||||
} catch {}
|
} catch {}
|
||||||
}, 1000 * 60 * 60)
|
}, THIRTY_MINUTES_IN_MS)
|
||||||
}
|
}
|
||||||
|
|
||||||
async onApplicationShutdown() {
|
async onApplicationShutdown() {
|
||||||
|
|||||||
@@ -33,7 +33,7 @@ enum JobType {
|
|||||||
TRANSACTION_TIMEOUT = "transaction_timeout",
|
TRANSACTION_TIMEOUT = "transaction_timeout",
|
||||||
}
|
}
|
||||||
|
|
||||||
const ONE_HOUR_IN_MS = 1000 * 60 * 60
|
const THIRTY_MINUTES_IN_MS = 1000 * 60 * 30
|
||||||
const REPEATABLE_CLEARER_JOB_ID = "clear-expired-executions"
|
const REPEATABLE_CLEARER_JOB_ID = "clear-expired-executions"
|
||||||
|
|
||||||
const invokingStatesSet = new Set([
|
const invokingStatesSet = new Set([
|
||||||
@@ -118,14 +118,6 @@ export class RedisDistributedTransactionStorage
|
|||||||
await this.worker?.close()
|
await this.worker?.close()
|
||||||
await this.jobWorker?.close()
|
await this.jobWorker?.close()
|
||||||
|
|
||||||
// Clean up repeatable jobs
|
|
||||||
const repeatableJobs = (await this.cleanerQueue_?.getRepeatableJobs()) ?? []
|
|
||||||
for (const job of repeatableJobs) {
|
|
||||||
if (job.id === REPEATABLE_CLEARER_JOB_ID) {
|
|
||||||
await this.cleanerQueue_?.removeRepeatableByKey(job.key)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
await this.cleanerWorker_?.close()
|
await this.cleanerWorker_?.close()
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -206,7 +198,7 @@ export class RedisDistributedTransactionStorage
|
|||||||
{},
|
{},
|
||||||
{
|
{
|
||||||
repeat: {
|
repeat: {
|
||||||
every: ONE_HOUR_IN_MS,
|
every: THIRTY_MINUTES_IN_MS,
|
||||||
},
|
},
|
||||||
jobId: REPEATABLE_CLEARER_JOB_ID,
|
jobId: REPEATABLE_CLEARER_JOB_ID,
|
||||||
removeOnComplete: true,
|
removeOnComplete: true,
|
||||||
|
|||||||
Reference in new issue
Block a user