Reconciler Message Queue and Scheduled Tasks
Webhook event logs stuck in queued state pile up
Every webhook Shopware sends starts the same way: a row lands in webhook_event_log with deliveryStatus set to queued, right as the message is handed to the message queue. Most of the time a consumer picks it up moments later and flips it to success or failed. But when that message never gets consumed, no worker running, a dropped connection, a deleted receiver, an app deactivated mid-flight, the row just stays at queued. Forever. Here is why Shopware left that gap open until recently, and a small script that finds the truly orphaned rows and clears them out safely.
Shopware writes a webhook_event_log row with deliveryStatus="queued" the moment a webhook message is dispatched, then updates it to success or failed only when a consumer actually processes it. If that message is lost, the row is orphaned at queued permanently, and until Shopware 6.7.4 the built-in cleanup task only ever deleted success/failed rows, so queued rows were never touched. Run a small Python or Node.js script that reads core.webhook.entryLifetimeSeconds, searches webhook-event-log for rows still queued and older than twice that lifetime, skips any row that still has a corresponding webhook_delivery record in flight, and hard deletes the rest with DELETE /api/webhook-event-log/{id}. If entryLifetimeSeconds is -1, cleanup is disabled on purpose and the script only reports, it never deletes. Full code, tests, and the reasoning are below.
The problem in plain words
Every webhook Shopware fires, an order placed, a product changed, a customer registered, goes through the same two steps. First Shopware writes a log row for it with deliveryStatus="queued" and dispatches a message to the message queue. Then, later, a consumer picks that message up, calls the receiving URL, and updates the same row to success or failed depending on what happened.
That second step depends on something actually consuming the message. If nothing does, because no worker is running, the message hits a serialization error, the app that owns the webhook gets deactivated mid-flight, or the message is simply dropped, the row never gets its second update. It just sits there at queued, looking exactly like a webhook that is still in progress, except it never will be. On a store with any queue reliability problem at all, these orphaned rows build up in webhook_event_log a little at a time, and nothing tells you they are dead.
Why it happens
The core cause is a timing gap, and it is made worse by a gap in the cleanup logic that only recently closed. A few concrete ways it shows up:
- A store with no reliable
messenger:consumeworker running, so dispatched messages sit unconsumed indefinitely, the same underlying issue behind a general message queue backlog. - A message that fails to deserialize because of a schema mismatch or a bad payload, so it never reaches the code path that would set
successorfailed. - The app or integration receiving the webhook gets deactivated or uninstalled while its message is still mid-flight, and there is no receiver left to report back to.
- A worker crash or a deploy that kills the consumer process between dispatch and delivery, losing that in-memory message entirely.
- Until Shopware 6.7.4, the
webhook_event_log.cleanupscheduled task, backed byWebhookCleanup::removeOldLogs, only ever deleted rows withdeliveryStatusofsuccessorfailedbased oncore.webhook.entryLifetimeSeconds. Queued rows were explicitly excluded, on purpose, because a queued row might still be legitimately in flight, but that same logic let permanently orphaned rows accumulate right alongside the ones still genuinely waiting.
The result is a table that grows without bound on any store whose queue infrastructure is not rock solid, and the built-in cleanup made things worse by looking like it was handling retention when it was quietly skipping the one status that most needed a second look.
Not every queued row is stale. A webhook dispatched thirty seconds ago is supposed to be queued, that is normal, healthy in-flight state. The only rows worth touching are the ones old enough that no reasonable consumer delay explains them, and even among those, some may still have an active webhook_delivery record tracking a retry. So the rule mirrors Shopware's own fix: only delete a queued row when it is older than twice the configured retention window and has no corresponding delivery record still tracking it. Everything else is left alone.
The fix, as a flow
We authenticate once, read the store's own retention setting so the cutoff matches what the merchant already configured, search for queued rows older than that cutoff, check each one for a live webhook_delivery record, and hard delete only the rows that clear both checks. If cleanup is disabled entirely, the script does not delete anything, it just reports what it found.
Build it step by step
Get Admin API credentials
Create an integration in your Shopware admin under 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 so the first run only reports what it would delete.
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 PAGE_LIMIT="500"
export DRY_RUN="true" # start safe, change to false to write// 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 PAGE_LIMIT="500"
export DRY_RUN="true" // start safe, change to false to writeAuthenticate 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"]
def auth_headers(token):
return {"Authorization": f"Bearer {token}", "Accept": "application/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;
}
function authHeaders(token) {
return { Authorization: `Bearer ${token}`, Accept: "application/json" };
}Read the store's own retention setting first
Read core.webhook.entryLifetimeSeconds before doing anything else, so the cutoff matches what the merchant already configured rather than a guessed number. A value of -1 means cleanup is disabled on purpose, in which case the script should never delete, only report.
def get_entry_lifetime_seconds(token):
r = requests.get(
f"{SHOPWARE_URL}/api/_action/system-config",
params={"domain": "core.webhook"},
headers=auth_headers(token),
timeout=30,
)
r.raise_for_status()
config = r.json()
value = config.get("core.webhook.entryLifetimeSeconds")
if value is None:
return -1
return int(value)async function getEntryLifetimeSeconds(token) {
const res = await fetch(`${SHOPWARE_URL}/api/_action/system-config?domain=core.webhook`, {
headers: authHeaders(token),
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const config = await res.json();
const value = config["core.webhook.entryLifetimeSeconds"];
return value === undefined || value === null ? -1 : Number(value);
}List the queued rows older than the cutoff
Search webhook-event-log for rows where deliveryStatus equals queued and createdAt is at or before a cutoff of now minus twice the entry lifetime. Sort by createdAt ascending so the oldest, most clearly orphaned rows come first, and page through with total-count-mode in case the backlog is large.
def queued_webhook_event_logs(token, cutoff_iso):
results = []
page = 1
while True:
r = requests.post(
f"{SHOPWARE_URL}/api/search/webhook-event-log",
json={
"page": page,
"limit": PAGE_LIMIT,
"filter": [
{"type": "equals", "field": "deliveryStatus", "value": "queued"},
{"type": "range", "field": "createdAt", "parameters": {"lte": cutoff_iso}},
],
"sort": [{"field": "createdAt", "order": "ASC"}],
"total-count-mode": 1,
},
headers={**auth_headers(token), "Content-Type": "application/json"},
timeout=30,
)
r.raise_for_status()
rows = r.json().get("data", [])
for row in rows:
attrs = row.get("attributes", row)
results.append({
"id": row["id"],
"deliveryStatus": attrs.get("deliveryStatus"),
"createdAt": attrs.get("createdAt"),
"webhookName": attrs.get("webhookName"),
"eventName": attrs.get("eventName"),
"url": attrs.get("url"),
})
if len(rows) < PAGE_LIMIT:
return results
page += 1async function queuedWebhookEventLogs(token, cutoffIso) {
const results = [];
let page = 1;
while (true) {
const res = await fetch(`${SHOPWARE_URL}/api/search/webhook-event-log`, {
method: "POST",
headers: { ...authHeaders(token), "Content-Type": "application/json" },
body: JSON.stringify({
page,
limit: PAGE_LIMIT,
filter: [
{ type: "equals", field: "deliveryStatus", value: "queued" },
{ type: "range", field: "createdAt", parameters: { lte: cutoffIso } },
],
sort: [{ field: "createdAt", order: "ASC" }],
"total-count-mode": 1,
}),
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const body = await res.json();
const rows = body.data || [];
for (const row of rows) {
const attrs = row.attributes || row;
results.push({
id: row.id,
deliveryStatus: attrs.deliveryStatus,
createdAt: attrs.createdAt,
webhookName: attrs.webhookName,
eventName: attrs.eventName,
url: attrs.url,
});
}
if (rows.length < PAGE_LIMIT) return results;
page++;
}
}Check each candidate for a live webhook_delivery record
Before a row is deleted, confirm it has no corresponding webhook_delivery record. That mirrors Shopware's own upstream guard, a NOT EXISTS check against the delivery table, since a queued row that still has an active delivery may genuinely be retrying rather than lost.
def has_delivery_record(token, webhook_event_log_id):
r = requests.post(
f"{SHOPWARE_URL}/api/search/webhook-event-log",
json={
"page": 1,
"limit": 1,
"filter": [{"type": "equals", "field": "id", "value": webhook_event_log_id}],
"associations": {"webhookDelivery": {}},
"total-count-mode": 1,
},
headers={**auth_headers(token), "Content-Type": "application/json"},
timeout=30,
)
r.raise_for_status()
rows = r.json().get("data", [])
if not rows:
return False
attrs = rows[0].get("attributes", rows[0])
delivery = attrs.get("webhookDelivery") or rows[0].get("relationships", {}).get("webhookDelivery")
return bool(delivery)async function hasDeliveryRecord(token, webhookEventLogId) {
const res = await fetch(`${SHOPWARE_URL}/api/search/webhook-event-log`, {
method: "POST",
headers: { ...authHeaders(token), "Content-Type": "application/json" },
body: JSON.stringify({
page: 1,
limit: 1,
filter: [{ type: "equals", field: "id", value: webhookEventLogId }],
associations: { webhookDelivery: {} },
"total-count-mode": 1,
}),
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const body = await res.json();
const rows = body.data || [];
if (rows.length === 0) return false;
const attrs = rows[0].attributes || rows[0];
const delivery = attrs.webhookDelivery || rows[0].relationships?.webhookDelivery;
return Boolean(delivery);
}Decide, with one pure function
Keep the decision in its own function that takes the rows, the current time, and the entry lifetime, and returns just the ids eligible for deletion. A pure function like this is easy to read and easy to test, which we do later. If the lifetime is -1, nothing is ever eligible. Otherwise a row is only eligible when it is still queued, its createdAt is strictly before the cutoff of now minus twice the lifetime, and it has no delivery record.
import datetime
def _iso_to_epoch(iso):
return datetime.datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp()
def select_stale_queued_logs(rows, now_iso, entry_lifetime_seconds):
"""rows: list of {id, deliveryStatus, createdAt, hasDelivery}. Pure, no I/O."""
if entry_lifetime_seconds == -1:
return []
now_epoch = _iso_to_epoch(now_iso)
cutoff_epoch = now_epoch - (entry_lifetime_seconds * 2)
stale_ids = []
for row in rows:
if row.get("deliveryStatus") != "queued":
continue
created_at = row.get("createdAt")
if not created_at:
continue
if _iso_to_epoch(created_at) >= cutoff_epoch:
continue
if row.get("hasDelivery"):
continue
stale_ids.append(row["id"])
return stale_idsexport function isoToEpoch(iso) {
return Date.parse(iso) / 1000;
}
export function selectStaleQueuedLogs(rows, nowIso, entryLifetimeSeconds) {
// rows: list of {id, deliveryStatus, createdAt, hasDelivery}. Pure, no I/O.
if (entryLifetimeSeconds === -1) return [];
const nowEpoch = isoToEpoch(nowIso);
const cutoffEpoch = nowEpoch - entryLifetimeSeconds * 2;
const staleIds = [];
for (const row of rows) {
if (row.deliveryStatus !== "queued") continue;
if (!row.createdAt) continue;
if (isoToEpoch(row.createdAt) >= cutoffEpoch) continue;
if (row.hasDelivery) continue;
staleIds.push(row.id);
}
return staleIds;
}Delete the confirmed-stale rows
deliveryStatus is a plain string field, not a state machine, so there is no /_action/state transition here, and nothing to mutate in place either. Shopware's own upstream fix deletes the row outright, so a plain DELETE /api/webhook-event-log/{id} per confirmed id is the correct action, one call per row for clear logging, or batched through POST /api/_action/sync with a delete action array if the backlog is large.
def delete_webhook_event_log(token, log_id):
r = requests.delete(
f"{SHOPWARE_URL}/api/webhook-event-log/{log_id}",
headers=auth_headers(token),
timeout=30,
)
if r.status_code not in (204, 200):
r.raise_for_status()async function deleteWebhookEventLog(token, logId) {
const res = await fetch(`${SHOPWARE_URL}/api/webhook-event-log/${logId}`, {
method: "DELETE",
headers: authHeaders(token),
});
if (!res.ok && res.status !== 204) throw new Error(`Shopware ${res.status}`);
}Wire it together with a dry run guard
The loop ties every piece together: get a token, read the entry lifetime, search for candidate rows past the cutoff, check each for a delivery record, run the pure decision function, then log and optionally delete. Leave DRY_RUN on for the first few runs so it only logs which rows it would delete. Run it on a schedule, once a day is plenty since orphaned rows are not time sensitive once they are already stale.
Always start with DRY_RUN=true, and never lower the cutoff below twice core.webhook.entryLifetimeSeconds. If that setting is -1, the merchant disabled cleanup on purpose, so the script reports and does not delete anything at all, no matter how old the rows look.
The full code
Here is the complete script in one file for each language. It reads settings from the environment, authenticates, reads the store's own retention setting, finds the stale queued rows, cross-checks each for a live delivery record, and deletes only what clears every guard. It is safe to run again and again because a row that no longer exists simply will not show up in the next search.
View this code on GitHub Full runnable folder with tests in the shopware-fixes repo.
"""Delete webhook_event_log rows permanently stuck in deliveryStatus="queued".
Shopware writes a webhook_event_log row with deliveryStatus="queued" the moment
a webhook message is dispatched to the message queue, then flips it to "success"
or "failed" only when a consumer actually processes it. If that queued message is
lost, dropped, or never consumed, the log row is orphaned in "queued" forever.
Until Shopware 6.7.4, the built-in webhook_event_log.cleanup scheduled task only
ever deleted success/failed rows based on core.webhook.entryLifetimeSeconds, so
queued rows were explicitly excluded from cleanup and just accumulate.
This script mirrors Shopware's own upstream fix: it hard deletes stale queued rows,
skipping any row that still has a corresponding webhook_delivery record (still
legitimately in-flight or retrying), and it never touches anything newer than
2x entryLifetimeSeconds. If entryLifetimeSeconds is -1 (cleanup disabled by the
merchant), it does not delete anything, it only flags and reports.
Run on a schedule. Safe to run again and again.
"""
import os
import logging
import datetime
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("cleanup_webhook_event_log")
SHOPWARE_URL = os.environ["SHOPWARE_URL"].rstrip("/")
CLIENT_ID = os.environ["SHOPWARE_CLIENT_ID"]
CLIENT_SECRET = os.environ["SHOPWARE_CLIENT_SECRET"]
PAGE_LIMIT = int(os.environ.get("PAGE_LIMIT", "500"))
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 auth_headers(token):
return {"Authorization": f"Bearer {token}", "Accept": "application/json"}
def get_entry_lifetime_seconds(token):
r = requests.get(
f"{SHOPWARE_URL}/api/_action/system-config",
params={"domain": "core.webhook"},
headers=auth_headers(token),
timeout=30,
)
r.raise_for_status()
config = r.json()
value = config.get("core.webhook.entryLifetimeSeconds")
if value is None:
return -1
return int(value)
def iso_now():
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
def queued_webhook_event_logs(token, cutoff_iso):
results = []
page = 1
while True:
r = requests.post(
f"{SHOPWARE_URL}/api/search/webhook-event-log",
json={
"page": page,
"limit": PAGE_LIMIT,
"filter": [
{"type": "equals", "field": "deliveryStatus", "value": "queued"},
{"type": "range", "field": "createdAt", "parameters": {"lte": cutoff_iso}},
],
"sort": [{"field": "createdAt", "order": "ASC"}],
"total-count-mode": 1,
},
headers={**auth_headers(token), "Content-Type": "application/json"},
timeout=30,
)
r.raise_for_status()
body = r.json()
rows = body.get("data", [])
for row in rows:
attrs = row.get("attributes", row)
results.append({
"id": row["id"],
"deliveryStatus": attrs.get("deliveryStatus"),
"createdAt": attrs.get("createdAt"),
"webhookName": attrs.get("webhookName"),
"eventName": attrs.get("eventName"),
"url": attrs.get("url"),
})
if len(rows) < PAGE_LIMIT:
return results
page += 1
def has_delivery_record(token, webhook_event_log_id):
r = requests.post(
f"{SHOPWARE_URL}/api/search/webhook-event-log",
json={
"page": 1,
"limit": 1,
"filter": [{"type": "equals", "field": "id", "value": webhook_event_log_id}],
"associations": {"webhookDelivery": {}},
"total-count-mode": 1,
},
headers={**auth_headers(token), "Content-Type": "application/json"},
timeout=30,
)
r.raise_for_status()
body = r.json()
rows = body.get("data", [])
if not rows:
return False
attrs = rows[0].get("attributes", rows[0])
delivery = attrs.get("webhookDelivery") or rows[0].get("relationships", {}).get("webhookDelivery")
return bool(delivery)
def select_stale_queued_logs(rows, now_iso, entry_lifetime_seconds):
"""Pure decision function. No I/O.
rows: list of {id, deliveryStatus, createdAt, hasDelivery}
now_iso: ISO8601 timestamp string used as "now" for the cutoff computation
entry_lifetime_seconds: core.webhook.entryLifetimeSeconds (-1 means disabled)
Returns the list of ids eligible for deletion.
"""
if entry_lifetime_seconds == -1:
return []
now_epoch = _iso_to_epoch(now_iso)
cutoff_epoch = now_epoch - (entry_lifetime_seconds * 2)
stale_ids = []
for row in rows:
if row.get("deliveryStatus") != "queued":
continue
created_at = row.get("createdAt")
if not created_at:
continue
if _iso_to_epoch(created_at) >= cutoff_epoch:
continue
if row.get("hasDelivery"):
continue
stale_ids.append(row["id"])
return stale_ids
def _iso_to_epoch(iso):
return datetime.datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp()
def delete_webhook_event_log(token, log_id):
r = requests.delete(
f"{SHOPWARE_URL}/api/webhook-event-log/{log_id}",
headers=auth_headers(token),
timeout=30,
)
if r.status_code not in (204, 200):
r.raise_for_status()
def run():
token = get_token()
entry_lifetime_seconds = get_entry_lifetime_seconds(token)
if entry_lifetime_seconds == -1:
log.warning(
"core.webhook.entryLifetimeSeconds is -1 (cleanup disabled by merchant choice). "
"Skipping deletion, reporting only."
)
now_iso = iso_now()
if entry_lifetime_seconds == -1:
cutoff_iso = now_iso
else:
cutoff_epoch = _iso_to_epoch(now_iso) - (entry_lifetime_seconds * 2)
cutoff_iso = datetime.datetime.fromtimestamp(
cutoff_epoch, datetime.timezone.utc
).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
candidates = queued_webhook_event_logs(token, cutoff_iso)
log.info("Found %d queued row(s) older than the cutoff %s.", len(candidates), cutoff_iso)
rows = []
for candidate in candidates:
has_delivery = has_delivery_record(token, candidate["id"])
rows.append({**candidate, "hasDelivery": has_delivery})
stale_ids = select_stale_queued_logs(rows, now_iso, entry_lifetime_seconds)
by_id = {row["id"]: row for row in rows}
for stale_id in stale_ids:
row = by_id[stale_id]
log.info(
"Stale queued log %s (webhook=%s event=%s createdAt=%s). %s",
stale_id, row.get("webhookName"), row.get("eventName"), row.get("createdAt"),
"would delete" if DRY_RUN else "deleting",
)
if not DRY_RUN:
delete_webhook_event_log(token, stale_id)
skipped_with_delivery = [r for r in rows if r["id"] not in stale_ids and r.get("hasDelivery")]
for row in skipped_with_delivery:
log.info("Skipping log %s, still has a webhook_delivery record in flight.", row["id"])
log.info(
"Done. %d stale queued row(s) %s.",
len(stale_ids), "to delete" if DRY_RUN else "deleted",
)
return stale_ids
if __name__ == "__main__":
run()/**
* Delete webhook_event_log rows permanently stuck in deliveryStatus="queued".
*
* Shopware writes a webhook_event_log row with deliveryStatus="queued" the moment
* a webhook message is dispatched to the message queue, then flips it to "success"
* or "failed" only when a consumer actually processes it. If that queued message is
* lost, dropped, or never consumed, the log row is orphaned in "queued" forever.
* Until Shopware 6.7.4, the built-in webhook_event_log.cleanup scheduled task only
* ever deleted success/failed rows based on core.webhook.entryLifetimeSeconds, so
* queued rows were explicitly excluded from cleanup and just accumulate.
*
* This script mirrors Shopware's own upstream fix: it hard deletes stale queued rows,
* skipping any row that still has a corresponding webhook_delivery record (still
* legitimately in-flight or retrying), and it never touches anything newer than
* 2x entryLifetimeSeconds. If entryLifetimeSeconds is -1 (cleanup disabled by the
* merchant), it does not delete anything, it only flags and reports.
* Run on a schedule. Safe to run again and again.
*
* Guide: https://www.allanninal.dev/shopware/webhook-event-log-backlog/
*/
import { pathToFileURL } from "node:url";
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 PAGE_LIMIT = Number(process.env.PAGE_LIMIT || 500);
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
export function isoToEpoch(iso) {
return Date.parse(iso) / 1000;
}
export function selectStaleQueuedLogs(rows, nowIso, entryLifetimeSeconds) {
if (entryLifetimeSeconds === -1) return [];
const nowEpoch = isoToEpoch(nowIso);
const cutoffEpoch = nowEpoch - entryLifetimeSeconds * 2;
const staleIds = [];
for (const row of rows) {
if (row.deliveryStatus !== "queued") continue;
if (!row.createdAt) continue;
if (isoToEpoch(row.createdAt) >= cutoffEpoch) continue;
if (row.hasDelivery) continue;
staleIds.push(row.id);
}
return staleIds;
}
function authHeaders(token) {
return { Authorization: `Bearer ${token}`, Accept: "application/json" };
}
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 getEntryLifetimeSeconds(token) {
const res = await fetch(`${SHOPWARE_URL}/api/_action/system-config?domain=core.webhook`, {
headers: authHeaders(token),
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const config = await res.json();
const value = config["core.webhook.entryLifetimeSeconds"];
return value === undefined || value === null ? -1 : Number(value);
}
async function queuedWebhookEventLogs(token, cutoffIso) {
const results = [];
let page = 1;
while (true) {
const res = await fetch(`${SHOPWARE_URL}/api/search/webhook-event-log`, {
method: "POST",
headers: { ...authHeaders(token), "Content-Type": "application/json" },
body: JSON.stringify({
page,
limit: PAGE_LIMIT,
filter: [
{ type: "equals", field: "deliveryStatus", value: "queued" },
{ type: "range", field: "createdAt", parameters: { lte: cutoffIso } },
],
sort: [{ field: "createdAt", order: "ASC" }],
"total-count-mode": 1,
}),
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const body = await res.json();
const rows = body.data || [];
for (const row of rows) {
const attrs = row.attributes || row;
results.push({
id: row.id,
deliveryStatus: attrs.deliveryStatus,
createdAt: attrs.createdAt,
webhookName: attrs.webhookName,
eventName: attrs.eventName,
url: attrs.url,
});
}
if (rows.length < PAGE_LIMIT) return results;
page++;
}
}
async function hasDeliveryRecord(token, webhookEventLogId) {
const res = await fetch(`${SHOPWARE_URL}/api/search/webhook-event-log`, {
method: "POST",
headers: { ...authHeaders(token), "Content-Type": "application/json" },
body: JSON.stringify({
page: 1,
limit: 1,
filter: [{ type: "equals", field: "id", value: webhookEventLogId }],
associations: { webhookDelivery: {} },
"total-count-mode": 1,
}),
});
if (!res.ok) throw new Error(`Shopware ${res.status}`);
const body = await res.json();
const rows = body.data || [];
if (rows.length === 0) return false;
const attrs = rows[0].attributes || rows[0];
const delivery = attrs.webhookDelivery || rows[0].relationships?.webhookDelivery;
return Boolean(delivery);
}
async function deleteWebhookEventLog(token, logId) {
const res = await fetch(`${SHOPWARE_URL}/api/webhook-event-log/${logId}`, {
method: "DELETE",
headers: authHeaders(token),
});
if (!res.ok && res.status !== 204) throw new Error(`Shopware ${res.status}`);
}
export async function run() {
const token = await getToken();
const entryLifetimeSeconds = await getEntryLifetimeSeconds(token);
if (entryLifetimeSeconds === -1) {
console.warn(
"core.webhook.entryLifetimeSeconds is -1 (cleanup disabled by merchant choice). " +
"Skipping deletion, reporting only."
);
}
const nowIso = new Date().toISOString();
const cutoffIso =
entryLifetimeSeconds === -1
? nowIso
: new Date((isoToEpoch(nowIso) - entryLifetimeSeconds * 2) * 1000).toISOString();
const candidates = await queuedWebhookEventLogs(token, cutoffIso);
console.log(`Found ${candidates.length} queued row(s) older than the cutoff ${cutoffIso}.`);
const rows = [];
for (const candidate of candidates) {
const hasDelivery = await hasDeliveryRecord(token, candidate.id);
rows.push({ ...candidate, hasDelivery });
}
const staleIds = selectStaleQueuedLogs(rows, nowIso, entryLifetimeSeconds);
const byId = new Map(rows.map((row) => [row.id, row]));
for (const staleId of staleIds) {
const row = byId.get(staleId);
console.log(
`Stale queued log ${staleId} (webhook=${row.webhookName} event=${row.eventName} ` +
`createdAt=${row.createdAt}). ${DRY_RUN ? "would delete" : "deleting"}`
);
if (!DRY_RUN) await deleteWebhookEventLog(token, staleId);
}
const skippedWithDelivery = rows.filter((r) => !staleIds.includes(r.id) && r.hasDelivery);
for (const row of skippedWithDelivery) {
console.log(`Skipping log ${row.id}, still has a webhook_delivery record in flight.`);
}
console.log(`Done. ${staleIds.length} stale queued row(s) ${DRY_RUN ? "to delete" : "deleted"}.`);
return staleIds;
}
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 which rows get permanently deleted. Since select_stale_queued_logs takes plain data and the current time as arguments and does no I/O, the tests need no network and no Shopware account, just fabricated rows and a fixed clock.
from cleanup_webhook_event_log import select_stale_queued_logs
NOW = "2026-07-10T00:00:00.000Z"
def row(**over):
base = {
"id": "a" * 32,
"deliveryStatus": "queued",
"createdAt": "2026-07-01T00:00:00.000Z", # 9 days old
"hasDelivery": False,
}
base.update(over)
return base
def test_disabled_cleanup_returns_empty():
rows = [row()]
assert select_stale_queued_logs(rows, NOW, -1) == []
def test_fresh_queued_row_is_kept():
fresh = row(id="b" * 32, createdAt="2026-07-09T23:00:00.000Z")
assert select_stale_queued_logs([fresh], NOW, 3600) == []
def test_stale_queued_with_delivery_record_is_kept():
stale_with_delivery = row(id="c" * 32, hasDelivery=True)
assert select_stale_queued_logs([stale_with_delivery], NOW, 3600) == []
def test_stale_queued_without_delivery_record_is_deleted():
stale = row(id="d" * 32)
assert select_stale_queued_logs([stale], NOW, 3600) == ["d" * 32]
def test_success_and_failed_rows_are_ignored():
success = row(id="e" * 32, deliveryStatus="success")
failed = row(id="f" * 32, deliveryStatus="failed")
assert select_stale_queued_logs([success, failed], NOW, 3600) == []import { test } from "node:test";
import assert from "node:assert/strict";
import { selectStaleQueuedLogs } from "./cleanup-webhook-event-log.js";
const NOW = "2026-07-10T00:00:00.000Z";
const row = (over = {}) => ({
id: "a".repeat(32),
deliveryStatus: "queued",
createdAt: "2026-07-01T00:00:00.000Z",
hasDelivery: false,
...over,
});
test("disabled cleanup returns empty", () => {
assert.deepEqual(selectStaleQueuedLogs([row()], NOW, -1), []);
});
test("fresh queued row is kept", () => {
const fresh = row({ id: "b".repeat(32), createdAt: "2026-07-09T23:00:00.000Z" });
assert.deepEqual(selectStaleQueuedLogs([fresh], NOW, 3600), []);
});
test("stale queued row with delivery record is kept", () => {
const staleWithDelivery = row({ id: "c".repeat(32), hasDelivery: true });
assert.deepEqual(selectStaleQueuedLogs([staleWithDelivery], NOW, 3600), []);
});
test("stale queued row without delivery record is deleted", () => {
const stale = row({ id: "d".repeat(32) });
assert.deepEqual(selectStaleQueuedLogs([stale], NOW, 3600), ["d".repeat(32)]);
});
test("success and failed rows are ignored", () => {
const success = row({ id: "e".repeat(32), deliveryStatus: "success" });
const failed = row({ id: "f".repeat(32), deliveryStatus: "failed" });
assert.deepEqual(selectStaleQueuedLogs([success, failed], NOW, 3600), []);
});Case studies
The store that restarted its worker every deploy
A mid-size store deployed several times a week, and each deploy briefly killed the messenger:consume process before the new one started. Most in-flight webhook messages survived, since Symfony Messenger requeues on a clean restart, but a handful landed in that exact restart window and were lost outright. Nothing failed loudly, the rows just sat at queued.
Nobody noticed until the webhook_event_log table had grown past a million rows and a slow admin query pointed at it. Running the script in dry run first surfaced exactly which rows were truly orphaned versus still legitimately recent, and a scheduled daily run has kept the table flat ever since.
The app that was uninstalled mid-delivery
A merchant uninstalled a third party app while it had several webhook deliveries in flight. The receiving endpoint disappeared along with the app, so those dispatched messages could never report a result, and their log rows stayed queued indefinitely, alongside a much smaller number of totally normal in-flight rows from other still-active apps.
The delivery record cross-check was what made the difference here. A naive "delete everything old and queued" pass would have been fine in this case, but the delivery check meant the team could trust the script even while other integrations had real in-flight retries at the same time.
After this runs on a schedule, webhook_event_log stops accumulating rows that will never resolve, without ever touching a row that might still be genuinely in progress. The table stays close to the size the merchant's own entryLifetimeSeconds setting implies, admin queries against it stay fast, and if cleanup is disabled on purpose, the script respects that and just tells you what it found instead of guessing.
FAQ
Why do Shopware webhook_event_log rows get stuck at queued?
Shopware writes the row with deliveryStatus queued the instant a webhook message is dispatched to the message queue, before any consumer has touched it. It only flips to success or failed once a worker actually processes that message. If the message is lost, dropped, or never consumed, for example no worker is running, a serialization error happens, the receiving app is deactivated mid-flight, or the message is simply never picked up, the log row stays at queued permanently.
Does Shopware clean up queued webhook_event_log rows on its own?
Not before Shopware 6.7.4. The built-in webhook_event_log.cleanup scheduled task used core.webhook.entryLifetimeSeconds to delete old rows, but it only ever matched success or failed rows. Queued rows were explicitly excluded from that cleanup, so on any store with an unreliable queue transport they just accumulate forever until you upgrade or clean them up yourself.
Is it safe to delete stale queued webhook_event_log rows with a script?
Yes, as long as you check core.webhook.entryLifetimeSeconds first, skip deletion entirely when it is -1 (cleanup disabled by merchant choice), only touch rows older than twice that lifetime, and skip any row that still has a corresponding webhook_delivery record, since that means it may still be legitimately in flight or retrying. deliveryStatus is a plain string field, not a state machine, so deleting the row is the correct action, matching Shopware's own upstream fix.
Related field notes
Citations
On the problem:
- shopware/shopware: fix-webhook-cleanup-for-queued-messages changelog (6.7.4.0). github.com/shopware/shopware changelog 6.7.4.0
- shopware/shopware: only-cleanup-successfully-delivered-webhooks changelog (6.6.5.0). github.com/shopware/shopware changelog 6.6.5.0
- Shopware Community Forum: Tabelle webhook_event_log. forum.shopware.com/t/tabelle-webhook-event-log/93672
On the solution:
- Shopware 6 Documentation: Message Queue and Scheduled Tasks (Tutorials and FAQ). docs.shopware.com/en/shopware-6-en/tutorials-and-faq/message-queue-and-scheduled-tasks
- Shopware Developer Documentation: Scheduled Task. developer.shopware.com/docs/guides/hosting/infrastructure/scheduled-task.html
- Shopware Developer Documentation: Webhook. developer.shopware.com/docs/guides/plugins/apps/webhook.html
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 webhook log backlog?
If this saved you from a bloated table or a slow admin query, 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