Diagnostic Message Queue and Scheduled Tasks
Message queue backlog when only the admin worker runs
Shopware 6 ships with a convenient trick for local development: while someone is logged into the Administration, the browser's own polling quietly drains the message queue and runs scheduled tasks. It is not meant for production. If a store has no real CLI consumer running in the background, messages and overdue scheduled tasks only get processed in short bursts whenever an admin happens to have a tab open, and everything else piles up the moment they log out. Here is why that happens and a small script that detects the backlog and raises an alert, without ever trying to start a worker itself.
Shopware 6's admin worker only processes the message queue and scheduled tasks while a browser has the Administration UI open and polling. On a production store with no bin/console messenger:consume process running under systemd or supervisor, the queue drains in short bursts and falls behind as soon as everyone logs out. A small Python or Node.js script can check GET /api/_info/message-stats.json and search the scheduled_task entity for rows stuck in queued past their own runInterval, then raise a queue_backlog_detected alert. There is no API endpoint that safely starts a worker, so the script only detects and reports, gated behind DRY_RUN. Full code, tests, and sources are below.
The problem in plain words
Shopware 6 runs almost everything that is not an immediate HTTP response through its message queue: search indexing, sending mail, running Flow Builder actions, and periodic cleanup. Those jobs are driven by scheduled tasks and messenger messages, and something has to actually consume them.
For local development, Shopware includes an admin worker that rides along on the Administration UI's own background polling. Every time an admin's browser tab makes one of its routine XHR calls, it also nudges the queue forward a little. That is a fine trick on a laptop where a developer is usually logged in anyway. It quietly falls apart in production, because nobody is meant to keep the Administration open all day. The moment the last admin logs out, the queue simply stops moving. Messages queue up, scheduled tasks miss their next run, and none of it errors loudly, it just gets later and later.
Why it happens
The admin worker exists on purpose, it is a real feature meant to make local development friction free. Stores end up depending on it in production for a few common reasons:
- The store was set up quickly and
bin/console messenger:consumewas never wired into systemd or supervisor, so nothing runs the queue outside of what the admin worker happens to catch. - A consumer process was configured once, then quietly died after a deploy or a server restart and nobody noticed, because the admin worker still made things look alive whenever staff were logged in.
- Traffic and admin usage patterns hide the problem for a while. During business hours someone is often in the Administration, so the queue looks fine, then it backs up overnight or over a weekend.
- Indexing, mail sending, Flow Builder actions, and cleanup jobs all share the same queue, so a backlog in one area quietly delays every other kind of background work too.
This is a well known gap between Shopware's documentation and how stores actually get deployed. The developer docs are explicit that the admin worker is for local use only, but it is easy to miss that instruction while getting a store live, and the failure mode is silent rather than a hard error. See the citations at the end for the exact docs and a related community report.
You cannot fix this from the Admin API. There is no REST call that starts a background worker process, that is an infrastructure decision, not a data change. The safe and honest job for a script here is detection: read the queue counts and the scheduled task states, decide whether the store is backlogged, and raise an alert that names the exact overdue tasks. The actual repair is deploying a real consumer and turning the admin worker off in configuration.
The fix, as a flow
We do not try to start anything. The script samples message-stats.json twice, a few minutes apart, and separately searches scheduled_task for rows still queued. A pure decision function looks at both signals and returns whether the store is backlogged, which stale tasks caused it, and a plain reason. If it is backlogged, the script logs an alert. Nothing is written back to Shopware, ever.
Build it step by step
Get an Admin API access token
Create an integration in Shopware under Settings, System, Integrations, and note its access key id and secret key. Exchange them for a bearer token with POST /api/oauth/token using grant_type: client_credentials. Keep the shop URL and credentials in environment variables, never in the file.
pip install requests
export SHOPWARE_URL="https://yourstore.example.com"
export SHOPWARE_CLIENT_ID="SWIAXXXXXXXXXXXXXXXXXXXXXX"
export SHOPWARE_CLIENT_SECRET="..."
export STALE_FACTOR="2" # overdue by more than 2x its own runInterval
export SAMPLE_GAP_SECONDS="300" # gap between the two message-stats.json samples
export DRY_RUN="true" # start safe, this script never writes anyway// Node 18+ has fetch built in, no dependencies needed
export SHOPWARE_URL="https://yourstore.example.com"
export SHOPWARE_CLIENT_ID="SWIAXXXXXXXXXXXXXXXXXXXXXX"
export SHOPWARE_CLIENT_SECRET="..."
export STALE_FACTOR="2" // overdue by more than 2x its own runInterval
export SAMPLE_GAP_SECONDS="300" // gap between the two message-stats.json samples
export DRY_RUN="true" // start safe, this script never writes anywayAuthenticate and call the Admin API
A small helper gets a bearer token once, then sends every request with Authorization: Bearer <token> and Accept: application/json. We reuse it for both the message stats endpoint and the scheduled task search.
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"]
def api_get(path, token):
r = requests.get(
f"{SHOPWARE_URL}{path}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
timeout=30,
)
r.raise_for_status()
return r.json()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;
}
async function apiGet(path, token) {
const res = await fetch(`${SHOPWARE_URL}${path}`, {
headers: { Authorization: `Bearer ${token}`, Accept: "application/json" },
});
if (!res.ok) throw new Error(`Shopware ${res.status} on ${path}`);
return res.json();
}Sample the queue and search scheduled_task
Read GET /api/_info/message-stats.json, the current, non-deprecated endpoint (the older /api/_info/queue.json is deprecated as of 6.7.8.0 and removed in 6.8.0.0). Take a total count now, wait a few minutes, and take it again. Separately, search scheduled_task for rows with status equal to queued, sorted by nextExecutionTime ascending, reading id, name, status, lastExecutionTime, and nextExecutionTime.
def message_stats_total(token):
stats = api_get("/api/_info/message-stats.json", token)
# stats is a dict keyed by message class name, values are queue counts
return sum(int(v) for v in stats.values())
def queued_scheduled_tasks(token):
r = requests.post(
f"{SHOPWARE_URL}/api/search/scheduled-task",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
json={
"filter": [{"type": "equals", "field": "status", "value": "queued"}],
"sort": [{"field": "nextExecutionTime", "order": "ASC"}],
"limit": 100,
},
timeout=30,
)
r.raise_for_status()
return r.json()["data"]async function messageStatsTotal(token) {
const stats = await apiGet("/api/_info/message-stats.json", token);
return Object.values(stats).reduce((sum, v) => sum + Number(v), 0);
}
async function queuedScheduledTasks(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" }],
sort: [{ field: "nextExecutionTime", order: "ASC" }],
limit: 100,
}),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on scheduled-task search`);
const body = await res.json();
return body.data;
}Decide, with one pure function
Keep the decision separate from any network call. Given the queued scheduled tasks, two or more chronological message-stats.json samples, and the current time, decide whether the store is backlogged. A task counts as stale when it is overdue by more than staleFactor times its own runInterval. The queue counts as not draining when every later sample is greater than or equal to the first, and the first sample was already nonzero.
from datetime import datetime, timezone
def _parse(iso):
return datetime.fromisoformat(iso.replace("Z", "+00:00"))
def is_queue_backlogged(tasks, stats_samples, now_iso, stale_factor=2):
now = _parse(now_iso)
stale_tasks = []
for task in tasks:
if task.get("status") != "queued":
continue
age_seconds = (now - _parse(task["nextExecutionTime"])).total_seconds()
if age_seconds > task["runInterval"] * stale_factor:
stale_tasks.append(task["id"])
queue_not_draining = False
if len(stats_samples) >= 2:
first_total = stats_samples[0]["totalCount"]
if first_total > 0:
queue_not_draining = all(s["totalCount"] >= first_total for s in stats_samples[1:])
backlogged = bool(stale_tasks) or queue_not_draining
if stale_tasks and queue_not_draining:
reason = f"{len(stale_tasks)} scheduled task(s) overdue past {stale_factor}x runInterval and message-stats.json is not draining"
elif stale_tasks:
reason = f"{len(stale_tasks)} scheduled task(s) overdue past {stale_factor}x runInterval"
elif queue_not_draining:
reason = "message-stats.json total count is not decreasing across samples"
else:
reason = "no stale scheduled tasks and message-stats.json is draining"
return {"backlogged": backlogged, "staleTasks": stale_tasks, "reason": reason}export function isQueueBacklogged(tasks, statsSamples, nowIso, staleFactor = 2) {
const now = Date.parse(nowIso);
const staleTasks = [];
for (const task of tasks) {
if (task.status !== "queued") continue;
const ageSeconds = (now - Date.parse(task.nextExecutionTime)) / 1000;
if (ageSeconds > task.runInterval * staleFactor) staleTasks.push(task.id);
}
let queueNotDraining = false;
if (statsSamples.length >= 2) {
const firstTotal = statsSamples[0].totalCount;
if (firstTotal > 0) {
queueNotDraining = statsSamples.slice(1).every((s) => s.totalCount >= firstTotal);
}
}
const backlogged = staleTasks.length > 0 || queueNotDraining;
let reason;
if (staleTasks.length && queueNotDraining) {
reason = `${staleTasks.length} scheduled task(s) overdue past ${staleFactor}x runInterval and message-stats.json is not draining`;
} else if (staleTasks.length) {
reason = `${staleTasks.length} scheduled task(s) overdue past ${staleFactor}x runInterval`;
} else if (queueNotDraining) {
reason = "message-stats.json total count is not decreasing across samples";
} else {
reason = "no stale scheduled tasks and message-stats.json is draining";
}
return { backlogged, staleTasks, reason };
}Raise an alert, never a fix
When isQueueBacklogged returns true, log a queue_backlog_detected event that lists the affected scheduled task ids and names, their overdue age against their own runInterval, and the raw message-stats.json snapshot. There is no state-changing call here. The only real fix is deploying bin/console messenger:consume under systemd or supervisor for the default, low_priority, and failed transports, then setting shopware.admin_worker.enable_admin_worker: false in shopware.yaml.
This script is read-only by design, and DRY_RUN stays true by default even though nothing here writes to Shopware. There is no Admin API call that starts a worker. Treat the alert as a page to your infrastructure team, not a signal to retry an API call.
The full code
Here is the complete script in one file for each language. It authenticates, samples message-stats.json twice with a configurable gap, searches scheduled_task for stale rows, and logs a queue_backlog_detected alert when the pure decision function says the store is backlogged.
View this code on GitHub Full runnable folder with tests in the shopware-fixes repo.
"""Detect a Shopware 6 message queue backlog caused by relying on the admin worker.
Samples GET /api/_info/message-stats.json twice, a few minutes apart, and searches
the scheduled_task entity for rows still queued. A pure function decides whether the
store is backlogged. This script never writes to Shopware: there is no Admin API call
that starts a worker, so it only logs a queue_backlog_detected alert. Run on a schedule.
"""
import os
import time
import logging
import requests
from datetime import datetime, timezone
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("detect_queue_backlog")
SHOPWARE_URL = os.environ["SHOPWARE_URL"].rstrip("/")
CLIENT_ID = os.environ["SHOPWARE_CLIENT_ID"]
CLIENT_SECRET = os.environ["SHOPWARE_CLIENT_SECRET"]
STALE_FACTOR = float(os.environ.get("STALE_FACTOR", "2"))
SAMPLE_GAP_SECONDS = float(os.environ.get("SAMPLE_GAP_SECONDS", "300"))
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"
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 api_get(path, token):
r = requests.get(
f"{SHOPWARE_URL}{path}",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
timeout=30,
)
r.raise_for_status()
return r.json()
def message_stats_total(token):
stats = api_get("/api/_info/message-stats.json", token)
return sum(int(v) for v in stats.values())
def queued_scheduled_tasks(token):
r = requests.post(
f"{SHOPWARE_URL}/api/search/scheduled-task",
headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
json={
"filter": [{"type": "equals", "field": "status", "value": "queued"}],
"sort": [{"field": "nextExecutionTime", "order": "ASC"}],
"limit": 100,
},
timeout=30,
)
r.raise_for_status()
return r.json()["data"]
def _parse(iso):
return datetime.fromisoformat(iso.replace("Z", "+00:00"))
def is_queue_backlogged(tasks, stats_samples, now_iso, stale_factor=2):
now = _parse(now_iso)
stale_tasks = []
for task in tasks:
if task.get("status") != "queued":
continue
age_seconds = (now - _parse(task["nextExecutionTime"])).total_seconds()
if age_seconds > task["runInterval"] * stale_factor:
stale_tasks.append(task["id"])
queue_not_draining = False
if len(stats_samples) >= 2:
first_total = stats_samples[0]["totalCount"]
if first_total > 0:
queue_not_draining = all(s["totalCount"] >= first_total for s in stats_samples[1:])
backlogged = bool(stale_tasks) or queue_not_draining
if stale_tasks and queue_not_draining:
reason = f"{len(stale_tasks)} scheduled task(s) overdue past {stale_factor}x runInterval and message-stats.json is not draining"
elif stale_tasks:
reason = f"{len(stale_tasks)} scheduled task(s) overdue past {stale_factor}x runInterval"
elif queue_not_draining:
reason = "message-stats.json total count is not decreasing across samples"
else:
reason = "no stale scheduled tasks and message-stats.json is draining"
return {"backlogged": backlogged, "staleTasks": stale_tasks, "reason": reason}
def run():
token = get_token()
first_total = message_stats_total(token)
first_sample = {"timestamp": datetime.now(timezone.utc).isoformat(), "totalCount": first_total}
log.info("First message-stats.json sample: total=%d", first_total)
time.sleep(SAMPLE_GAP_SECONDS)
second_total = message_stats_total(token)
second_sample = {"timestamp": datetime.now(timezone.utc).isoformat(), "totalCount": second_total}
log.info("Second message-stats.json sample: total=%d", second_total)
tasks = queued_scheduled_tasks(token)
now_iso = datetime.now(timezone.utc).isoformat()
result = is_queue_backlogged(tasks, [first_sample, second_sample], now_iso, STALE_FACTOR)
if result["backlogged"]:
log.warning(
"queue_backlog_detected: %s | stale task ids: %s | samples: %s",
result["reason"], result["staleTasks"], [first_sample, second_sample],
)
else:
log.info("Queue healthy. %s", result["reason"])
return result
if __name__ == "__main__":
run()/**
* Detect a Shopware 6 message queue backlog caused by relying on the admin worker.
*
* Samples GET /api/_info/message-stats.json twice, a few minutes apart, and searches
* the scheduled_task entity for rows still queued. A pure function decides whether the
* store is backlogged. This script never writes to Shopware: there is no Admin API call
* that starts a worker, so it only logs a queue_backlog_detected alert. Run on a schedule.
*
* Guide: https://www.allanninal.dev/shopware/message-queue-backlog-admin-worker/
*/
import { pathToFileURL } from "node:url";
const SHOPWARE_URL = (process.env.SHOPWARE_URL || "https://example.test").replace(/\/$/, "");
const CLIENT_ID = process.env.SHOPWARE_CLIENT_ID || "dummy-client-id";
const CLIENT_SECRET = process.env.SHOPWARE_CLIENT_SECRET || "dummy-secret";
const STALE_FACTOR = Number(process.env.STALE_FACTOR || 2);
const SAMPLE_GAP_SECONDS = Number(process.env.SAMPLE_GAP_SECONDS || 300);
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
export function isQueueBacklogged(tasks, statsSamples, nowIso, staleFactor = 2) {
const now = Date.parse(nowIso);
const staleTasks = [];
for (const task of tasks) {
if (task.status !== "queued") continue;
const ageSeconds = (now - Date.parse(task.nextExecutionTime)) / 1000;
if (ageSeconds > task.runInterval * staleFactor) staleTasks.push(task.id);
}
let queueNotDraining = false;
if (statsSamples.length >= 2) {
const firstTotal = statsSamples[0].totalCount;
if (firstTotal > 0) {
queueNotDraining = statsSamples.slice(1).every((s) => s.totalCount >= firstTotal);
}
}
const backlogged = staleTasks.length > 0 || queueNotDraining;
let reason;
if (staleTasks.length && queueNotDraining) {
reason = `${staleTasks.length} scheduled task(s) overdue past ${staleFactor}x runInterval and message-stats.json is not draining`;
} else if (staleTasks.length) {
reason = `${staleTasks.length} scheduled task(s) overdue past ${staleFactor}x runInterval`;
} else if (queueNotDraining) {
reason = "message-stats.json total count is not decreasing across samples";
} else {
reason = "no stale scheduled tasks and message-stats.json is draining";
}
return { backlogged, staleTasks, reason };
}
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 apiGet(path, token) {
const res = await fetch(`${SHOPWARE_URL}${path}`, {
headers: { Authorization: `Bearer ${token}`, Accept: "application/json" },
});
if (!res.ok) throw new Error(`Shopware ${res.status} on ${path}`);
return res.json();
}
async function messageStatsTotal(token) {
const stats = await apiGet("/api/_info/message-stats.json", token);
return Object.values(stats).reduce((sum, v) => sum + Number(v), 0);
}
async function queuedScheduledTasks(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" }],
sort: [{ field: "nextExecutionTime", order: "ASC" }],
limit: 100,
}),
});
if (!res.ok) throw new Error(`Shopware ${res.status} on scheduled-task search`);
const body = await res.json();
return body.data;
}
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
export async function run() {
const token = await getToken();
const firstTotal = await messageStatsTotal(token);
const firstSample = { timestamp: new Date().toISOString(), totalCount: firstTotal };
console.log(`First message-stats.json sample: total=${firstTotal}`);
await sleep(SAMPLE_GAP_SECONDS * 1000);
const secondTotal = await messageStatsTotal(token);
const secondSample = { timestamp: new Date().toISOString(), totalCount: secondTotal };
console.log(`Second message-stats.json sample: total=${secondTotal}`);
const tasks = await queuedScheduledTasks(token);
const nowIso = new Date().toISOString();
const result = isQueueBacklogged(tasks, [firstSample, secondSample], nowIso, STALE_FACTOR);
if (result.backlogged) {
console.warn(
`queue_backlog_detected: ${result.reason} | stale task ids: ${JSON.stringify(result.staleTasks)} | samples: ${JSON.stringify([firstSample, secondSample])}`
);
} else {
console.log(`Queue healthy. ${result.reason}`);
}
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 rule is the part most worth testing, because it decides whether an alert fires at all. Because isQueueBacklogged is pure and takes the current time as an argument, the test needs no network and no Shopware store. It just feeds in fixed clocks and fixture arrays.
from detect_queue_backlog import is_queue_backlogged
NOW = "2026-07-10T12:00:00Z"
def task(**over):
base = {
"id": "a" * 32,
"status": "queued",
"nextExecutionTime": "2026-07-10T11:00:00Z", # 1 hour ago
"runInterval": 900, # 15 minutes, so 1 hour is 4x, well past 2x
}
base.update(over)
return base
def sample(total, ts="2026-07-10T11:55:00Z"):
return {"timestamp": ts, "totalCount": total}
def test_backlogged_when_task_stale_past_factor():
result = is_queue_backlogged([task()], [sample(0), sample(0)], NOW, stale_factor=2)
assert result["backlogged"] is True
assert result["staleTasks"] == ["a" * 32]
def test_not_backlogged_when_task_within_factor():
fresh = task(nextExecutionTime="2026-07-10T11:58:00Z", runInterval=900)
result = is_queue_backlogged([fresh], [sample(0), sample(0)], NOW, stale_factor=2)
assert result["backlogged"] is False
assert result["staleTasks"] == []
def test_ignores_tasks_not_queued():
done = task(status="scheduled")
result = is_queue_backlogged([done], [sample(0), sample(0)], NOW, stale_factor=2)
assert result["backlogged"] is False
def test_backlogged_when_stats_not_draining():
samples = [sample(50, "2026-07-10T11:50:00Z"), sample(60, "2026-07-10T11:55:00Z")]
result = is_queue_backlogged([], samples, NOW, stale_factor=2)
assert result["backlogged"] is True
assert "not draining" in result["reason"] or "not decreasing" in result["reason"]
def test_not_backlogged_when_stats_draining_and_first_zero():
samples = [sample(0, "2026-07-10T11:50:00Z"), sample(0, "2026-07-10T11:55:00Z")]
result = is_queue_backlogged([], samples, NOW, stale_factor=2)
assert result["backlogged"] is False
def test_not_backlogged_when_stats_are_draining():
samples = [sample(50, "2026-07-10T11:50:00Z"), sample(10, "2026-07-10T11:55:00Z")]
result = is_queue_backlogged([], samples, NOW, stale_factor=2)
assert result["backlogged"] is False
def test_single_sample_cannot_prove_non_draining():
result = is_queue_backlogged([], [sample(50)], NOW, stale_factor=2)
assert result["backlogged"] is Falseimport { test } from "node:test";
import assert from "node:assert/strict";
import { isQueueBacklogged } from "./detect-queue-backlog.js";
const NOW = "2026-07-10T12:00:00Z";
const task = (over = {}) => ({
id: "a".repeat(32),
status: "queued",
nextExecutionTime: "2026-07-10T11:00:00Z", // 1 hour ago
runInterval: 900, // 15 minutes, so 1 hour is 4x, well past 2x
...over,
});
const sample = (total, ts = "2026-07-10T11:55:00Z") => ({ timestamp: ts, totalCount: total });
test("backlogged when task stale past factor", () => {
const result = isQueueBacklogged([task()], [sample(0), sample(0)], NOW, 2);
assert.equal(result.backlogged, true);
assert.deepEqual(result.staleTasks, ["a".repeat(32)]);
});
test("not backlogged when task within factor", () => {
const fresh = task({ nextExecutionTime: "2026-07-10T11:58:00Z", runInterval: 900 });
const result = isQueueBacklogged([fresh], [sample(0), sample(0)], NOW, 2);
assert.equal(result.backlogged, false);
assert.deepEqual(result.staleTasks, []);
});
test("ignores tasks not queued", () => {
const done = task({ status: "scheduled" });
const result = isQueueBacklogged([done], [sample(0), sample(0)], NOW, 2);
assert.equal(result.backlogged, false);
});
test("backlogged when stats not draining", () => {
const samples = [sample(50, "2026-07-10T11:50:00Z"), sample(60, "2026-07-10T11:55:00Z")];
const result = isQueueBacklogged([], samples, NOW, 2);
assert.equal(result.backlogged, true);
});
test("not backlogged when stats draining and first is zero", () => {
const samples = [sample(0, "2026-07-10T11:50:00Z"), sample(0, "2026-07-10T11:55:00Z")];
const result = isQueueBacklogged([], samples, NOW, 2);
assert.equal(result.backlogged, false);
});
test("not backlogged when stats are draining", () => {
const samples = [sample(50, "2026-07-10T11:50:00Z"), sample(10, "2026-07-10T11:55:00Z")];
const result = isQueueBacklogged([], samples, NOW, 2);
assert.equal(result.backlogged, false);
});
test("single sample cannot prove non draining", () => {
const result = isQueueBacklogged([], [sample(50)], NOW, 2);
assert.equal(result.backlogged, false);
});Case studies
The store that only indexed while staff were awake
A mid-size furniture store went live on Shopware 6 without ever wiring up messenger:consume. During the day, staff kept the Administration open to manage orders, so the admin worker quietly kept indexing and mail moving. Every night, product changes and price updates from the previous day sat unindexed until someone logged in the next morning.
Running the detection script on a five minute cron caught it fast: message-stats.json was non-zero and flat every night, and several scheduled tasks were sitting well past double their runInterval by sunrise. That alert was what finally got a real consumer deployed under supervisor.
The consumer that stopped after a server reboot
A B2B store had messenger:consume running, but it was started manually in a terminal session rather than under systemd. A routine server reboot for security patches killed the process, and nobody restarted it. Because a few admins were still logged in during business hours, the queue looked mostly fine for two days.
The scheduled task search surfaced it precisely: cleanup and mail-queue tasks were stuck in queued far past their interval, and the alert named the exact task ids. That made it a fast fix, moving the consumer into a proper systemd unit with a restart policy instead of a terminal tab.
After the detection script runs on a schedule, a stalled queue gets caught within minutes instead of showing up as late order emails or stale search results days later. The alert names the exact overdue scheduled tasks and the raw queue snapshot, so whoever is on call can go straight to checking the systemd unit for messenger:consume, rather than guessing. The real fix stays a deliberate infrastructure change: a real CLI consumer for the default, low_priority, and failed transports, and enable_admin_worker turned off.
FAQ
Why does my Shopware 6 message queue only drain when I am logged into the Administration?
Shopware 6 ships with an admin worker that processes the message queue and scheduled tasks by piggybacking on the browser polling the Administration UI already does. It is meant for local development convenience. On a production store with no CLI consumer running under systemd or supervisor, messages only drain in short bursts whenever an admin has a browser tab open, so the queue backs up as soon as everyone logs out.
How do I check if my Shopware 6 store is backlogged on the admin worker?
Query GET /api/_info/message-stats.json for per-message-type queue counts and sample it twice a few minutes apart. Also search the scheduled_task entity for rows with status queued sorted by nextExecutionTime. If a task's age since nextExecutionTime is more than twice its own runInterval, or the message-stats.json total is not shrinking between samples, the store is relying on the admin worker or has no consumer running at all.
Can I fix a Shopware 6 message queue backlog through the Admin API?
No. There is no REST endpoint that starts a worker process. The real fix is an infrastructure change: run bin/console messenger:consume for the default, low_priority, and failed transports under systemd or supervisor, then set shopware.admin_worker.enable_admin_worker to false in shopware.yaml. A script can only detect the backlog and raise an alert, it cannot safely repair it.
Related field notes
Citations
On the problem:
- Shopware Developer Documentation: Message Queue, including the admin worker and its local-only intent. developer.shopware.com/docs/guides/hosting/infrastructure/message-queue.html
- Shopware Docs: Tutorials and FAQ, Message Queue and Scheduled Tasks. docs.shopware.com/en/shopware-6-en/tutorials-and-faq/message-queue-and-scheduled-tasks
- GitHub shopware/shopware issue 3508: message consumer error while processing a queued message. github.com/shopware/shopware/issues/3508
On the solution:
- Shopware Developer Documentation: Message Queue, the CLI consumer and admin_worker config. developer.shopware.com/docs/guides/hosting/infrastructure/message-queue.html
- Shopware Developer Documentation: Scheduled Task. developer.shopware.com/docs/guides/hosting/infrastructure/scheduled-task.html
- Shopware 6.7.8.0 Release Notes: message-stats.json and the queue.json deprecation. developer.shopware.com/release-notes/6.7/6.7.8.0.html
Stuck on a tricky one?
If you have a problem in Shopware orders, the state machine, stock, or the message queue 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 help you catch a stalled queue?
If this saved you from a mystery indexing lag or a batch of late order emails, 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