Skip to content
allanninal.dev/medusa
Collaborate

Field notes / Events & Notifications

Reconciler Events & Notifications

Redis event bus intermittently drops or delays subscriber execution

A customer completes checkout, the order shows up fine in the Admin, and then no confirmation email ever goes out. Nothing errored. Nothing retried loudly. In Medusa v2, the default Redis Event Bus Module queues every emitted event, such as order.placed from the complete cart workflow, as a BullMQ job for a worker to consume, and if the worker that owns the subscriber starts late, restarts, autoscales, or crashes mid-job, that job can be picked up with nothing listening, retried past its limit, or dropped outright. Here is why that timing gap exists and a script that finds exactly which orders it happened to.

Python and Node.js Medusa Admin API Reconcile against the Notification module
Two server racks
Photo by Eric Stoynov on Unsplash
The short answer

Medusa v2's Redis Event Bus Module queues emitted domain events, like order.placed, as BullMQ jobs. Delivery is at-least-once, not guaranteed-ordered and not guaranteed-delivered, so a worker that subscribes after the event already published, a crash mid-job, or a horizontally scaled pool of instances racing the same queue can all leave a job with no matching subscriber attached, retried past its attempts, or dropped depending on removeOnFail and retry settings. You cannot safely tell "delayed" from "dropped" by looking at the order alone, so pull every order in a window from /admin/orders, pull every notification Medusa recorded from /admin/notifications, and diff the two: an order with a matching order.placed notification that arrived quickly is delivered, one that arrived late is delayed, one with no matching notification at all is dropped. Only confirmed drops are candidates for a manual, opt-in re-emit through emitEventStep. Full code, tests, and a dry run guard are below.

The problem in plain words

When completeCartWorkflow finishes turning a cart into an order, it emits order.placed so subscribers can send the confirmation email, sync inventory, or notify a warehouse. In production, Medusa's Redis Event Bus Module does not call those subscribers directly. It hands the event to BullMQ, which writes it as a job onto a Redis-backed queue, and a worker process picks that job up whenever it is free and calls the matching subscriber.

That extra hop is what makes the event bus durable across restarts, but it also opens a timing gap. The event is published the instant completeCartWorkflow runs. The subscriber only exists once a server process has finished loading and registered its handler. If those two things happen in the wrong order, or if the instance that would have handled the job is mid-restart, mid-autoscale, or has just crashed, the job can sit in the queue, get picked up by an instance with no handler for it, or get retried until it exhausts its attempt limit and is discarded. From Redis and BullMQ's point of view, none of that is a failure. The job existed and something eventually happened to it. From the store's point of view, the customer just never got their email.

completeCartWorkflow emits order.placed BullMQ / Redis job queued for a worker to consume ✕ timing gap or crash Worker instance no subscriber loaded, restarting, or racing Subscriber never runs
The order really was placed and the event really was queued. Whether it runs late or never runs at all depends entirely on which instance picks up the job and when.

Why it happens

Every one of these is a real gap in the timing and configuration of the queue, not a single bug with one cause:

This is a common source of confusion because nothing in the logs says "event dropped." BullMQ considers a job complete once it stops retrying, whether or not a subscriber actually ran. The only visible symptom is a customer asking where their receipt went, or a warehouse integration that quietly never got the order it was supposed to pack. See the citations at the end for the exact issues and docs.

The key insight

You cannot tell "delayed" from "dropped" just by looking at an order. Both look identical from the order's side, an order with no confirmation email. The only way to tell them apart is to measure elapsed time against the Notification module's own delivery log, which persists every attempted notification regardless of when the underlying event actually ran. If a matching notification exists but arrived late, the event was delayed. If no matching notification exists at all, the event was dropped and needs a human decision, not an automatic fix, because there is no safe idempotent resend API for arbitrary past events and blindly re-emitting risks re-running every subscriber, not just the missing one.

The fix, as a flow

We do not touch checkout and we do not replay raw events automatically. We pull every order placed in a reconciliation window, pull every notification Medusa actually recorded in that window, and diff the two with a pure function that classifies each order as delivered, delayed, or dropped based on whether a matching notification exists and how long it took. Only confirmed drops get written to an audit record. Re-emitting through the workflow engine is a separate, manual, opt-in step gated behind DRY_RUN.

List orders in window created_at, id List notifications resource_id, event_name Diff, pure function delivered, delayed, dropped Status is dropped? yes no, leave alone Audit record, then emitEventStep if approved
The script classifies first. Only a confirmed dropped order becomes a candidate for a manual, opt-in re-emit through the workflow engine.

Build it step by step

1

Get an admin session and the base URL

Point the script at your Medusa backend and an admin user with rights to read orders and notifications. 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.

setup (shell)
pip install requests

export MEDUSA_BACKEND_URL="http://localhost:9000"
export MEDUSA_ADMIN_EMAIL="admin@example.com"
export MEDUSA_ADMIN_PASSWORD="supersecret"
export DRY_RUN="true"   # start safe, only reports the dropped order_id values
setup (shell)
npm install @medusajs/js-sdk

export MEDUSA_BACKEND_URL="http://localhost:9000"
export MEDUSA_ADMIN_EMAIL="admin@example.com"
export MEDUSA_ADMIN_PASSWORD="supersecret"
export DRY_RUN="true"   // start safe, only reports the dropped order_id values
2

Authenticate against the Admin API

Both languages exchange credentials for a token the same way. The Python version talks to the REST route directly with requests. The Node version uses the official @medusajs/js-sdk, which wraps the same login call.

step2.py
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"]
step2.js
import Medusa from "@medusajs/js-sdk";

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;

const sdk = new Medusa({ baseUrl: BASE_URL, auth: { type: "jwt" } });

async function login() {
  await sdk.auth.login("user", "emailpass", { email: EMAIL, password: PASSWORD });
  return sdk;
}
3

List orders and notifications for the reconciliation window

Ask for every order created since the window start, with the fields the diff needs: id, display_id, status, created_at, and fulfillments. Page through with offset and limit until the offset reaches the reported count. In parallel, ask for every notification recorded since the same window start, then keep only the rows where resource_type is order and event_name is order.placed. /admin/notifications persists every delivery attempt regardless of which event actually triggered it, so it tells the truth even when the event bus itself lost the signal.

step3.py
ORDER_FIELDS = "id,display_id,status,created_at,*fulfillments"
NOTIFICATION_FIELDS = "id,to,channel,template,trigger_type,resource_id,resource_type,event_name,original_notification_id,created_at"

def list_orders_since(token, window_start, limit=200):
    headers = {"Authorization": f"Bearer {token}"}
    out, offset = [], 0
    while True:
        r = requests.get(
            f"{BASE_URL}/admin/orders",
            params={
                "created_at[$gte]": window_start,
                "fields": ORDER_FIELDS,
                "limit": limit,
                "offset": offset,
            },
            headers=headers,
            timeout=30,
        )
        r.raise_for_status()
        body = r.json()
        out.extend(body["orders"])
        offset += limit
        if offset >= body["count"]:
            return out


def list_notifications_since(token, window_start, limit=200):
    headers = {"Authorization": f"Bearer {token}"}
    out, offset = [], 0
    while True:
        r = requests.get(
            f"{BASE_URL}/admin/notifications",
            params={
                "created_at[$gte]": window_start,
                "fields": NOTIFICATION_FIELDS,
                "limit": limit,
                "offset": offset,
            },
            headers=headers,
            timeout=30,
        )
        r.raise_for_status()
        body = r.json()
        out.extend(body["notifications"])
        offset += limit
        if offset >= body["count"]:
            return out
step3.js
const ORDER_FIELDS = "id,display_id,status,created_at,*fulfillments";
const NOTIFICATION_FIELDS = "id,to,channel,template,trigger_type,resource_id,resource_type,event_name,original_notification_id,created_at";

async function listOrdersSince(sdk, windowStart, limit = 200) {
  const out = [];
  let offset = 0;
  while (true) {
    const body = await sdk.admin.order.list({
      "created_at[$gte]": windowStart,
      fields: ORDER_FIELDS,
      limit,
      offset,
    });
    out.push(...body.orders);
    offset += limit;
    if (offset >= body.count) return out;
  }
}

async function listNotificationsSince(sdk, windowStart, limit = 200) {
  const out = [];
  let offset = 0;
  while (true) {
    const body = await sdk.client.fetch("/admin/notifications", {
      method: "GET",
      query: { "created_at[$gte]": windowStart, fields: NOTIFICATION_FIELDS, limit, offset },
    });
    out.push(...body.notifications);
    offset += limit;
    if (offset >= body.count) return out;
  }
}
4

Diff and classify, with one pure function

Keep the decision in a function with no network calls. For each order, find the earliest notification with resource_type order, matching resource_id, and event_name order.placed. No match means dropped. A match means delayed or delivered depending on whether the elapsed time between the order and the notification crosses a threshold, sixty seconds by default. This is the only place the classification logic lives, so it is the only place worth testing carefully.

decide.py
from datetime import datetime

def _to_ms(iso):
    return datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp() * 1000


def diff_event_delivery(orders, notifications, window_start, window_end, delay_threshold_ms=60000):
    """Pure: no I/O. orders and notifications are plain dicts/lists already fetched."""
    by_order = {}
    for n in notifications:
        if n.get("resource_type") != "order" or n.get("event_name") != "order.placed":
            continue
        rid = n.get("resource_id")
        ts = _to_ms(n["created_at"])
        if rid not in by_order or ts < by_order[rid]:
            by_order[rid] = ts

    results = []
    for order in orders:
        order_id = order["id"]
        created_ms = _to_ms(order["created_at"])
        match_ms = by_order.get(order_id)
        if match_ms is None:
            results.append({"order_id": order_id, "status": "dropped", "delay_ms": None})
            continue
        delay_ms = match_ms - created_ms
        status = "delayed" if delay_ms > delay_threshold_ms else "delivered"
        results.append({"order_id": order_id, "status": status, "delay_ms": delay_ms})
    return results
decide.js
function toMs(iso) {
  return Date.parse(iso);
}

export function diffEventDelivery(orders, notifications, windowStart, windowEnd, delayThresholdMs = 60000) {
  // Pure: no I/O. orders and notifications are plain arrays already fetched.
  const byOrder = new Map();
  for (const n of notifications) {
    if (n.resource_type !== "order" || n.event_name !== "order.placed") continue;
    const ts = toMs(n.created_at);
    const existing = byOrder.get(n.resource_id);
    if (existing === undefined || ts < existing) byOrder.set(n.resource_id, ts);
  }

  return orders.map((order) => {
    const createdMs = toMs(order.created_at);
    const matchMs = byOrder.get(order.id);
    if (matchMs === undefined) {
      return { order_id: order.id, status: "dropped", delay_ms: null };
    }
    const delayMs = matchMs - createdMs;
    const status = delayMs > delayThresholdMs ? "delayed" : "delivered";
    return { order_id: order.id, status, delay_ms: delayMs };
  });
}
5

Write an audit record for every confirmed drop

Do not mutate the order or the notification record, since there is no safe idempotent resend API for arbitrary past events and re-invoking checkout side effects blindly is not safe. For each order classified dropped, write an audit record, or POST to an internal ops webhook, listing the order id, display id, the expected event, the window, and how long it has been since the order was created. That is the full extent of the automatic behavior. Everything else needs a human.

apply.py
def write_audit_record(order_id, display_id, window_start, window_end, elapsed_ms):
    record = {
        "order_id": order_id,
        "display_id": display_id,
        "expected_event": "order.placed",
        "window_start": window_start,
        "window_end": window_end,
        "elapsed_ms_since_created": elapsed_ms,
    }
    log.warning("DROPPED %s", record)
    return record


def reemit_order_placed(token, order_id):
    """Manual, opt-in only. Re-triggers every subscriber attached to order.placed."""
    headers = {"Authorization": f"Bearer {token}"}
    r = requests.post(
        f"{BASE_URL}/admin/workflows/reemit-order-placed/run",
        json={"input": {"order_id": order_id}},
        headers=headers,
        timeout=30,
    )
    r.raise_for_status()
    return r.json()
apply.js
function writeAuditRecord(orderId, displayId, windowStart, windowEnd, elapsedMs) {
  const record = {
    order_id: orderId,
    display_id: displayId,
    expected_event: "order.placed",
    window_start: windowStart,
    window_end: windowEnd,
    elapsed_ms_since_created: elapsedMs,
  };
  console.warn("DROPPED", record);
  return record;
}

async function reemitOrderPlaced(sdk, orderId) {
  // Manual, opt-in only. Re-triggers every subscriber attached to order.placed.
  return sdk.client.fetch("/admin/workflows/reemit-order-placed/run", {
    method: "POST",
    body: { input: { order_id: orderId } },
  });
}
6

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 writes audit records for confirmed drops and never touches the workflow engine. Read the output, get a human to build a manual confirmation list from it, then switch DRY_RUN off only for that confirmed list to let it call a small custom workflow that runs emitEventStep({ eventName: "order.placed", data: { id: order.id } }) from @medusajs/medusa/core-flows. This must never run unattended, since it re-triggers every subscriber attached to order.placed, including customer emails.

Run it safe

Never mutate an order or a notification record directly, and never re-emit automatically for the whole dropped list. Always start with DRY_RUN=true, always require a human to review the audit records and build an explicit confirmation list, and only then re-emit through the workflow engine for that confirmed list, one order at a time.

The full code

Here is the complete script in one file for each language. It authenticates, pulls orders and notifications for the window, classifies every order with a pure function, writes an audit record for every drop, and only re-emits through the workflow engine when DRY_RUN is false and the order id appears on an explicit, manually built confirmation list.

Get this script on GitHub Follow @allanninal Python and Node.js, with tests. Dry run by default. One of 88 Medusa fixes, free and open source.
reconcile_event_delivery.py
"""Find Medusa v2 orders whose order.placed event never reached its
subscriber because the Redis Event Bus Module (BullMQ) queued the job
before a worker's subscriber-loader finished, or a worker restarted,
autoscaled, or crashed mid-job. Classifies every order in the window as
delivered, delayed, or dropped by diffing against the Notification
module's own delivery log. Never mutates orders or notifications.
DRY_RUN=true only writes audit records for confirmed drops. Re-emitting
through the workflow engine is manual, opt-in, and gated behind an
explicit confirmation list built by a human from the audit output.
"""
import os
import logging
from datetime import datetime, timedelta, timezone

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("reconcile_event_delivery")

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"
WINDOW_HOURS = float(os.environ.get("WINDOW_HOURS", "24"))
DELAY_THRESHOLD_MS = float(os.environ.get("DELAY_THRESHOLD_MS", "60000"))
# Comma-separated order_id values a human has confirmed should be re-emitted.
CONFIRMED_REEMIT_IDS = {
    x.strip() for x in os.environ.get("CONFIRMED_REEMIT_IDS", "").split(",") if x.strip()
}

ORDER_FIELDS = "id,display_id,status,created_at,*fulfillments"
NOTIFICATION_FIELDS = "id,to,channel,template,trigger_type,resource_id,resource_type,event_name,original_notification_id,created_at"


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_orders_since(token, window_start, limit=200):
    headers = {"Authorization": f"Bearer {token}"}
    out, offset = [], 0
    while True:
        r = requests.get(
            f"{BASE_URL}/admin/orders",
            params={
                "created_at[$gte]": window_start,
                "fields": ORDER_FIELDS,
                "limit": limit,
                "offset": offset,
            },
            headers=headers,
            timeout=30,
        )
        r.raise_for_status()
        body = r.json()
        out.extend(body["orders"])
        offset += limit
        if offset >= body["count"]:
            return out


def list_notifications_since(token, window_start, limit=200):
    headers = {"Authorization": f"Bearer {token}"}
    out, offset = [], 0
    while True:
        r = requests.get(
            f"{BASE_URL}/admin/notifications",
            params={
                "created_at[$gte]": window_start,
                "fields": NOTIFICATION_FIELDS,
                "limit": limit,
                "offset": offset,
            },
            headers=headers,
            timeout=30,
        )
        r.raise_for_status()
        body = r.json()
        out.extend(body["notifications"])
        offset += limit
        if offset >= body["count"]:
            return out


def _to_ms(iso):
    return datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp() * 1000


def diff_event_delivery(orders, notifications, window_start, window_end, delay_threshold_ms=60000):
    """Pure: no I/O. orders and notifications are plain dicts/lists already fetched."""
    by_order = {}
    for n in notifications:
        if n.get("resource_type") != "order" or n.get("event_name") != "order.placed":
            continue
        rid = n.get("resource_id")
        ts = _to_ms(n["created_at"])
        if rid not in by_order or ts < by_order[rid]:
            by_order[rid] = ts

    results = []
    for order in orders:
        order_id = order["id"]
        created_ms = _to_ms(order["created_at"])
        match_ms = by_order.get(order_id)
        if match_ms is None:
            results.append({"order_id": order_id, "status": "dropped", "delay_ms": None})
            continue
        delay_ms = match_ms - created_ms
        status = "delayed" if delay_ms > delay_threshold_ms else "delivered"
        results.append({"order_id": order_id, "status": status, "delay_ms": delay_ms})
    return results


def write_audit_record(order_id, display_id, window_start, window_end, elapsed_ms):
    record = {
        "order_id": order_id,
        "display_id": display_id,
        "expected_event": "order.placed",
        "window_start": window_start,
        "window_end": window_end,
        "elapsed_ms_since_created": elapsed_ms,
    }
    log.warning("DROPPED %s", record)
    return record


def reemit_order_placed(token, order_id):
    """Manual, opt-in only. Re-triggers every subscriber attached to order.placed."""
    headers = {"Authorization": f"Bearer {token}"}
    r = requests.post(
        f"{BASE_URL}/admin/workflows/reemit-order-placed/run",
        json={"input": {"order_id": order_id}},
        headers=headers,
        timeout=30,
    )
    r.raise_for_status()
    return r.json()


def run():
    token = get_token()
    now = datetime.now(timezone.utc)
    window_start = (now - timedelta(hours=WINDOW_HOURS)).isoformat()
    window_end = now.isoformat()

    orders = list_orders_since(token, window_start)
    notifications = list_notifications_since(token, window_start)
    by_id = {o["id"]: o for o in orders}

    results = diff_event_delivery(orders, notifications, window_start, window_end, DELAY_THRESHOLD_MS)
    dropped = [r for r in results if r["status"] == "dropped"]
    delayed = [r for r in results if r["status"] == "delayed"]

    log.info(
        "Window %s to %s: %d order(s), %d delivered, %d delayed, %d dropped.",
        window_start, window_end, len(orders), len(results) - len(delayed) - len(dropped),
        len(delayed), len(dropped),
    )

    for item in dropped:
        order = by_id[item["order_id"]]
        elapsed_ms = _to_ms(window_end) - _to_ms(order["created_at"])
        write_audit_record(order["id"], order.get("display_id"), window_start, window_end, elapsed_ms)

    if not DRY_RUN:
        for item in dropped:
            order_id = item["order_id"]
            if order_id not in CONFIRMED_REEMIT_IDS:
                log.info("Order %s dropped but not on the confirmed re-emit list. Skipping.", order_id)
                continue
            log.warning("Order %s: re-emitting order.placed via the workflow engine.", order_id)
            reemit_order_placed(token, order_id)

    log.info("Done. %d dropped order(s) %s.", len(dropped), "audited" if DRY_RUN else "processed")


if __name__ == "__main__":
    run()
reconcile-event-delivery.js
/**
 * Find Medusa v2 orders whose order.placed event never reached its
 * subscriber because the Redis Event Bus Module (BullMQ) queued the job
 * before a worker's subscriber-loader finished, or a worker restarted,
 * autoscaled, or crashed mid-job. Classifies every order in the window as
 * delivered, delayed, or dropped by diffing against the Notification
 * module's own delivery log. Never mutates orders or notifications.
 * DRY_RUN=true only writes audit records for confirmed drops. Re-emitting
 * through the workflow engine is manual, opt-in, and gated behind an
 * explicit confirmation list built by a human from the audit output.
 */
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 WINDOW_HOURS = Number(process.env.WINDOW_HOURS || 24);
const DELAY_THRESHOLD_MS = Number(process.env.DELAY_THRESHOLD_MS || 60000);
// Comma-separated order_id values a human has confirmed should be re-emitted.
const CONFIRMED_REEMIT_IDS = new Set(
  (process.env.CONFIRMED_REEMIT_IDS || "").split(",").map((x) => x.trim()).filter(Boolean)
);

const ORDER_FIELDS = "id,display_id,status,created_at,*fulfillments";
const NOTIFICATION_FIELDS = "id,to,channel,template,trigger_type,resource_id,resource_type,event_name,original_notification_id,created_at";

function toMs(iso) {
  return Date.parse(iso);
}

export function diffEventDelivery(orders, notifications, windowStart, windowEnd, delayThresholdMs = 60000) {
  // Pure: no I/O. orders and notifications are plain arrays already fetched.
  const byOrder = new Map();
  for (const n of notifications) {
    if (n.resource_type !== "order" || n.event_name !== "order.placed") continue;
    const ts = toMs(n.created_at);
    const existing = byOrder.get(n.resource_id);
    if (existing === undefined || ts < existing) byOrder.set(n.resource_id, ts);
  }

  return orders.map((order) => {
    const createdMs = toMs(order.created_at);
    const matchMs = byOrder.get(order.id);
    if (matchMs === undefined) {
      return { order_id: order.id, status: "dropped", delay_ms: null };
    }
    const delayMs = matchMs - createdMs;
    const status = delayMs > delayThresholdMs ? "delayed" : "delivered";
    return { order_id: order.id, status, delay_ms: delayMs };
  });
}

async function login() {
  const { default: Medusa } = await import("@medusajs/js-sdk");
  const sdk = new Medusa({ baseUrl: BASE_URL, auth: { type: "jwt" } });
  await sdk.auth.login("user", "emailpass", { email: EMAIL, password: PASSWORD });
  return sdk;
}

async function listOrdersSince(sdk, windowStart, limit = 200) {
  const out = [];
  let offset = 0;
  while (true) {
    const body = await sdk.admin.order.list({
      "created_at[$gte]": windowStart,
      fields: ORDER_FIELDS,
      limit,
      offset,
    });
    out.push(...body.orders);
    offset += limit;
    if (offset >= body.count) return out;
  }
}

async function listNotificationsSince(sdk, windowStart, limit = 200) {
  const out = [];
  let offset = 0;
  while (true) {
    const body = await sdk.client.fetch("/admin/notifications", {
      method: "GET",
      query: { "created_at[$gte]": windowStart, fields: NOTIFICATION_FIELDS, limit, offset },
    });
    out.push(...body.notifications);
    offset += limit;
    if (offset >= body.count) return out;
  }
}

function writeAuditRecord(orderId, displayId, windowStart, windowEnd, elapsedMs) {
  const record = {
    order_id: orderId,
    display_id: displayId,
    expected_event: "order.placed",
    window_start: windowStart,
    window_end: windowEnd,
    elapsed_ms_since_created: elapsedMs,
  };
  console.warn("DROPPED", record);
  return record;
}

async function reemitOrderPlaced(sdk, orderId) {
  // Manual, opt-in only. Re-triggers every subscriber attached to order.placed.
  return sdk.client.fetch("/admin/workflows/reemit-order-placed/run", {
    method: "POST",
    body: { input: { order_id: orderId } },
  });
}

export async function run() {
  const sdk = await login();
  const now = new Date();
  const windowStart = new Date(now.getTime() - WINDOW_HOURS * 3600 * 1000).toISOString();
  const windowEnd = now.toISOString();

  const orders = await listOrdersSince(sdk, windowStart);
  const notifications = await listNotificationsSince(sdk, windowStart);
  const byId = new Map(orders.map((o) => [o.id, o]));

  const results = diffEventDelivery(orders, notifications, windowStart, windowEnd, DELAY_THRESHOLD_MS);
  const dropped = results.filter((r) => r.status === "dropped");
  const delayed = results.filter((r) => r.status === "delayed");

  console.log(
    `Window ${windowStart} to ${windowEnd}: ${orders.length} order(s), ${results.length - delayed.length - dropped.length} delivered, ${delayed.length} delayed, ${dropped.length} dropped.`
  );

  for (const item of dropped) {
    const order = byId.get(item.order_id);
    const elapsedMs = toMs(windowEnd) - toMs(order.created_at);
    writeAuditRecord(order.id, order.display_id, windowStart, windowEnd, elapsedMs);
  }

  if (!DRY_RUN) {
    for (const item of dropped) {
      if (!CONFIRMED_REEMIT_IDS.has(item.order_id)) {
        console.log(`Order ${item.order_id} dropped but not on the confirmed re-emit list. Skipping.`);
        continue;
      }
      console.warn(`Order ${item.order_id}: re-emitting order.placed via the workflow engine.`);
      await reemitOrderPlaced(sdk, item.order_id);
    }
  }

  console.log(`Done. ${dropped.length} dropped order(s) ${DRY_RUN ? "audited" : "processed"}.`);
}

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, diff_event_delivery. It is pure, no network and no database, so the tests feed in plain order and notification arrays with a fixed clock and check the classification.

test_redis_event_diff.py
from reconcile_event_delivery import diff_event_delivery

WINDOW_START = "2026-07-09T00:00:00Z"
WINDOW_END = "2026-07-10T00:00:00Z"


def order(**over):
    base = {"id": "order_1", "created_at": "2026-07-09T12:00:00Z"}
    base.update(over)
    return base


def notification(**over):
    base = {
        "resource_id": "order_1",
        "resource_type": "order",
        "event_name": "order.placed",
        "created_at": "2026-07-09T12:00:10Z",
    }
    base.update(over)
    return base


def test_delivered_when_notification_arrives_quickly():
    result = diff_event_delivery([order()], [notification()], WINDOW_START, WINDOW_END)
    assert result == [{"order_id": "order_1", "status": "delivered", "delay_ms": 10000}]


def test_delayed_when_notification_arrives_past_threshold():
    late = notification(created_at="2026-07-09T12:05:00Z")
    result = diff_event_delivery([order()], [late], WINDOW_START, WINDOW_END, delay_threshold_ms=60000)
    assert result == [{"order_id": "order_1", "status": "delayed", "delay_ms": 300000}]


def test_dropped_when_no_matching_notification():
    result = diff_event_delivery([order()], [], WINDOW_START, WINDOW_END)
    assert result == [{"order_id": "order_1", "status": "dropped", "delay_ms": None}]


def test_ignores_notification_for_a_different_event():
    other_event = notification(event_name="order.fulfillment_created")
    result = diff_event_delivery([order()], [other_event], WINDOW_START, WINDOW_END)
    assert result == [{"order_id": "order_1", "status": "dropped", "delay_ms": None}]


def test_ignores_notification_for_a_different_resource_type():
    other_type = notification(resource_type="customer")
    result = diff_event_delivery([order()], [other_type], WINDOW_START, WINDOW_END)
    assert result == [{"order_id": "order_1", "status": "dropped", "delay_ms": None}]


def test_uses_the_earliest_matching_notification():
    first = notification(created_at="2026-07-09T12:00:05Z")
    second = notification(created_at="2026-07-09T12:10:00Z")
    result = diff_event_delivery([order()], [second, first], WINDOW_START, WINDOW_END)
    assert result == [{"order_id": "order_1", "status": "delivered", "delay_ms": 5000}]


def test_handles_multiple_orders_independently():
    orders = [order(), order(id="order_2", created_at="2026-07-09T13:00:00Z")]
    notifications = [notification()]
    result = diff_event_delivery(orders, notifications, WINDOW_START, WINDOW_END)
    assert result == [
        {"order_id": "order_1", "status": "delivered", "delay_ms": 10000},
        {"order_id": "order_2", "status": "dropped", "delay_ms": None},
    ]


def test_exactly_at_threshold_is_delivered():
    at_threshold = notification(created_at="2026-07-09T12:01:00Z")
    result = diff_event_delivery([order()], [at_threshold], WINDOW_START, WINDOW_END, delay_threshold_ms=60000)
    assert result == [{"order_id": "order_1", "status": "delivered", "delay_ms": 60000}]
redis-event-diff.test.js
import { test } from "node:test";
import assert from "node:assert/strict";
import { diffEventDelivery } from "./reconcile-event-delivery.js";

const WINDOW_START = "2026-07-09T00:00:00Z";
const WINDOW_END = "2026-07-10T00:00:00Z";

const order = (over = {}) => ({
  id: "order_1",
  created_at: "2026-07-09T12:00:00Z",
  ...over,
});

const notification = (over = {}) => ({
  resource_id: "order_1",
  resource_type: "order",
  event_name: "order.placed",
  created_at: "2026-07-09T12:00:10Z",
  ...over,
});

test("delivered when notification arrives quickly", () => {
  const result = diffEventDelivery([order()], [notification()], WINDOW_START, WINDOW_END);
  assert.deepEqual(result, [{ order_id: "order_1", status: "delivered", delay_ms: 10000 }]);
});

test("delayed when notification arrives past threshold", () => {
  const late = notification({ created_at: "2026-07-09T12:05:00Z" });
  const result = diffEventDelivery([order()], [late], WINDOW_START, WINDOW_END, 60000);
  assert.deepEqual(result, [{ order_id: "order_1", status: "delayed", delay_ms: 300000 }]);
});

test("dropped when no matching notification", () => {
  const result = diffEventDelivery([order()], [], WINDOW_START, WINDOW_END);
  assert.deepEqual(result, [{ order_id: "order_1", status: "dropped", delay_ms: null }]);
});

test("ignores notification for a different event", () => {
  const otherEvent = notification({ event_name: "order.fulfillment_created" });
  const result = diffEventDelivery([order()], [otherEvent], WINDOW_START, WINDOW_END);
  assert.deepEqual(result, [{ order_id: "order_1", status: "dropped", delay_ms: null }]);
});

test("ignores notification for a different resource type", () => {
  const otherType = notification({ resource_type: "customer" });
  const result = diffEventDelivery([order()], [otherType], WINDOW_START, WINDOW_END);
  assert.deepEqual(result, [{ order_id: "order_1", status: "dropped", delay_ms: null }]);
});

test("uses the earliest matching notification", () => {
  const first = notification({ created_at: "2026-07-09T12:00:05Z" });
  const second = notification({ created_at: "2026-07-09T12:10:00Z" });
  const result = diffEventDelivery([order()], [second, first], WINDOW_START, WINDOW_END);
  assert.deepEqual(result, [{ order_id: "order_1", status: "delivered", delay_ms: 5000 }]);
});

test("handles multiple orders independently", () => {
  const orders = [order(), order({ id: "order_2", created_at: "2026-07-09T13:00:00Z" })];
  const notifications = [notification()];
  const result = diffEventDelivery(orders, notifications, WINDOW_START, WINDOW_END);
  assert.deepEqual(result, [
    { order_id: "order_1", status: "delivered", delay_ms: 10000 },
    { order_id: "order_2", status: "dropped", delay_ms: null },
  ]);
});

test("exactly at threshold is delivered", () => {
  const atThreshold = notification({ created_at: "2026-07-09T12:01:00Z" });
  const result = diffEventDelivery([order()], [atThreshold], WINDOW_START, WINDOW_END, 60000);
  assert.deepEqual(result, [{ order_id: "order_1", status: "delivered", delay_ms: 60000 }]);
});

Case studies

Autoscaling worker pool

The confirmation emails delayed by a scaling event

A store running a Black Friday promotion had its worker pool autoscale from two instances to six in the middle of the traffic spike. Each new instance took a few seconds to finish loading Medusa's modules and register its subscribers, but BullMQ started handing them jobs the moment they connected to Redis.

Running the reconciler against that afternoon's orders found no orders truly dropped, but eleven were classified delayed, with a matching order.placed notification arriving between forty and ninety seconds after the order was created. That matched exactly the startup window of the new instances. Nobody needed to re-emit anything, since every one of those orders eventually got its confirmation email; the classification just confirmed the delay was cosmetic and not a data-loss problem.

Worker crash mid-job

The batch that never came back after a bad deploy

A deploy introduced a bug that crashed the worker process on a specific order shape. Four orders placed in the ten minutes before the crash had their order.placed jobs picked up, attempted, and failed against the broken worker, then discarded once they exhausted their retry attempts because removeOnFail was set aggressively to keep the queue small.

The reconciler classified all four as dropped, with zero matching notifications even hours later. The team reviewed the four order_id values, confirmed none of them had a notification appear since the scan ran, and manually approved a re-emit through the workflow engine for exactly those four orders once the underlying bug was fixed and DRY_RUN was switched off for that confirmed list only.

What good looks like

Run this reconciler on a schedule, especially after a deploy or a scaling event. It never mutates an order or a notification and never re-emits automatically, so it can never send a customer two confirmation emails or paper over a real gap. It only tells you, with elapsed time attached, which orders were delivered on time, which were merely delayed, and which never got their order.placed notification at all. Re-emitting through emitEventStep stays a manual, opt-in step gated behind DRY_RUN and an explicit confirmation list built by a human, because that is what keeps the fix honest.

FAQ

Why does order.placed sometimes never reach my subscriber in Medusa?

Medusa v2's default Redis Event Bus Module is backed by BullMQ, which queues every emitted event as a job for a worker process to consume. If the instance that owns the subscriber registration starts after the event was already published, restarts mid-job, or is one of several autoscaled instances racing to consume the same queue, the job can be picked up with no matching subscriber attached, retried past its attempt limit, or dropped depending on the queue's retry and removeOnFail settings. Nothing throws an error, because from BullMQ's point of view the job was still processed, so the gap only shows up as a customer asking why they never got a confirmation email.

How is a delayed event different from a dropped event?

A delayed event still arrives, just later than expected, usually because the job waited for a retry or for a worker to finish booting before a matching subscriber picked it up. A dropped event never arrives at all, because the job was retried past its attempt limit and discarded, or removeOnFail cleared it before anything could reprocess it. Both look identical from the order's perspective, an order with no confirmation email, so you have to measure the elapsed time between the order and the matching notification to tell them apart.

Is it safe to just re-emit order.placed for orders that look affected?

Not blindly. Re-emitting a raw event re-triggers every subscriber attached to it, not just the one whose side effect is missing, which risks a second confirmation email or a duplicate inventory action for orders that were only delayed, not dropped. The safe pattern is to first classify each order with a pure diff against the Notification module's own delivery log, confirm which ones are truly dropped rather than delayed, and only then re-emit through emitEventStep for the confirmed list, gated behind a dry run and manual approval.

Related field notes

Shopify orders stuck Unpaid after a payment made outside checkout Shopify orders stuck Unfulfilled past your shipping SLA Browse all Medusa field notes

Citations

On the problem:

  1. medusajs/medusa GitHub issue #7850: Events in Redis event bus not triggering subscribers (notification service). github.com/medusajs/medusa/issues/7850
  2. medusajs/medusa GitHub issue #4089: Events not triggered consistently using Redis EventBusModule. github.com/medusajs/medusa/issues/4089
  3. medusajs/medusa GitHub issue #7223: Events being lost in worker mode. github.com/medusajs/medusa/issues/7223

On the solution:

  1. Medusa Documentation: Events and Subscribers. docs.medusajs.com/learn/fundamentals/events-and-subscribers
  2. Medusa Documentation: Emit Workflow and Service Events (emitEventStep). docs.medusajs.com/learn/fundamentals/events-and-subscribers/emit-event
  3. Medusa Documentation: Event Processing Priority. docs.medusajs.com/learn/fundamentals/events-and-subscribers/event-priority

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.

Contact me on LinkedIn

Did this catch a silently dropped event?

If this saved you from a customer support ticket or a confusing missing email, 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

Back to all Medusa field notes