Diagnostic Message Queue and Scheduled Tasks
Failed messages accumulate in the messenger failure queue
A webhook payload comes in malformed, an indexing job times out, a third party API in a flow action errors out. Symfony Messenger retries the message a few times with backoff, and when it still fails it quietly moves into the separate failed transport. Almost nobody runs a worker against that queue, so the failures just sit there, growing the messenger_messages table and hiding real recurring problems behind an ever bigger pile. Here is why it happens and a small script that flags the backlog before it gets out of hand.
Shopware 6 runs on Symfony Messenger. A message that throws during processing is retried automatically (three attempts with backoff by default), and if it still fails it is moved into the transport configured by MESSENGER_TRANSPORT_FAILURE_DSN, which defaults to doctrine://default?queue_name=failed. That is the same messenger_messages Doctrine table, just with queue_name='failed'. Shopware's own docs warn that you must run a CLI worker against that failed queue separately, and most installs never do, so the run a small Python or Node.js script that polls GET /api/_info/message-stats.json, reads the size of the transport named failed, tracks whether it is growing between polls, and reports it. It does not retry or delete anything automatically. Full code, tests, and the reasoning are below.
The problem in plain words
Every background job in Shopware, sending an order confirmation email, running a product indexer, delivering a flow action, calling a third party API, goes through Symfony Messenger. Messenger's whole design assumes some messages will fail, so it retries them a few times with a growing delay between attempts.
The part that trips up almost every store is what happens after the retries run out. The message does not vanish, and it is not requeued forever. It gets moved into a second, separate transport, the failed transport, so that a stuck message stops blocking the queue behind it. That is good design. The trouble is that this failed transport needs its own worker to ever be looked at again, and the standard setup most people copy from a tutorial only runs messenger:consume against default and async. The failed queue is left completely unconsumed, so every rejected message, a bad webhook body, a timed out indexing job, a database deadlock, a third party API error inside a flow action, just accumulates there forever.
Why it happens
Symfony Messenger's failure handling is correct by design. The surprise is entirely in what Shopware installs typically run in production:
- The Shopware docs run a worker like
bin/console messenger:consume default async low_priority, and it is easy to copy that command and never notice the failed transport was never named. - Shopware's documentation says explicitly that you must set up a CLI worker for the failed queue too, otherwise failed messages are never processed, but that line is easy to miss in a hosting guide read once during setup.
- There is no Shopware Admin API entity for individual failed messages.
messenger_messagesis a raw Symfony transport table, not a Shopware entity with a/api/{entity}route, so there is nothing to search or page through the normal way. - Bad webhook payloads, indexing jobs that time out, transient database deadlocks, and third party API errors thrown from flow actions all land in the same bucket, so a growing failed queue can hide one urgent recurring bug inside a hundred harmless one-off retries.
The result is a table that quietly grows in the database while every dashboard looks fine, because dashboards usually watch the live queues, not the failure transport sitting off to the side.
You cannot safely automate the fix here, only the detection. Symfony and Shopware only expose retry and purge through the CLI or direct database access, and a generic script that blindly retries arbitrary failed jobs can cause duplicate side effects, a second confirmation email, a second stock adjustment, because it has no way to know if the original attempt partially succeeded. So the right shaped tool is a reporter: read the failed transport's size and its growth trend, corroborate with scheduled task health, and hand a clear recommendation to a human who runs messenger:failed:retry or messenger:failed:remove deliberately.
The fix, as a flow
The script authenticates once, polls the message stats endpoint, and compares the failed transport's current size against the previous poll and against a threshold. It also checks the scheduled_task list for tasks stuck in queued with a next execution time far in the past, a corroborating sign that no worker is consuming anything, failed queue included. It never writes. It only decides ok, warn, or critical and prints the reasons.
Build it step by step
Get Admin API credentials
Create an Integration in Settings, System, Integrations and note the access key id and secret access key. Keep the shop URL and both values in environment variables, never in the file. Start with DRY_RUN on, though for this script mutation is always effectively off since it never writes.
pip install requests
export SHOPWARE_URL="https://your-shop.example.com"
export SHOPWARE_CLIENT_ID="your access key id"
export SHOPWARE_CLIENT_SECRET="your secret access key"
export FAILED_THRESHOLD="0"
export STALE_TASK_AGE_SECONDS="900"
export DRY_RUN="true" # this script only ever reports, never writes// Node 18+ has fetch built in, no dependencies needed
export SHOPWARE_URL="https://your-shop.example.com"
export SHOPWARE_CLIENT_ID="your access key id"
export SHOPWARE_CLIENT_SECRET="your secret access key"
export FAILED_THRESHOLD="0"
export STALE_TASK_AGE_SECONDS="900"
export DRY_RUN="true" // this script only ever reports, never writesAuthenticate against the Admin API
Exchange the client id and secret for a bearer token at POST /api/oauth/token with grant_type: client_credentials. Every later call sends that token in the Authorization header along with Accept: application/json.
import os, requests
SHOPWARE_URL = os.environ["SHOPWARE_URL"].rstrip("/")
CLIENT_ID = os.environ["SHOPWARE_CLIENT_ID"]
CLIENT_SECRET = os.environ["SHOPWARE_CLIENT_SECRET"]
def get_token():
r = requests.post(
f"{SHOPWARE_URL}/api/oauth/token",
json={
"grant_type": "client_credentials",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
},
timeout=30,
)
r.raise_for_status()
return r.json()["access_token"]const SHOPWARE_URL = (process.env.SHOPWARE_URL || "").replace(/\/$/, "");
const CLIENT_ID = process.env.SHOPWARE_CLIENT_ID;
const CLIENT_SECRET = process.env.SHOPWARE_CLIENT_SECRET;
async function getToken() {
const res = await fetch(`${SHOPWARE_URL}/api/oauth/token`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
grant_type: "client_credentials",
client_id: CLIENT_ID,
client_secret: CLIENT_SECRET,
}),
});
if (!res.ok) throw new Error(`Shopware auth ${res.status}`);
const body = await res.json();
return body.access_token;
}Read the failed transport size
Call GET /api/_info/message-stats.json. It returns per-transport stats keyed by name, such as default, async, low_priority, and failed. Read the size for the transport named failed. The older GET /api/_info/queue.json route is deprecated as of 6.7.8.0, so use the stats route.
def get_failed_transport_size(token):
r = requests.get(
f"{SHOPWARE_URL}/api/_info/message-stats.json",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
timeout=30,
)
r.raise_for_status()
stats = r.json()
# stats is a list or dict of transports; normalize to a list of entries
entries = stats if isinstance(stats, list) else stats.get("transports", stats)
for entry in entries if isinstance(entries, list) else entries.values():
name = entry.get("name") or entry.get("queue_name") or entry.get("queueName")
if name == "failed":
return int(entry.get("size", 0))
return 0async function getFailedTransportSize(token) {
const res = await fetch(`${SHOPWARE_URL}/api/_info/message-stats.json`, {
headers: { Authorization: `Bearer ${token}`, Accept: "application/json" },
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const stats = await res.json();
const entries = Array.isArray(stats) ? stats : Object.values(stats.transports || stats);
for (const entry of entries) {
const name = entry.name || entry.queue_name || entry.queueName;
if (name === "failed") return Number(entry.size || 0);
}
return 0;
}Check for a stale queued scheduled task
Cross-check with POST /api/search/scheduled-task, reading each task's status and nextExecutionTime. A task stuck in queued with a next execution time far in the past is a corroborating symptom that no worker is consuming any transport, the failed one included.
import datetime
def get_oldest_queued_task_age_seconds(token):
r = requests.post(
f"{SHOPWARE_URL}/api/search/scheduled-task",
json={"filter": [{"type": "equals", "field": "status", "value": "queued"}]},
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/json",
"Content-Type": "application/json",
},
timeout=30,
)
r.raise_for_status()
tasks = r.json().get("data", [])
if not tasks:
return None
now = datetime.datetime.now(datetime.timezone.utc)
oldest = None
for task in tasks:
attrs = task.get("attributes", task)
next_exec = attrs.get("nextExecutionTime")
if not next_exec:
continue
when = datetime.datetime.fromisoformat(next_exec.replace("Z", "+00:00"))
age = (now - when).total_seconds()
if oldest is None or age > oldest:
oldest = age
return oldestasync function getOldestQueuedTaskAgeSeconds(token) {
const res = await fetch(`${SHOPWARE_URL}/api/search/scheduled-task`, {
method: "POST",
headers: {
Authorization: `Bearer ${token}`,
Accept: "application/json",
"Content-Type": "application/json",
},
body: JSON.stringify({ filter: [{ type: "equals", field: "status", value: "queued" }] }),
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const body = await res.json();
const tasks = body.data || [];
if (tasks.length === 0) return null;
const now = Date.now();
let oldest = null;
for (const task of tasks) {
const attrs = task.attributes || task;
const nextExec = attrs.nextExecutionTime;
if (!nextExec) continue;
const age = (now - Date.parse(nextExec)) / 1000;
if (oldest === null || age > oldest) oldest = age;
}
return oldest;
}Decide, with one pure function
The decision has no I/O in it at all. It takes the current failed size, the previous failed size from the last poll, a threshold count, the oldest queued task age if any, and a staleness threshold, and returns a status of ok, warn, or critical along with the reasons. Critical means the backlog is both over the threshold and growing. Warn means it is over the threshold but not growing, or a scheduled task looks stuck even if the failed count is small.
def evaluate_failed_queue_backlog(
current_failed_size,
previous_failed_size,
threshold_count,
oldest_queued_task_age_seconds,
stale_task_age_threshold_seconds,
):
growing = previous_failed_size is None or current_failed_size > previous_failed_size
reasons = []
over_threshold = current_failed_size > threshold_count
task_is_stale = (
oldest_queued_task_age_seconds is not None
and oldest_queued_task_age_seconds > stale_task_age_threshold_seconds
)
if over_threshold:
reasons.append(
f"failed transport has {current_failed_size} message(s), over threshold {threshold_count}"
)
if growing:
reasons.append(
f"backlog is growing (previous {previous_failed_size}, current {current_failed_size})"
)
if task_is_stale:
reasons.append(
f"oldest queued scheduled task is {int(oldest_queued_task_age_seconds)}s "
f"past its next execution time (threshold {stale_task_age_threshold_seconds}s)"
)
if over_threshold and growing:
status = "critical"
elif over_threshold or task_is_stale:
status = "warn"
else:
status = "ok"
return {"status": status, "growing": growing, "reasons": reasons}export function evaluateFailedQueueBacklog(
currentFailedSize,
previousFailedSize,
thresholdCount,
oldestQueuedTaskAgeSeconds,
staleTaskAgeThresholdSeconds
) {
const growing = previousFailedSize === null || currentFailedSize > previousFailedSize;
const reasons = [];
const overThreshold = currentFailedSize > thresholdCount;
const taskIsStale =
oldestQueuedTaskAgeSeconds !== null &&
oldestQueuedTaskAgeSeconds > staleTaskAgeThresholdSeconds;
if (overThreshold) {
reasons.push(`failed transport has ${currentFailedSize} message(s), over threshold ${thresholdCount}`);
if (growing) {
reasons.push(`backlog is growing (previous ${previousFailedSize}, current ${currentFailedSize})`);
}
}
if (taskIsStale) {
reasons.push(
`oldest queued scheduled task is ${Math.floor(oldestQueuedTaskAgeSeconds)}s past its next execution time (threshold ${staleTaskAgeThresholdSeconds}s)`
);
}
let status;
if (overThreshold && growing) status = "critical";
else if (overThreshold || taskIsStale) status = "warn";
else status = "ok";
return { status, growing, reasons };
}Wire it together and persist the last size
The loop authenticates, reads the failed transport size and the oldest stale task age, loads the previous size from a small state file, evaluates the decision, and logs the status and reasons. It then saves the current size for next time. There is no write path to the Admin API at all here, retrying or purging failed messages stays a manual, deliberate CLI step.
This script never calls messenger:failed:retry or messenger:failed:remove for you, and it cannot, those exist only on the CLI or through direct database access. When it reports critical, run bin/console messenger:failed:show yourself, look at what actually failed, and decide case by case. If the backlog is chronic rather than a one-off spike, stand up a dedicated messenger:consume failed worker instead of clearing it by hand every time.
The full code
Here is the complete script in one file for each language. It reads settings from the environment, authenticates, polls the failed transport size and scheduled task health, evaluates the pure decision function, and prints a report. It keeps its last-seen size in a small local state file so it can tell whether the backlog is growing between runs.
View this code on GitHub Full runnable folder with tests in the shopware-fixes repo.
"""Report on failed messages piling up in Shopware's messenger failure queue.
Symfony Messenger retries a failing message a few times, and if every retry fails
it moves the message into the failed transport (MESSENGER_TRANSPORT_FAILURE_DSN).
Most installs never run a worker against that transport, so it grows silently.
This script only detects and reports the backlog. It never retries or deletes
anything, that stays a deliberate, manual CLI step.
Run on a schedule. Safe to run again and again.
"""
import os
import json
import logging
import datetime
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("messenger_failed_queue_backlog")
SHOPWARE_URL = os.environ["SHOPWARE_URL"].rstrip("/")
CLIENT_ID = os.environ["SHOPWARE_CLIENT_ID"]
CLIENT_SECRET = os.environ["SHOPWARE_CLIENT_SECRET"]
FAILED_THRESHOLD = int(os.environ.get("FAILED_THRESHOLD", "0"))
STALE_TASK_AGE_SECONDS = float(os.environ.get("STALE_TASK_AGE_SECONDS", "900"))
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"
STATE_FILE = os.environ.get("STATE_FILE", ".messenger_failed_queue_backlog_state.json")
def get_token():
r = requests.post(
f"{SHOPWARE_URL}/api/oauth/token",
json={
"grant_type": "client_credentials",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
},
timeout=30,
)
r.raise_for_status()
return r.json()["access_token"]
def get_failed_transport_size(token):
r = requests.get(
f"{SHOPWARE_URL}/api/_info/message-stats.json",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
timeout=30,
)
r.raise_for_status()
stats = r.json()
entries = stats if isinstance(stats, list) else stats.get("transports", stats)
for entry in entries if isinstance(entries, list) else entries.values():
name = entry.get("name") or entry.get("queue_name") or entry.get("queueName")
if name == "failed":
return int(entry.get("size", 0))
return 0
def get_oldest_queued_task_age_seconds(token):
r = requests.post(
f"{SHOPWARE_URL}/api/search/scheduled-task",
json={"filter": [{"type": "equals", "field": "status", "value": "queued"}]},
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/json",
"Content-Type": "application/json",
},
timeout=30,
)
r.raise_for_status()
tasks = r.json().get("data", [])
if not tasks:
return None
now = datetime.datetime.now(datetime.timezone.utc)
oldest = None
for task in tasks:
attrs = task.get("attributes", task)
next_exec = attrs.get("nextExecutionTime")
if not next_exec:
continue
when = datetime.datetime.fromisoformat(next_exec.replace("Z", "+00:00"))
age = (now - when).total_seconds()
if oldest is None or age > oldest:
oldest = age
return oldest
def evaluate_failed_queue_backlog(
current_failed_size,
previous_failed_size,
threshold_count,
oldest_queued_task_age_seconds,
stale_task_age_threshold_seconds,
):
growing = previous_failed_size is None or current_failed_size > previous_failed_size
reasons = []
over_threshold = current_failed_size > threshold_count
task_is_stale = (
oldest_queued_task_age_seconds is not None
and oldest_queued_task_age_seconds > stale_task_age_threshold_seconds
)
if over_threshold:
reasons.append(
f"failed transport has {current_failed_size} message(s), over threshold {threshold_count}"
)
if growing:
reasons.append(
f"backlog is growing (previous {previous_failed_size}, current {current_failed_size})"
)
if task_is_stale:
reasons.append(
f"oldest queued scheduled task is {int(oldest_queued_task_age_seconds)}s "
f"past its next execution time (threshold {stale_task_age_threshold_seconds}s)"
)
if over_threshold and growing:
status = "critical"
elif over_threshold or task_is_stale:
status = "warn"
else:
status = "ok"
return {"status": status, "growing": growing, "reasons": reasons}
def load_previous_size():
try:
with open(STATE_FILE, "r") as f:
return json.load(f).get("failed_size")
except (FileNotFoundError, json.JSONDecodeError):
return None
def save_current_size(size):
with open(STATE_FILE, "w") as f:
json.dump({"failed_size": size}, f)
def run():
token = get_token()
current_size = get_failed_transport_size(token)
previous_size = load_previous_size()
oldest_age = get_oldest_queued_task_age_seconds(token)
result = evaluate_failed_queue_backlog(
current_size, previous_size, FAILED_THRESHOLD, oldest_age, STALE_TASK_AGE_SECONDS
)
log.info(
"Failed transport size=%d previous=%s status=%s growing=%s",
current_size, previous_size, result["status"], result["growing"],
)
for reason in result["reasons"]:
log.warning("Reason: %s", reason)
if result["status"] in ("warn", "critical"):
log.warning(
"Run 'bin/console messenger:failed:show' and either "
"'messenger:failed:retry' or 'messenger:failed:remove' manually, "
"or stand up a dedicated 'messenger:consume failed' worker if this is chronic."
)
if not DRY_RUN:
save_current_size(current_size)
else:
log.info("DRY_RUN is on, not persisting state for the next comparison.")
return result
if __name__ == "__main__":
run()/**
* Report on failed messages piling up in Shopware's messenger failure queue.
*
* Symfony Messenger retries a failing message a few times, and if every retry fails
* it moves the message into the failed transport (MESSENGER_TRANSPORT_FAILURE_DSN).
* Most installs never run a worker against that transport, so it grows silently.
* This script only detects and reports the backlog. It never retries or deletes
* anything, that stays a deliberate, manual CLI step.
* Run on a schedule. Safe to run again and again.
*
* Guide: https://www.allanninal.dev/shopware/messenger-failed-queue-backlog/
*/
import { pathToFileURL } from "node:url";
import { readFile, writeFile } from "node:fs/promises";
const SHOPWARE_URL = (process.env.SHOPWARE_URL || "https://demo.example.com").replace(/\/$/, "");
const CLIENT_ID = process.env.SHOPWARE_CLIENT_ID || "SWIAxxx";
const CLIENT_SECRET = process.env.SHOPWARE_CLIENT_SECRET || "secret_dummy";
const FAILED_THRESHOLD = Number(process.env.FAILED_THRESHOLD || 0);
const STALE_TASK_AGE_SECONDS = Number(process.env.STALE_TASK_AGE_SECONDS || 900);
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
const STATE_FILE = process.env.STATE_FILE || ".messenger_failed_queue_backlog_state.json";
export function evaluateFailedQueueBacklog(
currentFailedSize,
previousFailedSize,
thresholdCount,
oldestQueuedTaskAgeSeconds,
staleTaskAgeThresholdSeconds
) {
const growing = previousFailedSize === null || currentFailedSize > previousFailedSize;
const reasons = [];
const overThreshold = currentFailedSize > thresholdCount;
const taskIsStale =
oldestQueuedTaskAgeSeconds !== null &&
oldestQueuedTaskAgeSeconds > staleTaskAgeThresholdSeconds;
if (overThreshold) {
reasons.push(`failed transport has ${currentFailedSize} message(s), over threshold ${thresholdCount}`);
if (growing) {
reasons.push(`backlog is growing (previous ${previousFailedSize}, current ${currentFailedSize})`);
}
}
if (taskIsStale) {
reasons.push(
`oldest queued scheduled task is ${Math.floor(oldestQueuedTaskAgeSeconds)}s past its next execution time (threshold ${staleTaskAgeThresholdSeconds}s)`
);
}
let status;
if (overThreshold && growing) status = "critical";
else if (overThreshold || taskIsStale) status = "warn";
else status = "ok";
return { status, growing, reasons };
}
async function getToken() {
const res = await fetch(`${SHOPWARE_URL}/api/oauth/token`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
grant_type: "client_credentials",
client_id: CLIENT_ID,
client_secret: CLIENT_SECRET,
}),
});
if (!res.ok) throw new Error(`Shopware auth ${res.status}`);
const body = await res.json();
return body.access_token;
}
async function getFailedTransportSize(token) {
const res = await fetch(`${SHOPWARE_URL}/api/_info/message-stats.json`, {
headers: { Authorization: `Bearer ${token}`, Accept: "application/json" },
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const stats = await res.json();
const entries = Array.isArray(stats) ? stats : Object.values(stats.transports || stats);
for (const entry of entries) {
const name = entry.name || entry.queue_name || entry.queueName;
if (name === "failed") return Number(entry.size || 0);
}
return 0;
}
async function getOldestQueuedTaskAgeSeconds(token) {
const res = await fetch(`${SHOPWARE_URL}/api/search/scheduled-task`, {
method: "POST",
headers: {
Authorization: `Bearer ${token}`,
Accept: "application/json",
"Content-Type": "application/json",
},
body: JSON.stringify({ filter: [{ type: "equals", field: "status", value: "queued" }] }),
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const body = await res.json();
const tasks = body.data || [];
if (tasks.length === 0) return null;
const now = Date.now();
let oldest = null;
for (const task of tasks) {
const attrs = task.attributes || task;
const nextExec = attrs.nextExecutionTime;
if (!nextExec) continue;
const age = (now - Date.parse(nextExec)) / 1000;
if (oldest === null || age > oldest) oldest = age;
}
return oldest;
}
async function loadPreviousSize() {
try {
const raw = await readFile(STATE_FILE, "utf8");
const parsed = JSON.parse(raw);
return typeof parsed.failedSize === "number" ? parsed.failedSize : null;
} catch {
return null;
}
}
async function saveCurrentSize(size) {
await writeFile(STATE_FILE, JSON.stringify({ failedSize: size }));
}
export async function run() {
const token = await getToken();
const currentSize = await getFailedTransportSize(token);
const previousSize = await loadPreviousSize();
const oldestAge = await getOldestQueuedTaskAgeSeconds(token);
const result = evaluateFailedQueueBacklog(
currentSize, previousSize, FAILED_THRESHOLD, oldestAge, STALE_TASK_AGE_SECONDS
);
console.log(
`Failed transport size=${currentSize} previous=${previousSize} status=${result.status} growing=${result.growing}`
);
for (const reason of result.reasons) {
console.warn(`Reason: ${reason}`);
}
if (result.status === "warn" || result.status === "critical") {
console.warn(
"Run 'bin/console messenger:failed:show' and either 'messenger:failed:retry' or " +
"'messenger:failed:remove' manually, or stand up a dedicated 'messenger:consume failed' " +
"worker if this is chronic."
);
}
if (!DRY_RUN) {
await saveCurrentSize(currentSize);
} else {
console.log("DRY_RUN is on, not persisting state for the next comparison.");
}
return result;
}
if (import.meta.url === pathToFileURL(process.argv[1]).href) {
run().catch((err) => { console.error(err); process.exit(1); });
}Add a test
The decision function is the part most worth testing, because it decides whether the operator gets paged. Since evaluate_failed_queue_backlog takes every input as a plain argument and does no I/O, the tests need no network and no Shopware account, just plain numbers in and a status out.
from messenger_failed_queue_backlog import evaluate_failed_queue_backlog
def test_ok_when_under_threshold_and_no_stale_task():
result = evaluate_failed_queue_backlog(0, None, 0, None, 900)
assert result["status"] == "ok"
assert result["reasons"] == []
def test_critical_when_over_threshold_and_growing():
result = evaluate_failed_queue_backlog(12, 5, 0, None, 900)
assert result["status"] == "critical"
assert result["growing"] is True
def test_warn_when_over_threshold_but_not_growing():
result = evaluate_failed_queue_backlog(5, 10, 0, None, 900)
assert result["status"] == "warn"
assert result["growing"] is False
def test_critical_when_no_previous_size_counts_as_growing():
result = evaluate_failed_queue_backlog(3, None, 0, None, 900)
assert result["status"] == "critical"
assert result["growing"] is True
def test_warn_when_stale_task_even_with_small_failed_count():
result = evaluate_failed_queue_backlog(0, 0, 0, 1200, 900)
assert result["status"] == "warn"
assert any("queued scheduled task" in r for r in result["reasons"])
def test_ok_when_stale_task_age_under_threshold():
result = evaluate_failed_queue_backlog(0, 0, 5, 300, 900)
assert result["status"] == "ok"
def test_reasons_include_backlog_and_growth_lines():
result = evaluate_failed_queue_backlog(8, 2, 0, None, 900)
assert len(result["reasons"]) == 2import { test } from "node:test";
import assert from "node:assert/strict";
import { evaluateFailedQueueBacklog } from "./messenger-failed-queue-backlog.js";
test("ok when under threshold and no stale task", () => {
const result = evaluateFailedQueueBacklog(0, null, 0, null, 900);
assert.equal(result.status, "ok");
assert.deepEqual(result.reasons, []);
});
test("critical when over threshold and growing", () => {
const result = evaluateFailedQueueBacklog(12, 5, 0, null, 900);
assert.equal(result.status, "critical");
assert.equal(result.growing, true);
});
test("warn when over threshold but not growing", () => {
const result = evaluateFailedQueueBacklog(5, 10, 0, null, 900);
assert.equal(result.status, "warn");
assert.equal(result.growing, false);
});
test("critical when no previous size counts as growing", () => {
const result = evaluateFailedQueueBacklog(3, null, 0, null, 900);
assert.equal(result.status, "critical");
assert.equal(result.growing, true);
});
test("warn when stale task even with small failed count", () => {
const result = evaluateFailedQueueBacklog(0, 0, 0, 1200, 900);
assert.equal(result.status, "warn");
assert.ok(result.reasons.some((r) => r.includes("queued scheduled task")));
});
test("ok when stale task age under threshold", () => {
const result = evaluateFailedQueueBacklog(0, 0, 5, 300, 900);
assert.equal(result.status, "ok");
});
test("reasons include backlog and growth lines", () => {
const result = evaluateFailedQueueBacklog(8, 2, 0, null, 900);
assert.equal(result.reasons.length, 2);
});Case studies
The integration that quietly stopped mattering
A store connected a third party fulfillment service through webhooks and flow actions. A schema change on the fulfillment side made a small percentage of payloads invalid. Messenger retried each one three times and moved every one of them to the failed transport, which nobody was watching. Months later, support noticed a pattern of orders never reaching fulfillment, and by then thousands of rows sat in messenger_messages.
Running the report script on a schedule surfaced the growing failed count within the first day, well before it reached a scale where digging through it by hand was painful. The team ran messenger:failed:show, spotted the schema mismatch, fixed the integration, and cleared the backlog deliberately.
The reindex that looked fine from the admin
A large catalog import triggered a wave of product indexing messages. A handful timed out against an overloaded database during the import window and ended up in the failed transport. Nobody had configured a worker for that queue, since the setup guide the team followed only mentioned default and async.
Adding the poll caught the failed count sitting above zero and not shrinking across repeated checks, which is exactly the growing-critical case the pure function is built to catch. The fix was mundane, run messenger:consume failed once to drain the backlog, then decide whether a permanent worker for that transport made sense given how often it happened.
After this runs on a schedule, the failed transport is no longer a blind spot. A one-off spike shows up as a warn that stops growing and can wait, while a true recurring problem shows up as critical fast, with reasons attached, long before it buries a real bug under thousands of harmless retries. Nothing gets retried or deleted without a human looking at it first, which is what keeps the backlog trustworthy as a signal instead of becoming one more thing to ignore.
FAQ
Why do failed messages pile up in Shopware's messenger queue?
Shopware runs on Symfony Messenger. A message that throws is retried automatically, and if every retry fails it moves into the separate failed transport. Most installs only run messenger:consume against the default or async queues, so nothing ever consumes the failed transport and rejected messages just sit there and grow.
Can I detect the failed queue backlog through the Shopware Admin API?
There is no entity endpoint for individual failed messages, since messenger_messages is a raw Symfony transport table, not a Shopware entity. You detect the backlog through the aggregate GET /api/_info/message-stats.json route, which returns the size of each transport by name, including the one named failed.
Should a script automatically retry or delete failed messenger messages?
No. Retrying a stale indexing message or an expired webhook call can cause duplicate side effects such as double emails or double stock adjustments. The safe pattern is to detect and report the backlog and its growth, then let a human run messenger:failed:show and decide between messenger:failed:retry and messenger:failed:remove.
Related field notes
Citations
On the problem:
- Shopware Documentation: Message Queue, the failure transport, retry-then-move behavior, and the CLI worker requirement for the failed queue. developer.shopware.com/docs/guides/hosting/infrastructure/message-queue.html
- Shopware Documentation: Tutorials and FAQ, Message Queue and Scheduled Tasks. docs.shopware.com/en/shopware-6-en/tutorials-and-faq/message-queue-and-scheduled-tasks
- Shopware Community Forum: Messenger queue full of errors. forum.shopware.com/t/messenger-queue-voller-fehler/106267
On the solution:
- Shopware Documentation: Message Queue, MESSENGER_TRANSPORT_FAILURE_DSN and worker setup. developer.shopware.com/docs/guides/hosting/infrastructure/message-queue.html
- Shopware Documentation: Scheduled Task. developer.shopware.com/docs/guides/hosting/infrastructure/scheduled-task.html
- Shopware Admin API reference (Stoplight): MessageQueueStats schema. shopware.stoplight.io/docs/admin-api/adminapi.json/components/schemas/MessageQueueStats
Stuck on a tricky one?
If you have a problem in Shopware order states, stock, message queue, or data integrity 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 clear your failed queue confusion?
If this saved you from a silent backlog or a masked bug, 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