Diagnostic Workflows & Background Jobs
Scheduled jobs execute twice per configured interval
You wrote one job in src/jobs, gave it one cron schedule, and every tick it runs twice. Sometimes three or four times. No code path in the job itself loops or retries, and the schedule string is correct. The cause is not in the job at all. It is in how many processes are alive and doing background work at the same time. Here is why Medusa v2 lets every one of those processes fire the same tick on its own, and a script that proves it from the Workflow Engine Module's own execution history.
Medusa v2 scheduled jobs are registered and fired independently by every running process whose workerMode includes background processing, shared or worker. There is no distributed lock or single-leader coordination across instances. Run more than one such process against the same database, a common accident with medusa develop's dual internal processes, a duplicated worker deployment, or a container orchestrator scaling replicas without splitting workerMode, and every tick produces N executions, one per process still doing background work. Because scheduled jobs run as workflows, every firing is recorded by the Workflow Engine Module. Resolve it and call listWorkflowExecutions({ workflow_id: "job-name" }), bucket the rows by cron tick, and any bucket with more than one transaction_id is a duplicate-fire tick. Full code, tests, and a dry run guard are below.
The problem in plain words
A scheduled job in Medusa v2 is a file in src/jobs that exports a handler and a config object with a name and a schedule. At startup, Medusa reads that file, wraps the handler in a workflow, and registers a cron tick for it on whichever process just booted. That registration step is the whole story, and it repeats identically on every process that boots with background processing enabled.
Nothing in Medusa asks "has some other process already registered this job." Nothing elects one process as the leader. Each process that matches workerMode: "shared" or workerMode: "worker" independently sets up its own cron timer for the same job name, against the same database, and fires the job's exported function on its own schedule. Run two such processes and the tick that was meant to happen once now happens twice, at nearly the same moment, each producing its own workflow execution.
Why it happens
Every one of these is a real gap in the default setup, not one single bug:
- Medusa v2 scheduled jobs are registered and fired independently by every running process whose
workerModeincludes background processing,sharedorworker. There is no distributed lock or single-leader coordination across instances. medusa developruns dual internal processes during local development, and both can end up doing background work if the setup is copied into a script or container without noticing.- A duplicated worker deployment, for example a rollout that leaves the old worker replica running alongside the new one for a few minutes, means both are ticking the same job against the same database at once.
- A container orchestrator scaling replicas without splitting
workerModeper replica is the most common production cause. Scale a service from one replica to three and, unlessworkerModeis set toserveron all but one, all three now tick every scheduled job. - Medusa's own v2.6.1 changelog, Improved Scheduled Jobs, acknowledged a related registration bug and restricted execution to worker or shared mode instances only, but it does not dedupe across multiple such instances. GitHub issue #12977, Scheduled Jobs Execute Twice, reports exactly this across several users on 2.8.x.
This is a common source of confusion because the job's own code looks completely correct, the schedule string is right, and the logs show the handler running exactly as written, just more than once per tick. See the citations at the end for the exact issues and docs.
You do not need to guess how many processes are ticking the job. Scheduled jobs run as workflows under the hood, so every firing leaves a row in the Workflow Engine Module's own execution history. Resolve the workflow engine service and call listWorkflowExecutions({ workflow_id: "job-name" }, { order: { created_at: "ASC" } }), or hit the same data through the Admin API. Group the rows by the cron tick they belong to. A bucket with more than one transaction_id for the same workflow_id is not a retry, it is a second process firing the same tick.
The fix, as a flow
This is an infrastructure and config defect, not a data-corruption one, so the safe action is to flag and report, never to auto-mutate application data. Duplicated emails or exports already sent cannot be un-sent. The script lists executions for the job, buckets them by tick, flags any bucket with more than one transaction_id, and, only under DRY_RUN=false, writes an audit report or alert. It never resends suppressed side effects and never deletes workflow execution rows.
Build it step by step
Get an admin session and the base URL
Point the script at your Medusa backend and an admin user with rights to read workflow execution data. Exchange the email and password for a JWT once, then send it as a Bearer token on every admin call. Keep everything in environment variables, never hardcoded.
pip install requests
export MEDUSA_BACKEND_URL="http://localhost:9000"
export MEDUSA_ADMIN_EMAIL="admin@example.com"
export MEDUSA_ADMIN_PASSWORD="supersecret"
export JOB_WORKFLOW_ID="job-name"
export JOB_CRON="*/15 * * * *" # the schedule from your src/jobs config
export DRY_RUN="true" # start safe, only reports the duplicate ticks// Node 18+ has fetch built in, no dependencies needed
export MEDUSA_BACKEND_URL="http://localhost:9000"
export MEDUSA_ADMIN_EMAIL="admin@example.com"
export MEDUSA_ADMIN_PASSWORD="supersecret"
export JOB_WORKFLOW_ID="job-name"
export JOB_CRON="*/15 * * * *" // the schedule from your src/jobs config
export DRY_RUN="true" // start safe, only reports the duplicate ticksAuthenticate against the Admin API
Every call sends a Bearer JWT from POST /auth/user/emailpass. A small helper exchanges the credentials once and returns the token, which every later request reuses.
import os, requests
BASE_URL = os.environ.get("MEDUSA_BACKEND_URL", "http://localhost:9000")
EMAIL = os.environ["MEDUSA_ADMIN_EMAIL"]
PASSWORD = os.environ["MEDUSA_ADMIN_PASSWORD"]
def get_token():
r = requests.post(
f"{BASE_URL}/auth/user/emailpass",
json={"email": EMAIL, "password": PASSWORD},
timeout=30,
)
r.raise_for_status()
return r.json()["token"]const BASE_URL = process.env.MEDUSA_BACKEND_URL || "http://localhost:9000";
const EMAIL = process.env.MEDUSA_ADMIN_EMAIL;
const PASSWORD = process.env.MEDUSA_ADMIN_PASSWORD;
async function getToken() {
const res = await fetch(`${BASE_URL}/auth/user/emailpass`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ email: EMAIL, password: PASSWORD }),
});
if (!res.ok) throw new Error(`Medusa auth ${res.status}`);
const body = await res.json();
return body.token;
}List the workflow executions for the job
Hit GET /admin/workflows-executions filtered to the job's workflow_id, asking only for the fields the decision needs: id, transaction_id, workflow_id, created_at, and state. Order by created_at ascending so ticks line up in the order they actually fired.
EXECUTION_FIELDS = "id,transaction_id,workflow_id,created_at,state"
def list_workflow_executions(token, workflow_id, limit=200):
headers = {"Authorization": f"Bearer {token}"}
r = requests.get(
f"{BASE_URL}/admin/workflows-executions",
params={"workflow_id": workflow_id, "fields": EXECUTION_FIELDS, "limit": limit,
"order": "created_at"},
headers=headers,
timeout=30,
)
r.raise_for_status()
return r.json()["workflow_executions"]const EXECUTION_FIELDS = "id,transaction_id,workflow_id,created_at,state";
async function listWorkflowExecutions(token, workflowId, limit = 200) {
const params = new URLSearchParams({
workflow_id: workflowId, fields: EXECUTION_FIELDS, limit: String(limit), order: "created_at",
});
const res = await fetch(`${BASE_URL}/admin/workflows-executions?${params}`, {
headers: { Authorization: `Bearer ${token}` },
});
if (!res.ok) throw new Error(`Medusa ${res.status}`);
const body = await res.json();
return body.workflow_executions;
}Decide, with one pure function
Keep the decision in a function with no network calls. Given the raw execution rows, the job's own cron schedule, and a small tolerance in milliseconds, work out the nearest expected tick boundary for each execution, bucket the rows that land within tolerance of the same boundary, and keep only the buckets that have more than one distinct transaction_id. This is the part that gets tested in isolation, with fixed input arrays and no clock access.
from datetime import datetime, timedelta
def _parse_field(field, lo, hi):
if field == "*":
return set(range(lo, hi + 1))
values = set()
for part in field.split(","):
if part.startswith("*/"):
values.update(range(lo, hi + 1, int(part[2:])))
elif "-" in part:
a, b = part.split("-")
values.update(range(int(a), int(b) + 1))
else:
values.add(int(part))
return values
def _parse_cron(cron_expression):
minute, hour, dom, month, dow = cron_expression.strip().split()
return {
"minute": _parse_field(minute, 0, 59),
"hour": _parse_field(hour, 0, 23),
"dom": _parse_field(dom, 1, 31),
"month": _parse_field(month, 1, 12),
"dow": _parse_field(dow, 0, 6),
}
def _matches(dt, spec):
if dt.minute not in spec["minute"] or dt.hour not in spec["hour"] or dt.month not in spec["month"]:
return False
dom_ok = dt.day in spec["dom"]
dow_ok = (dt.weekday() + 1) % 7 in spec["dow"]
if spec["dom"] != set(range(1, 32)) and spec["dow"] != set(range(0, 7)):
return dom_ok or dow_ok
return dom_ok and dow_ok
def nearest_tick_boundary(cron_expression, at, search_minutes=1440):
"""Minute-aligned tick boundary matching the cron spec closest to `at`."""
spec = _parse_cron(cron_expression)
base = at.replace(second=0, microsecond=0)
if _matches(base, spec):
return base
for offset in range(1, search_minutes + 1):
earlier = base - timedelta(minutes=offset)
if _matches(earlier, spec):
return earlier
later = base + timedelta(minutes=offset)
if _matches(later, spec):
return later
raise RuntimeError("No matching tick boundary found within search window")
def find_duplicate_ticks(executions, cron_schedule, bucket_tolerance_ms=5000):
"""Pure: no I/O, no clock access. executions is a list of
{"workflow_id", "transaction_id", "created_at"} dicts already fetched."""
by_workflow = {}
for execution in executions:
by_workflow.setdefault(execution["workflow_id"], []).append(execution)
duplicates = []
for workflow_id, rows in by_workflow.items():
buckets = {}
for row in rows:
created_at = datetime.fromisoformat(row["created_at"].replace("Z", "+00:00"))
tick = nearest_tick_boundary(cron_schedule, created_at)
delta_ms = abs((created_at - tick).total_seconds() * 1000)
if delta_ms > bucket_tolerance_ms:
continue
key = tick.isoformat()
buckets.setdefault(key, set()).add(row["transaction_id"])
for tick_bucket, tx_ids in buckets.items():
if len(tx_ids) > 1:
duplicates.append({
"tickBucket": tick_bucket,
"transactionIds": sorted(tx_ids),
})
duplicates.sort(key=lambda d: d["tickBucket"])
return duplicatesfunction parseField(field, lo, hi) {
if (field === "*") return new Set(Array.from({ length: hi - lo + 1 }, (_, i) => lo + i));
const values = new Set();
for (const part of field.split(",")) {
if (part.startsWith("*/")) {
const step = Number(part.slice(2));
for (let v = lo; v <= hi; v += step) values.add(v);
} else if (part.includes("-")) {
const [a, b] = part.split("-").map(Number);
for (let v = a; v <= b; v++) values.add(v);
} else {
values.add(Number(part));
}
}
return values;
}
function parseCron(cronExpression) {
const [minute, hour, dom, month, dow] = cronExpression.trim().split(/\s+/);
return {
minute: parseField(minute, 0, 59),
hour: parseField(hour, 0, 23),
dom: parseField(dom, 1, 31),
month: parseField(month, 1, 12),
dow: parseField(dow, 0, 6),
};
}
function fullRange(lo, hi) {
return new Set(Array.from({ length: hi - lo + 1 }, (_, i) => lo + i));
}
function matches(date, spec) {
if (!spec.minute.has(date.getUTCMinutes())) return false;
if (!spec.hour.has(date.getUTCHours())) return false;
if (!spec.month.has(date.getUTCMonth() + 1)) return false;
const domOk = spec.dom.has(date.getUTCDate());
const dowOk = spec.dow.has(date.getUTCDay());
const domIsFull = spec.dom.size === fullRange(1, 31).size;
const dowIsFull = spec.dow.size === fullRange(0, 6).size;
if (!domIsFull && !dowIsFull) return domOk || dowOk;
return domOk && dowOk;
}
export function nearestTickBoundary(cronExpression, at, searchMinutes = 1440) {
// Minute-aligned tick boundary matching the cron spec closest to `at`.
const spec = parseCron(cronExpression);
const base = new Date(at.getTime());
base.setUTCSeconds(0, 0);
if (matches(base, spec)) return base;
for (let offset = 1; offset <= searchMinutes; offset++) {
const earlier = new Date(base.getTime() - offset * 60000);
if (matches(earlier, spec)) return earlier;
const later = new Date(base.getTime() + offset * 60000);
if (matches(later, spec)) return later;
}
throw new Error("No matching tick boundary found within search window");
}
export function findDuplicateTicks(executions, cronSchedule, bucketToleranceMs = 5000) {
// Pure: no I/O, no clock access. executions is a plain array of
// { workflow_id, transaction_id, created_at } already fetched.
const byWorkflow = new Map();
for (const execution of executions) {
const list = byWorkflow.get(execution.workflow_id) || [];
list.push(execution);
byWorkflow.set(execution.workflow_id, list);
}
const duplicates = [];
for (const rows of byWorkflow.values()) {
const buckets = new Map();
for (const row of rows) {
const createdAt = new Date(row.created_at);
const tick = nearestTickBoundary(cronSchedule, createdAt);
const deltaMs = Math.abs(createdAt.getTime() - tick.getTime());
if (deltaMs > bucketToleranceMs) continue;
const key = tick.toISOString();
const set = buckets.get(key) || new Set();
set.add(row.transaction_id);
buckets.set(key, set);
}
for (const [tickBucket, txIds] of buckets.entries()) {
if (txIds.size > 1) {
duplicates.push({ tickBucket, transactionIds: [...txIds].sort() });
}
}
}
return duplicates.sort((a, b) => a.tickBucket.localeCompare(b.tickBucket));
}Report the duplicate ticks, never mutate
Marking a tick as duplicated is a finding, not a repair. There is no safe way to un-send an email or un-write an export that already happened twice, so the write path is limited to producing an audit report or an alert. It never deletes workflow execution rows and never resends anything the job itself already triggered.
def write_audit_report(job_name, duplicates):
"""The only write this script does: an audit log line per duplicate tick.
Never resends a suppressed side effect, never deletes an execution row."""
for item in duplicates:
log.warning(
"DUPLICATE TICK job=%s tick=%s transaction_ids=%s inferred_replicas=%d",
job_name, item["tickBucket"], item["transactionIds"], len(item["transactionIds"]),
)function writeAuditReport(jobName, duplicates) {
// The only write this script does: an audit log line per duplicate tick.
// Never resends a suppressed side effect, never deletes an execution row.
for (const item of duplicates) {
console.warn(
`DUPLICATE TICK job=${jobName} tick=${item.tickBucket} transaction_ids=${JSON.stringify(item.transactionIds)} inferred_replicas=${item.transactionIds.length}`
);
}
}Wire it together with a dry run guard
The loop ties every piece together. On the first few runs, leave DRY_RUN on so the script only prints the duplicate ticks it found, sorted by tick time. Confirm from your deployment that more than one instance was running in shared or worker mode during those ticks, fix workerMode per instance in medusa-config.ts, then switch DRY_RUN off only to let the audit report get written somewhere durable, such as a log aggregator or a paging system.
Never let this script resend a suppressed email, export, or webhook just because it found a duplicate. Never delete a workflow execution row, since that destroys the very evidence you are trying to read. The only allowed write, even with DRY_RUN=false, is writing the audit report or alert. The real fix is setting workerMode explicitly per instance.
The full code
Here is the complete script in one file for each language. It authenticates, lists the workflow executions for the job, buckets them by cron tick with a pure function, and either reports the duplicate ticks in dry run or writes the audit report, depending on DRY_RUN. It never resends a side effect and never deletes an execution row.
"""Find Medusa v2 scheduled job ticks that fired more than once, because
more than one process is running in shared or worker WORKER_MODE against
the same database, with no distributed lock coordinating them.
This is an infrastructure and config defect, not a data problem. It only
reports the duplicate ticks, it never resends a suppressed side effect
and never deletes a workflow_execution row. Safe to run again and again.
"""
import os
import logging
from datetime import datetime, timedelta
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("find_duplicate_ticks")
BASE_URL = os.environ.get("MEDUSA_BACKEND_URL", "http://localhost:9000")
EMAIL = os.environ.get("MEDUSA_ADMIN_EMAIL", "admin@example.com")
PASSWORD = os.environ.get("MEDUSA_ADMIN_PASSWORD", "supersecret")
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"
JOB_WORKFLOW_ID = os.environ.get("JOB_WORKFLOW_ID", "job-name")
JOB_CRON = os.environ.get("JOB_CRON", "*/15 * * * *")
BUCKET_TOLERANCE_MS = float(os.environ.get("BUCKET_TOLERANCE_MS", "5000"))
EXECUTION_FIELDS = "id,transaction_id,workflow_id,created_at,state"
def get_token():
r = requests.post(
f"{BASE_URL}/auth/user/emailpass",
json={"email": EMAIL, "password": PASSWORD},
timeout=30,
)
r.raise_for_status()
return r.json()["token"]
def list_workflow_executions(token, workflow_id, limit=200):
headers = {"Authorization": f"Bearer {token}"}
r = requests.get(
f"{BASE_URL}/admin/workflows-executions",
params={"workflow_id": workflow_id, "fields": EXECUTION_FIELDS, "limit": limit,
"order": "created_at"},
headers=headers,
timeout=30,
)
r.raise_for_status()
return r.json()["workflow_executions"]
def _parse_field(field, lo, hi):
"""Parse one cron field (*, N, N-M, N,M, */S) into a set of allowed ints."""
if field == "*":
return set(range(lo, hi + 1))
values = set()
for part in field.split(","):
if part.startswith("*/"):
values.update(range(lo, hi + 1, int(part[2:])))
elif "-" in part:
a, b = part.split("-")
values.update(range(int(a), int(b) + 1))
else:
values.add(int(part))
return values
def _parse_cron(cron_expression):
minute, hour, dom, month, dow = cron_expression.strip().split()
return {
"minute": _parse_field(minute, 0, 59),
"hour": _parse_field(hour, 0, 23),
"dom": _parse_field(dom, 1, 31),
"month": _parse_field(month, 1, 12),
"dow": _parse_field(dow, 0, 6),
}
def _matches(dt, spec):
if dt.minute not in spec["minute"] or dt.hour not in spec["hour"] or dt.month not in spec["month"]:
return False
dom_ok = dt.day in spec["dom"]
dow_ok = (dt.weekday() + 1) % 7 in spec["dow"]
if spec["dom"] != set(range(1, 32)) and spec["dow"] != set(range(0, 7)):
return dom_ok or dow_ok
return dom_ok and dow_ok
def nearest_tick_boundary(cron_expression, at, search_minutes=1440):
"""Minute-aligned tick boundary matching the cron spec closest to `at`."""
spec = _parse_cron(cron_expression)
base = at.replace(second=0, microsecond=0)
if _matches(base, spec):
return base
for offset in range(1, search_minutes + 1):
earlier = base - timedelta(minutes=offset)
if _matches(earlier, spec):
return earlier
later = base + timedelta(minutes=offset)
if _matches(later, spec):
return later
raise RuntimeError("No matching tick boundary found within search window")
def find_duplicate_ticks(executions, cron_schedule, bucket_tolerance_ms=5000):
"""Pure: no I/O, no clock access. executions is a list of
{"workflow_id", "transaction_id", "created_at"} dicts already fetched."""
by_workflow = {}
for execution in executions:
by_workflow.setdefault(execution["workflow_id"], []).append(execution)
duplicates = []
for workflow_id, rows in by_workflow.items():
buckets = {}
for row in rows:
created_at = datetime.fromisoformat(row["created_at"].replace("Z", "+00:00"))
tick = nearest_tick_boundary(cron_schedule, created_at)
delta_ms = abs((created_at - tick).total_seconds() * 1000)
if delta_ms > bucket_tolerance_ms:
continue
key = tick.isoformat()
buckets.setdefault(key, set()).add(row["transaction_id"])
for tick_bucket, tx_ids in buckets.items():
if len(tx_ids) > 1:
duplicates.append({
"tickBucket": tick_bucket,
"transactionIds": sorted(tx_ids),
})
duplicates.sort(key=lambda d: d["tickBucket"])
return duplicates
def write_audit_report(job_name, duplicates):
"""The only write this script does: an audit log line per duplicate tick.
Never resends a suppressed side effect, never deletes an execution row."""
for item in duplicates:
log.warning(
"DUPLICATE TICK job=%s tick=%s transaction_ids=%s inferred_replicas=%d",
job_name, item["tickBucket"], item["transactionIds"], len(item["transactionIds"]),
)
def run():
token = get_token()
executions = list_workflow_executions(token, JOB_WORKFLOW_ID)
duplicates = find_duplicate_ticks(executions, JOB_CRON, BUCKET_TOLERANCE_MS)
if not duplicates:
log.info("No duplicate ticks across %d execution(s) for %s.", len(executions), JOB_WORKFLOW_ID)
return
for item in duplicates:
log.warning(
"Tick %s fired %d time(s): %s",
item["tickBucket"], len(item["transactionIds"]), item["transactionIds"],
)
if not DRY_RUN:
write_audit_report(JOB_WORKFLOW_ID, duplicates)
log.info("Done. %d duplicate tick(s) %s.", len(duplicates), "to review" if DRY_RUN else "reported")
if __name__ == "__main__":
run()/**
* Find Medusa v2 scheduled job ticks that fired more than once, because
* more than one process is running in shared or worker WORKER_MODE against
* the same database, with no distributed lock coordinating them.
* This is an infrastructure and config defect, not a data problem. It only
* reports the duplicate ticks, it never resends a suppressed side effect
* and never deletes a workflow_execution row. Safe to run again and again.
*/
import { pathToFileURL } from "node:url";
const BASE_URL = process.env.MEDUSA_BACKEND_URL || "http://localhost:9000";
const EMAIL = process.env.MEDUSA_ADMIN_EMAIL || "admin@example.com";
const PASSWORD = process.env.MEDUSA_ADMIN_PASSWORD || "supersecret";
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
const JOB_WORKFLOW_ID = process.env.JOB_WORKFLOW_ID || "job-name";
const JOB_CRON = process.env.JOB_CRON || "*/15 * * * *";
const BUCKET_TOLERANCE_MS = Number(process.env.BUCKET_TOLERANCE_MS || 5000);
const EXECUTION_FIELDS = "id,transaction_id,workflow_id,created_at,state";
function parseField(field, lo, hi) {
if (field === "*") return new Set(Array.from({ length: hi - lo + 1 }, (_, i) => lo + i));
const values = new Set();
for (const part of field.split(",")) {
if (part.startsWith("*/")) {
const step = Number(part.slice(2));
for (let v = lo; v <= hi; v += step) values.add(v);
} else if (part.includes("-")) {
const [a, b] = part.split("-").map(Number);
for (let v = a; v <= b; v++) values.add(v);
} else {
values.add(Number(part));
}
}
return values;
}
function parseCron(cronExpression) {
const [minute, hour, dom, month, dow] = cronExpression.trim().split(/\s+/);
return {
minute: parseField(minute, 0, 59),
hour: parseField(hour, 0, 23),
dom: parseField(dom, 1, 31),
month: parseField(month, 1, 12),
dow: parseField(dow, 0, 6),
};
}
function fullRange(lo, hi) {
return new Set(Array.from({ length: hi - lo + 1 }, (_, i) => lo + i));
}
function matches(date, spec) {
if (!spec.minute.has(date.getUTCMinutes())) return false;
if (!spec.hour.has(date.getUTCHours())) return false;
if (!spec.month.has(date.getUTCMonth() + 1)) return false;
const domOk = spec.dom.has(date.getUTCDate());
const dowOk = spec.dow.has(date.getUTCDay());
const domIsFull = spec.dom.size === fullRange(1, 31).size;
const dowIsFull = spec.dow.size === fullRange(0, 6).size;
if (!domIsFull && !dowIsFull) return domOk || dowOk;
return domOk && dowOk;
}
export function nearestTickBoundary(cronExpression, at, searchMinutes = 1440) {
// Minute-aligned tick boundary matching the cron spec closest to `at`.
const spec = parseCron(cronExpression);
const base = new Date(at.getTime());
base.setUTCSeconds(0, 0);
if (matches(base, spec)) return base;
for (let offset = 1; offset <= searchMinutes; offset++) {
const earlier = new Date(base.getTime() - offset * 60000);
if (matches(earlier, spec)) return earlier;
const later = new Date(base.getTime() + offset * 60000);
if (matches(later, spec)) return later;
}
throw new Error("No matching tick boundary found within search window");
}
export function findDuplicateTicks(executions, cronSchedule, bucketToleranceMs = 5000) {
// Pure: no I/O, no clock access. executions is a plain array of
// { workflow_id, transaction_id, created_at } already fetched.
const byWorkflow = new Map();
for (const execution of executions) {
const list = byWorkflow.get(execution.workflow_id) || [];
list.push(execution);
byWorkflow.set(execution.workflow_id, list);
}
const duplicates = [];
for (const rows of byWorkflow.values()) {
const buckets = new Map();
for (const row of rows) {
const createdAt = new Date(row.created_at);
const tick = nearestTickBoundary(cronSchedule, createdAt);
const deltaMs = Math.abs(createdAt.getTime() - tick.getTime());
if (deltaMs > bucketToleranceMs) continue;
const key = tick.toISOString();
const set = buckets.get(key) || new Set();
set.add(row.transaction_id);
buckets.set(key, set);
}
for (const [tickBucket, txIds] of buckets.entries()) {
if (txIds.size > 1) {
duplicates.push({ tickBucket, transactionIds: [...txIds].sort() });
}
}
}
return duplicates.sort((a, b) => a.tickBucket.localeCompare(b.tickBucket));
}
async function getToken() {
const res = await fetch(`${BASE_URL}/auth/user/emailpass`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ email: EMAIL, password: PASSWORD }),
});
if (!res.ok) throw new Error(`Medusa auth ${res.status}`);
const body = await res.json();
return body.token;
}
async function listWorkflowExecutions(token, workflowId, limit = 200) {
const params = new URLSearchParams({
workflow_id: workflowId, fields: EXECUTION_FIELDS, limit: String(limit), order: "created_at",
});
const res = await fetch(`${BASE_URL}/admin/workflows-executions?${params}`, {
headers: { Authorization: `Bearer ${token}` },
});
if (!res.ok) throw new Error(`Medusa ${res.status}`);
const body = await res.json();
return body.workflow_executions;
}
function writeAuditReport(jobName, duplicates) {
// The only write this script does: an audit log line per duplicate tick.
// Never resends a suppressed side effect, never deletes an execution row.
for (const item of duplicates) {
console.warn(
`DUPLICATE TICK job=${jobName} tick=${item.tickBucket} transaction_ids=${JSON.stringify(item.transactionIds)} inferred_replicas=${item.transactionIds.length}`
);
}
}
export async function run() {
const token = await getToken();
const executions = await listWorkflowExecutions(token, JOB_WORKFLOW_ID);
const duplicates = findDuplicateTicks(executions, JOB_CRON, BUCKET_TOLERANCE_MS);
if (duplicates.length === 0) {
console.log(`No duplicate ticks across ${executions.length} execution(s) for ${JOB_WORKFLOW_ID}.`);
return;
}
for (const item of duplicates) {
console.warn(`Tick ${item.tickBucket} fired ${item.transactionIds.length} time(s): ${JSON.stringify(item.transactionIds)}`);
}
if (!DRY_RUN) {
writeAuditReport(JOB_WORKFLOW_ID, duplicates);
}
console.log(`Done. ${duplicates.length} duplicate tick(s) ${DRY_RUN ? "to review" : "reported"}.`);
}
if (import.meta.url === pathToFileURL(process.argv[1]).href) {
run().catch((err) => { console.error(err); process.exit(1); });
}Add a test
The function worth testing is the one that decides the outcome, find_duplicate_ticks. It is pure, no network and no external cron library, just date math against a tiny cron parser and a tolerance window, so the tests feed in plain execution rows plus a fixed schedule and check the answer.
from find_duplicate_ticks import find_duplicate_ticks
HOURLY = "0 * * * *"
def execution(**over):
base = {"workflow_id": "job-name", "transaction_id": "tx_1", "created_at": "2026-07-10T00:00:00Z"}
base.update(over)
return base
def test_no_duplicate_for_a_single_execution():
result = find_duplicate_ticks([execution()], HOURLY)
assert result == []
def test_duplicate_when_two_transactions_share_a_tick():
rows = [
execution(transaction_id="tx_1", created_at="2026-07-10T00:00:00Z"),
execution(transaction_id="tx_2", created_at="2026-07-10T00:00:02Z"),
]
result = find_duplicate_ticks(rows, HOURLY)
assert len(result) == 1
assert sorted(result[0]["transactionIds"]) == ["tx_1", "tx_2"]
def test_not_duplicate_when_on_different_ticks():
rows = [
execution(transaction_id="tx_1", created_at="2026-07-10T00:00:00Z"),
execution(transaction_id="tx_2", created_at="2026-07-10T01:00:00Z"),
]
result = find_duplicate_ticks(rows, HOURLY)
assert result == []
def test_same_transaction_twice_is_not_a_duplicate_tick():
rows = [
execution(transaction_id="tx_1", created_at="2026-07-10T00:00:00Z"),
execution(transaction_id="tx_1", created_at="2026-07-10T00:00:01Z"),
]
result = find_duplicate_ticks(rows, HOURLY)
assert result == []
def test_tolerance_window_still_groups_slightly_offset_fires():
rows = [
execution(transaction_id="tx_1", created_at="2026-07-10T00:00:00Z"),
execution(transaction_id="tx_2", created_at="2026-07-10T00:00:04Z"),
]
result = find_duplicate_ticks(rows, HOURLY, bucket_tolerance_ms=5000)
assert len(result) == 1
def test_three_processes_produce_three_transaction_ids():
rows = [
execution(transaction_id="tx_1", created_at="2026-07-10T00:00:00Z"),
execution(transaction_id="tx_2", created_at="2026-07-10T00:00:01Z"),
execution(transaction_id="tx_3", created_at="2026-07-10T00:00:02Z"),
]
result = find_duplicate_ticks(rows, HOURLY)
assert len(result) == 1
assert len(result[0]["transactionIds"]) == 3
def test_different_workflow_ids_are_kept_independent():
rows = [
execution(workflow_id="job-a", transaction_id="tx_1", created_at="2026-07-10T00:00:00Z"),
execution(workflow_id="job-a", transaction_id="tx_2", created_at="2026-07-10T00:00:01Z"),
execution(workflow_id="job-b", transaction_id="tx_3", created_at="2026-07-10T00:00:00Z"),
]
result = find_duplicate_ticks(rows, HOURLY)
assert len(result) == 1import { test } from "node:test";
import assert from "node:assert/strict";
import { findDuplicateTicks } from "./find-duplicate-ticks.js";
const HOURLY = "0 * * * *";
const execution = (over = {}) => ({
workflow_id: "job-name",
transaction_id: "tx_1",
created_at: "2026-07-10T00:00:00Z",
...over,
});
test("no duplicate for a single execution", () => {
const result = findDuplicateTicks([execution()], HOURLY);
assert.deepEqual(result, []);
});
test("duplicate when two transactions share a tick", () => {
const rows = [
execution({ transaction_id: "tx_1", created_at: "2026-07-10T00:00:00Z" }),
execution({ transaction_id: "tx_2", created_at: "2026-07-10T00:00:02Z" }),
];
const result = findDuplicateTicks(rows, HOURLY);
assert.equal(result.length, 1);
assert.deepEqual(result[0].transactionIds.sort(), ["tx_1", "tx_2"]);
});
test("not duplicate when on different ticks", () => {
const rows = [
execution({ transaction_id: "tx_1", created_at: "2026-07-10T00:00:00Z" }),
execution({ transaction_id: "tx_2", created_at: "2026-07-10T01:00:00Z" }),
];
const result = findDuplicateTicks(rows, HOURLY);
assert.deepEqual(result, []);
});
test("same transaction twice is not a duplicate tick", () => {
const rows = [
execution({ transaction_id: "tx_1", created_at: "2026-07-10T00:00:00Z" }),
execution({ transaction_id: "tx_1", created_at: "2026-07-10T00:00:01Z" }),
];
const result = findDuplicateTicks(rows, HOURLY);
assert.deepEqual(result, []);
});
test("tolerance window still groups slightly offset fires", () => {
const rows = [
execution({ transaction_id: "tx_1", created_at: "2026-07-10T00:00:00Z" }),
execution({ transaction_id: "tx_2", created_at: "2026-07-10T00:00:04Z" }),
];
const result = findDuplicateTicks(rows, HOURLY, 5000);
assert.equal(result.length, 1);
});
test("three processes produce three transaction ids", () => {
const rows = [
execution({ transaction_id: "tx_1", created_at: "2026-07-10T00:00:00Z" }),
execution({ transaction_id: "tx_2", created_at: "2026-07-10T00:00:01Z" }),
execution({ transaction_id: "tx_3", created_at: "2026-07-10T00:00:02Z" }),
];
const result = findDuplicateTicks(rows, HOURLY);
assert.equal(result.length, 1);
assert.equal(result[0].transactionIds.length, 3);
});
test("different workflow ids are kept independent", () => {
const rows = [
execution({ workflow_id: "job-a", transaction_id: "tx_1", created_at: "2026-07-10T00:00:00Z" }),
execution({ workflow_id: "job-a", transaction_id: "tx_2", created_at: "2026-07-10T00:00:01Z" }),
execution({ workflow_id: "job-b", transaction_id: "tx_3", created_at: "2026-07-10T00:00:00Z" }),
];
const result = findDuplicateTicks(rows, HOURLY);
assert.equal(result.length, 1);
});Case studies
The old worker replica that would not die fast enough
A team ran a single worker-mode replica alongside several server-mode API replicas, exactly as the docs describe. A rolling deploy started the new worker replica before the orchestrator finished draining the old one, and for about ninety seconds both were alive and both were ticking every scheduled job in the project.
A fifteen minute inventory sync job fired twice during that window, and a nightly digest email went out to every customer twice the next time a deploy landed at the wrong minute. Running the reconciler against listWorkflowExecutions for the digest job's workflow_id showed one tick bucket with two distinct transaction_id values two seconds apart, matching the deploy timestamps exactly.
Three replicas, one workerMode setting copied to all of them
A store scaled its Medusa deployment from one container to three during a traffic spike, using the same environment file for every replica because that was the simplest way to configure the orchestrator. Nobody split workerMode across the three, so every one of them registered every scheduled job as shared mode.
An abandoned cart reminder job began firing three times per configured interval, and support started getting complaints about customers receiving the same reminder email three times in a row. The reconciler's duplicate-tick report showed three transaction_id values per tick bucket, an exact match for the replica count, which pointed the team straight at the deployment config instead of the job code.
After the reconciler confirms which ticks fired more than once and by how many processes, the fix happens in medusa-config.ts, not in the job or the data. Set projectConfig.workerMode to worker on exactly one background instance, server on every HTTP-only replica, and never leave more than one replica on shared or worker at the same time. Set numberOfExecutions: 1 on the job's config export temporarily to confirm single-firing before rolling the change out, then remove it. The reconciler itself never mutates anything, so it stays safe to run on a schedule as an ongoing check.
FAQ
Why does my Medusa scheduled job execute more than once per interval?
Medusa v2 has no distributed lock or single-leader coordination for scheduled jobs. Every running process whose workerMode includes background processing, shared or worker, independently registers the job and fires it on its own copy of the cron schedule. If two or more such processes run against the same database, for example a duplicated worker deployment or a container orchestrator scaling replicas without splitting workerMode, every tick produces one execution per process still doing background work.
How do I detect duplicate scheduled job executions in Medusa?
Scheduled jobs run as workflows, so every firing is recorded by the Workflow Engine Module. Resolve the workflow engine service and call listWorkflowExecutions with the job's workflow_id, ordered by created_at. Group the rows into the tick each one belongs to using the job's cron schedule, and any tick bucket with more than one transaction_id for the same workflow_id is a duplicate-fire tick.
How do I fix a Medusa scheduled job that fires more than once?
This is an infrastructure and configuration defect, not a data problem, so the fix is to set projectConfig.workerMode explicitly per deployed instance in medusa-config.ts, worker on exactly one background instance, server on every HTTP-only replica, and never leave more than one replica on shared or worker mode at the same time. Use numberOfExecutions: 1 on the job's config export only to verify single-firing before rolling the change to production.
Related field notes
Citations
On the problem:
- medusajs/medusa GitHub issue #12977: Bug: Scheduled Jobs Execute Twice. github.com/medusajs/medusa/issues/12977
- medusajs/medusa release notes: v2.6.1, Improved Scheduled Jobs. github.com/medusajs/medusa/releases/tag/v2.6.1
- medusajs/medusa GitHub issue #13393: background jobs sometimes don't run or run in wrong timezone. github.com/medusajs/medusa/issues/13393
On the solution:
- Medusa Documentation: Scheduled Jobs. docs.medusajs.com/learn/fundamentals/scheduled-jobs
- Medusa Documentation: Scheduled Job Number of Executions. docs.medusajs.com/learn/fundamentals/scheduled-jobs/execution-number
- Medusa Documentation: Worker Mode of Medusa Instance. docs.medusajs.com/learn/production/worker-mode
Stuck on a tricky one?
If you have a problem in Medusa pricing, inventory, orders, promotions, or workflows that you would rather hand off, this is the kind of work I do. Message me and we can work through it together.
Did this catch a duplicate-firing job?
If this saved you from a doubled email blast or a duplicated sync, you can buy me a coffee. It is the best way to keep these field notes free and growing.
Buy me a coffee on Ko-fi