Logistics Cleanup: Implementing Public HTTPS Queue Push Subscriber Recovery

Bottom line: For a Node.js queue push webhook subscriber on a public HTTPS endpoint, verify the signature over the untouched request bytes, durably record the delivery, and ack only that record; run the logistics cleanup later in a retryable worker.

The HTTP response should mean “accepted into local custody,” not “every object has been deleted.” That boundary keeps a slow warehouse cleanup from holding a web request open, and it gives operators something durable to inspect after a process restart. A successful cleanup and a successful webhook are separate state transitions.

This is the constraint that drives the design: a daily job finds expired proof-of-delivery images and shipping-label exports, but deleting ten thousand objects may take minutes. The queue can redeliver. The process can stop between any two writes. An object delete may succeed while the database update doesn't. Recovery, rather than nominal throughput, decides where the acknowledgment belongs.

How does a Node.js API verify public HTTPS queue push signatures before ack?

Use a small, explicit delivery contract. The producer sends X-Delivery-ID, X-Delivery-Timestamp, and X-Delivery-Signature; the signature is an HMAC-SHA256 over the timestamp, a period, and the exact body bytes. The subscriber rejects a malformed timestamp with 400, an old delivery or bad signature with 401, and an invalid payload with 422. It returns 204 only after inserting the message into a durable inbox, or after finding that the same delivery ID is already there.

The raw bytes matter. Parsing JSON and serializing it again can change whitespace or key order, so verification must happen before a body parser replaces the original representation. In Node.js, capture a raw Buffer for the route and use a constant-time comparison; the Python below spells out the same wire contract because every code sample in this article uses Python. The protocol does not depend on either runtime.

import hashlib
import hmac
import json
import time
from dataclasses import dataclass


MAX_AGE_SECONDS = 300


@dataclass(frozen=True)
class VerifiedDelivery:
    delivery_id: str
    payload: dict


def sign(secret: bytes, timestamp: str, body: bytes) -> str:
    signed = timestamp.encode("ascii") + b"." + body
    digest = hmac.new(secret, signed, hashlib.sha256).hexdigest()
    return f"sha256={digest}"


def verify_delivery(headers: dict[str, str], body: bytes, secret: bytes) -> VerifiedDelivery:
    delivery_id = headers.get("x-delivery-id", "")
    timestamp = headers.get("x-delivery-timestamp", "")
    supplied = headers.get("x-delivery-signature", "")
    if not delivery_id or not timestamp or not supplied:
        raise ValueError("400 missing delivery headers")

    try:
        sent_at = int(timestamp)
    except ValueError as exc:
        raise ValueError("400 invalid delivery timestamp") from exc

    if abs(int(time.time()) - sent_at) > MAX_AGE_SECONDS:
        raise PermissionError("401 expired delivery")

    expected = sign(secret, timestamp, body)
    if not hmac.compare_digest(expected, supplied):
        raise PermissionError("401 invalid signature")

    try:
        payload = json.loads(body)
    except json.JSONDecodeError as exc:
        raise ValueError("422 invalid JSON") from exc

    return VerifiedDelivery(delivery_id=delivery_id, payload=payload)

Five minutes is an example policy in this contract, not a universal queue limit. Clock skew, expected delivery latency, and replay exposure should determine the actual window. I'm not sure what delay bound applies to your queue because that is provider- and topology-specific; its delivery documentation and production latency histogram are the evidence needed to set it. Keep the timestamp check independent from delivery-ID deduplication: one limits replay time, while the other makes a valid retry harmless.

Don't put the signing secret in the payload or a query string. Rotate it by accepting a current and previous key for a bounded overlap, identify the key with a non-secret header, and remove the old key after the longest permitted delivery delay. Authentication also doesn't replace network controls: terminate HTTPS at a managed ingress or reverse proxy, cap request size, and apply rate limits before the handler allocates memory for an arbitrary body.

Put custody state beside retained object data

The inbox insert and the 204 response form the critical handoff. A relational database is useful here because a unique key turns redelivery into an ordinary no-op, and because the cleanup state can be queried without scraping application logs. The message should carry a cleanup run ID and selection boundary, such as a warehouse and an expired_before timestamp; it should not contain a giant list of object keys that can exceed transport limits or become stale during retries.

This minimal handler uses SQLite to make the transaction visible. A production service can use another transactional database, but the invariant stays fixed: commit before ack. The schema's primary key deduplicates the transport delivery, while run_id deduplicates the business operation when a scheduler accidentally publishes the same cleanup twice.

import json
import sqlite3


SCHEMA = """
CREATE TABLE IF NOT EXISTS cleanup_inbox (
    delivery_id TEXT PRIMARY KEY,
    run_id TEXT NOT NULL UNIQUE,
    payload_json TEXT NOT NULL,
    state TEXT NOT NULL CHECK (state IN ('pending', 'running', 'done')),
    attempts INTEGER NOT NULL DEFAULT 0,
    last_error TEXT,
    received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)
"""


def accept_delivery(
    db: sqlite3.Connection,
    headers: dict[str, str],
    raw_body: bytes,
    secret: bytes,
) -> int:
    delivery = verify_delivery(headers, raw_body, secret)
    payload = delivery.payload
    required = {"run_id", "warehouse_id", "expired_before"}
    if not required.issubset(payload):
        return 422

    with db:
        db.execute(
            """
            INSERT OR IGNORE INTO cleanup_inbox
                (delivery_id, run_id, payload_json, state)
            VALUES (?, ?, ?, 'pending')
            """,
            (delivery.delivery_id, payload["run_id"], json.dumps(payload)),
        )
    return 204

There is a deliberately small promise behind that response. It says the subscriber can recover the work from its database. It says nothing about how soon the object store will be swept, how many objects qualify, or whether a downstream delete has completed. Consider the awkward window around a label stored as wh-17/parcel-8041.pdf: the worker selects version g42, deletes it, and stops before updating the inbox; meanwhile a corrected label is written under the same human-readable key as version g43. A retry that blindly deletes by key can erase the correction, while a retry that asks to delete the originally selected version can safely observe that g42 is absent and continue. This is why the inbox must preserve a stable cutoff and the storage adapter must preserve version identity. Deduplicating the incoming delivery ID alone cannot protect a mutable object namespace.

Ack earlier and a process stop can lose the run. Ack after cleanup and a normal timeout can trigger parallel redeliveries of the whole run. The middle boundary is less dramatic, but operationally much cleaner.

Test every interrupted cleanup and retention boundary

An inbox alone prevents transport duplicates; it doesn't make side effects atomic. The worker therefore needs a stable selection boundary, bounded batches, idempotent deletion semantics in the storage adapter, and a checkpoint that advances only after each batch is confirmed. If the object store and database cannot participate in one transaction — the usual shape of this problem — recovery must tolerate the gap instead of pretending it has exactly-once execution.

Treat “already absent” as a completed object-level outcome. Use conditional deletion if the storage system exposes an object version or generation, because deleting by a reused key without a version guard can remove a newer shipment artifact. Retention policy belongs in the query too: legal holds, dispute evidence, and active manifests must be excluded before any delete call. The dangerous failure isn't a duplicate request. It is a retry acting on a different object than the one originally selected.

import json
import sqlite3
from typing import Protocol


class CleanupStore(Protocol):
    def list_expired(
        self, warehouse_id: str, expired_before: str, limit: int
    ) -> list[tuple[str, str]]:
        """Return (object_key, version) pairs not protected by retention."""

    def delete_version(self, object_key: str, version: str) -> None:
        """Delete the selected version; an already absent version counts as success."""


def run_one(db: sqlite3.Connection, store: CleanupStore, batch_size: int = 200) -> bool:
    row = db.execute(
        """
        SELECT delivery_id, payload_json
        FROM cleanup_inbox
        WHERE state IN ('pending', 'running')
        ORDER BY received_at
        LIMIT 1
        """
    ).fetchone()
    if row is None:
        return False

    delivery_id, payload_json = row
    payload = json.loads(payload_json)
    with db:
        db.execute(
            """
            UPDATE cleanup_inbox
            SET state = 'running', attempts = attempts + 1,
                updated_at = CURRENT_TIMESTAMP
            WHERE delivery_id = ?
            """,
            (delivery_id,),
        )

    objects = store.list_expired(
        payload["warehouse_id"], payload["expired_before"], batch_size
    )
    for object_key, version in objects:
        store.delete_version(object_key, version)

    if objects:
        return True

    with db:
        db.execute(
            """
            UPDATE cleanup_inbox
            SET state = 'done', last_error = NULL,
                updated_at = CURRENT_TIMESTAMP
            WHERE delivery_id = ?
            """,
            (delivery_id,),
        )
    return True

The example intentionally keeps object enumeration behind a Protocol; object listing, version tokens, retention controls, and conditional deletion differ by storage implementation. It also avoids claiming that a queue can manufacture exactly-once side effects. AWS documents that FIFO queues use a five-minute deduplication interval, and messages with the same deduplication ID sent after that interval are treated as new. That is useful transport behavior, but a cleanup run still needs its own durable run_id because business retries can outlive a queue's deduplication window.

Recovery should be boring.

On startup, workers scan both pending and running rows, resume them, and rely on version-qualified deletes to absorb repeated work. Record the attempt count and last failure category, but don't let an unbounded poison run monopolize the worker; after a configured attempt ceiling, move it to an operator-visible terminal state in the real schema and page on run age. The precise ceiling depends on the cleanup deadline and escalation coverage, so copying a generic number would be false precision.

Failure point	Queue-visible result	Durable state	Recovery action
Before signature verification	No ack	None	Reject; allow authenticated redelivery
After verification, before commit	No ack	None	Redelivery inserts the inbox row
After commit, before 204 arrives	Retry may occur	pending	Unique delivery ID returns 204 again
During an object batch	Already acked	running	Resume and repeat version-qualified deletes
After final delete, before done	Already acked	running	Empty rescan proves completion, then mark done

The catch is latency: this pattern adds a database write before every acknowledgment and requires an inbox sweeper, retention for old rows, and an operational view for stuck work. It is not suitable when the caller requires the HTTP response to contain the cleanup result. In that case, use an asynchronous operation resource that the caller polls, or keep a synchronous endpoint only when the work is strictly bounded below every relevant timeout. Stick with a simple local scheduler when one process owns the job, missed runs are acceptable, and there is no need for cross-host recovery; a public push endpoint would add attack surface without buying useful durability.

Roll out through one warehouse with a recovery query

The scheduler is less important than its failure semantics. A hosted cron trigger can enqueue a run, a database-backed scheduler can claim due rows, and a source-control workflow can make occasional maintenance visible in the repository. GitHub documents that scheduled workflows run from the latest commit on the default branch, can be delayed during high load, and may be dropped under sufficiently high load; those boundaries make that mechanism a poor fit for a cleanup deadline that needs explicit replay and custody, though it can still suit non-critical housekeeping.

Approach	Recovery record	Public endpoint	Best fit	Limitation
Local process timer	Process-local unless added	No	One host, loose deadlines	Restart and leadership need explicit handling
Repository-scheduled workflow	Workflow run history	No	Auditable, non-critical maintenance	Schedule delay or dropped runs can violate a hard deadline
Queue push plus durable inbox	Inbox row and queue delivery	Yes	Cross-host recovery and slow cleanup	More security and database operations
Database due-row polling	Due row and lease	No	Existing transactional control plane	Polling and lease expiry need tuning

Roll out in three compact stages. First, deploy the endpoint in record-only mode against signed synthetic deliveries and prove that replaying one delivery ID creates one inbox row. Next, run the worker in dry-run mode, comparing selected object versions with retention exclusions and measuring queue-to-inbox age plus inbox-to-completion age. Finally, enable deletion for one warehouse, set alerts on oldest pending or running row, and exercise restart recovery between a delete and its completion update.

Keep the old cleanup path available until the new worker has completed enough scheduled windows to expose timing and retention mistakes. Then disable the old producer before removing its consumer; two schedulers generating different run IDs defeat business-level deduplication even when each queue delivery is individually safe.

Ship the recovery query with the endpoint.

References

• https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-fifo-queues.html
• https://docs.github.com/en/actions/using-workflows/events-that-trigger-workflows