Skip to content

servicebus

azure_bootstrap.servicebus

Tier 3 Service Bus helpers: consumer watchdog, DLQ growth alarm, daily digest.

Modules:

Name Description
async_ext

Service Bus extensions — async consumer, replay guard, multi-queue router.
consumer

Service-Bus-flavored re-exports of heartbeat primitives.
consumer_wrapper

End-to-end Service Bus message handler.
dlq_alarm

Dead-letter queue growth-rate alarm.
dlq_digest

Daily DLQ digest emails with embedded resubmit link + pending-alert summary.

Classes:

Name Description
MultiQueueRouter

Route sends to named queues.
ReplayGuard

Bounded idempotency cache for message deduplication.
InvalidResubmitToken

Resubmit token is invalid (alias of InvalidActionToken).

Functions:

Name Description
run_async_consumer

Async receive loop with graceful shutdown on stop_event.
service_bus_transport_type

Return amqp or websocket from SERVICE_BUS_TRANSPORT_TYPE.
record_consumer_iteration

Stamp the last-iteration time. Best-effort, never raises.
record_message_settled

Stamp the last-settled time. Best-effort, never raises.
start_consumer_watchdog

Spawn a daemon thread that alerts when consumer iteration stalls.
handle_message

Handle one received message end-to-end.
check_dlq_growth_rate

Peek the DLQ, compare to the previous sample, alert on excessive growth.
reset_state

Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1.
build_dlq_digest_body

5-column escaped HTML table: attachment, sender, dlq_time, reason, detail.
run_dlq_digest

Daily DLQ digest + pending-alerts summary email.

MultiQueueRouter 

MultiQueueRouter(client: Any)

Route sends to named queues.

Source code in azure_bootstrap/servicebus/async_ext.py 
80
81
82
def __init__(self, client: Any) -> None:
    self._client = client
    self._senders: dict[str, Any] = {}

ReplayGuard 

ReplayGuard(*, max_size: int = 10000, ttl_seconds: float = 3600.0)

Bounded idempotency cache for message deduplication.

Source code in azure_bootstrap/servicebus/async_ext.py 
28
29
30
31
32
def __init__(self, *, max_size: int = 10_000, ttl_seconds: float = 3600.0) -> None:
    self._max_size = max_size
    self._ttl = ttl_seconds
    self._seen: OrderedDict[str, float] = OrderedDict()
    self._lock = threading.Lock()

InvalidResubmitToken

Bases: InvalidActionToken

Resubmit token is invalid (alias of InvalidActionToken).

run_async_consumer async 

run_async_consumer(client: Any, queue_name: str, handler: MessageHandler, *, stop_event: Event | None = None, max_wait: float = 5.0) -> None

Async receive loop with graceful shutdown on stop_event.

Source code in azure_bootstrap/servicebus/async_ext.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def run_async_consumer(
    client: Any,
    queue_name: str,
    handler: MessageHandler,
    *,
    stop_event: asyncio.Event | None = None,
    max_wait: float = 5.0,
) -> None:
    """Async receive loop with graceful shutdown on *stop_event*."""
    receiver = client.get_queue_receiver(queue_name=queue_name)
    stop = stop_event or asyncio.Event()
    async with receiver:
        while not stop.is_set():
            messages = await receiver.receive_messages(max_wait_time=max_wait)
            for msg in messages:
                if stop.is_set():
                    await receiver.abandon_message(msg)
                    break
                try:
                    await handler(msg)
                    await receiver.complete_message(msg)
                except Exception:
                    _logger.exception("async consumer handler failed")
                    await receiver.abandon_message(msg)

service_bus_transport_type 

service_bus_transport_type() -> str

Return amqp or websocket from SERVICE_BUS_TRANSPORT_TYPE.

Source code in azure_bootstrap/servicebus/async_ext.py 
19
20
21
22
def service_bus_transport_type() -> str:
    """Return ``amqp`` or ``websocket`` from ``SERVICE_BUS_TRANSPORT_TYPE``."""
    mode = os.environ.get("SERVICE_BUS_TRANSPORT_TYPE", "amqp").lower()
    return "websocket" if mode == "websocket" else "amqp"

record_consumer_iteration 

record_consumer_iteration() -> None

Stamp the last-iteration time. Best-effort, never raises.

Source code in azure_bootstrap/heartbeat/__init__.py 
50
51
52
53
54
55
56
57
58
59
def record_consumer_iteration() -> None:
    """Stamp the last-iteration time. Best-effort, never raises."""
    global _last_consumer_iteration_at
    try:
        with _state_lock:
            _last_consumer_iteration_at = time.monotonic()
        if _logger.isEnabledFor(logging.DEBUG):
            _logger.debug("consumer_iteration recorded")
    except Exception:
        pass

record_message_settled 

record_message_settled() -> None

Stamp the last-settled time. Best-effort, never raises.

Source code in azure_bootstrap/heartbeat/__init__.py 
40
41
42
43
44
45
46
47
def record_message_settled() -> None:
    """Stamp the last-settled time. Best-effort, never raises."""
    global _last_message_settled_at
    try:
        with _state_lock:
            _last_message_settled_at = time.monotonic()
    except Exception:
        pass

start_consumer_watchdog 

start_consumer_watchdog(stop_event: Event, *, interval_seconds: float | None = None, silence_threshold_seconds: float | None = None, resilence_window_seconds: float | None = None) -> Thread

Spawn a daemon thread that alerts when consumer iteration stalls.

The default resilence_window (1 hour) is intentionally longer than the alerts dispatcher's 10-min dedup so sustained incidents page hourly, not every 10 minutes.

Source code in azure_bootstrap/heartbeat/__init__.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def start_consumer_watchdog(
    stop_event: threading.Event,
    *,
    interval_seconds: float | None = None,
    silence_threshold_seconds: float | None = None,
    resilence_window_seconds: float | None = None,
) -> threading.Thread:
    """Spawn a daemon thread that alerts when consumer iteration stalls.

    The default ``resilence_window`` (1 hour) is intentionally longer than
    the alerts dispatcher's 10-min dedup so sustained incidents page hourly,
    not every 10 minutes.
    """
    if interval_seconds is None:
        interval_seconds = _env_float("WATCHDOG_INTERVAL_SECONDS", 60.0)
    if silence_threshold_seconds is None:
        silence_threshold_seconds = _env_float("WATCHDOG_SB_SILENCE_SECONDS", 1800.0)
    if resilence_window_seconds is None:
        resilence_window_seconds = _env_float("WATCHDOG_RESILENCE_SECONDS", 3600.0)
    if interval_seconds <= 0:
        return _make_inert_thread("azure-bootstrap-watchdog-disabled")

    def _loop() -> None:
        global _last_watchdog_alert_at
        while not stop_event.wait(interval_seconds):
            try:
                age = _last_iteration_age_seconds()
                if age is None or age <= silence_threshold_seconds:
                    continue
                now = time.monotonic()
                with _state_lock:
                    last_alert = _last_watchdog_alert_at
                if last_alert != 0.0 and (now - last_alert) < resilence_window_seconds:
                    continue
                _logger.warning(
                    "SB consumer loop has not progressed",
                    extra={
                        "operation": "watchdog_tick",
                        "silence_seconds": int(age),
                        "threshold_seconds": silence_threshold_seconds,
                    },
                )
                try:
                    from azure_bootstrap.alerts import AlertSeverity, alert_dev_team

                    alert_dev_team(
                        AlertSeverity.ERROR,
                        subject="SB consumer silent — consumer loop is not progressing",
                        context={
                            "silence_seconds": int(age),
                            "threshold_seconds": silence_threshold_seconds,
                            "last_sb_settle_age_seconds": _last_settle_age_seconds(),
                        },
                        dedup_key="watchdog:consumer_silent",
                    )
                except Exception:
                    pass
                with _state_lock:
                    _last_watchdog_alert_at = now
            except Exception:
                _logger.exception("watchdog tick failed")

    t = threading.Thread(target=_loop, name="azure-bootstrap-watchdog", daemon=True)
    t.start()
    return t

handle_message 

handle_message(receiver: SbReceiverProtocol, msg: Any, processor: MessageProcessor, *, schema: MessageSchema | None = None, correlation_field: str = 'correlation_id', extra_correlation_fields: tuple[str, ...] = (), source: str = 'consumer', lock_renewer: Any | None = None, counter_namespace: str = 'sb') -> tuple[bool, bool]

Handle one received message end-to-end.

Returns (processed: bool, failed: bool).

Source code in azure_bootstrap/servicebus/consumer_wrapper.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def handle_message(
    receiver: SbReceiverProtocol,
    msg: Any,
    processor: MessageProcessor,
    *,
    schema: MessageSchema | None = None,
    correlation_field: str = "correlation_id",
    extra_correlation_fields: tuple[str, ...] = (),
    source: str = "consumer",
    lock_renewer: Any | None = None,
    counter_namespace: str = "sb",
) -> tuple[bool, bool]:
    """Handle one received message end-to-end.

    Returns ``(processed: bool, failed: bool)``.
    """
    processed = False
    failed = False
    try:
        if lock_renewer is not None:
            try:
                lock_renewer.register_message(receiver, msg, max_lock_renewal_duration=3600)
            except Exception:
                _logger.warning("lock_renewer registration failed", exc_info=True)

        # 1. Parse JSON body
        raw = _msg_body(msg)
        try:
            if isinstance(raw, bytes):
                payload = json.loads(raw.decode("utf-8"))
            elif isinstance(raw, str):
                payload = json.loads(raw)
            else:
                raise InvalidMessageError(f"unsupported body type: {type(raw).__name__}")
        except (json.JSONDecodeError, UnicodeDecodeError, InvalidMessageError) as exc:
            bump_counter(f"{counter_namespace}.dead_lettered")
            _settle(
                receiver,
                msg,
                action="dead_letter",
                reason="invalid_json",
                description=str(exc)[:200],
            )
            return False, True

        # 2. Schema validation
        if schema is not None:
            try:
                payload = validate_message(payload, schema)
            except InvalidMessageError as exc:
                bump_counter(f"{counter_namespace}.dead_lettered")
                _settle(
                    receiver,
                    msg,
                    action="dead_letter",
                    reason=type(exc).__name__,
                    description=str(exc)[:200],
                )
                return False, True

        # 3. Correlation context + processor invocation
        fields: dict[str, str] = {}
        for name in extra_correlation_fields:
            value = payload.get(name)
            if isinstance(value, str) and value:
                fields[name] = value
        cid = payload.get(correlation_field) if isinstance(payload, dict) else None

        with correlation_scope(cid if isinstance(cid, str) and cid else None, **fields):
            try:
                processor.process(payload)
            except BaseException as exc:
                if is_unrecoverable(exc):
                    bump_counter(f"{counter_namespace}.dead_lettered")
                    try:
                        processor.notify_failure(payload, exc)  # type: ignore[arg-type]
                    except Exception:
                        _logger.exception(
                            "processor.notify_failure raised — proceeding with dead-letter"
                        )
                    _settle(
                        receiver,
                        msg,
                        action="dead_letter",
                        reason=type(exc).__name__,
                        description=str(exc)[:200],
                    )
                    try:
                        from azure_bootstrap.alerts import AlertSeverity, alert_dev_team

                        alert_dev_team(
                            AlertSeverity.ERROR,
                            subject=(f"SB dead-lettered ({source}): " f"{type(exc).__name__}"),
                            context={
                                "exception_type": type(exc).__name__,
                                "error": str(exc)[:300],
                            },
                            dedup_key=f"sb.dead_lettered:{type(exc).__name__}",
                        )
                    except Exception:
                        pass
                    return False, True

                # Transient: abandon
                bump_counter(f"{counter_namespace}.abandoned")
                _settle(receiver, msg, action="abandon")
                try:
                    from azure_bootstrap.alerts import AlertSeverity, alert_dev_team

                    alert_dev_team(
                        AlertSeverity.ERROR,
                        subject=(f"SB abandoned ({source}): {type(exc).__name__}"),
                        context={
                            "exception_type": type(exc).__name__,
                            "error": str(exc)[:300],
                        },
                        dedup_key=f"sb.abandoned:{type(exc).__name__}",
                    )
                except Exception:
                    pass
                return False, True

            # 4. Success
            bump_counter(f"{counter_namespace}.completed")
            _settle(receiver, msg, action="complete")
            processed = True
            return True, False
    finally:
        if lock_renewer is not None:
            try:
                lock_renewer.close()
            except Exception:
                pass
        record_message_settled()
    return processed, failed

check_dlq_growth_rate 

check_dlq_growth_rate(service_bus_repo: SbRepoProtocol, *, alert_threshold: int = 5, sample_window_minutes: int = 60) -> dict[str, int]

Peek the DLQ, compare to the previous sample, alert on excessive growth.

Returns {'current': N, 'delta': N, 'alerted': 0 or 1}.

Source code in azure_bootstrap/servicebus/dlq_alarm.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def check_dlq_growth_rate(
    service_bus_repo: SbRepoProtocol,
    *,
    alert_threshold: int = 5,
    sample_window_minutes: int = 60,
) -> dict[str, int]:
    """Peek the DLQ, compare to the previous sample, alert on excessive growth.

    Returns ``{'current': N, 'delta': N, 'alerted': 0 or 1}``.
    """
    global _last_sample_ts, _last_sample_count
    try:
        messages = service_bus_repo.peek_dead_letter_messages(max_count=500)
        current = len(messages)
    except Exception:
        _logger.exception("dlq_alarm: peek failed")
        return {"current": 0, "delta": 0, "alerted": 0}

    now = time.monotonic()
    window_seconds = sample_window_minutes * 60
    delta = 0
    with _lock:
        if _last_sample_ts is not None and (now - _last_sample_ts) <= window_seconds:
            delta = current - (_last_sample_count or 0)
        _last_sample_ts = now
        _last_sample_count = current

    alerted = 0
    if delta > alert_threshold:
        try:
            from azure_bootstrap.alerts import AlertSeverity, alert_dev_team

            alert_dev_team(
                AlertSeverity.CRITICAL,
                subject=(
                    f"DLQ growth rate exceeded threshold " f"(+{delta} in {sample_window_minutes}m)"
                ),
                context={
                    "current_depth": current,
                    "delta": delta,
                    "window_minutes": sample_window_minutes,
                    "threshold": alert_threshold,
                },
                dedup_key="dlq_alarm.growth_rate_exceeded",
            )
            alerted = 1
        except Exception:
            pass
    return {"current": current, "delta": delta, "alerted": alerted}

reset_state 

reset_state() -> None

Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1.

Source code in azure_bootstrap/servicebus/dlq_alarm.py 
27
28
29
30
31
32
33
34
def reset_state() -> None:
    """Test-only. Refuses unless AZURE_BOOTSTRAP_ALLOW_RESET=1."""
    if os.environ.get("AZURE_BOOTSTRAP_ALLOW_RESET") != "1":
        raise RuntimeError("reset_state is test-only — set AZURE_BOOTSTRAP_ALLOW_RESET=1")
    global _last_sample_ts, _last_sample_count
    with _lock:
        _last_sample_ts = None
        _last_sample_count = None

build_dlq_digest_body 

build_dlq_digest_body(entries: list[dict[str, Any]], resubmit_url: str | None) -> str

5-column escaped HTML table: attachment, sender, dlq_time, reason, detail.

Includes a "Resubmit all" button when resubmit_url is provided, else a CLI hint.

Source code in azure_bootstrap/servicebus/dlq_digest.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def build_dlq_digest_body(
    entries: list[dict[str, Any]],
    resubmit_url: str | None,
) -> str:
    """5-column escaped HTML table: attachment, sender, dlq_time, reason, detail.

    Includes a "Resubmit all" button when ``resubmit_url`` is provided, else a
    CLI hint.
    """
    rows: list[str] = []
    for e in entries:
        attachment = escape(str(e.get("attachment_name") or e.get("subject") or ""))
        sender = escape(str(e.get("sender") or ""))
        ts = escape(str(e.get("dlq_time") or e.get("enqueued_time_utc") or ""))
        reason = escape(str(e.get("dead_letter_reason") or e.get("reason") or ""))
        detail = escape(str(e.get("dead_letter_error_description") or e.get("detail") or "")[:200])
        rows.append(
            "<tr>"
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{attachment}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{sender}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{ts}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{reason}</td>'
            f'<td style="padding:4px 8px;border:1px solid #ddd;">{detail}</td>'
            "</tr>"
        )
    table_body = "".join(rows) or (
        '<tr><td colspan="5" style="padding:8px;text-align:center;color:#888;">'
        "No dead-letter messages.</td></tr>"
    )
    if resubmit_url:
        action_block = (
            '<p style="margin-top:16px;">'
            f'<a href="{escape(resubmit_url)}" '
            'style="background:#0078d4;color:#fff;padding:10px 16px;'
            'text-decoration:none;border-radius:4px;">Resubmit all</a></p>'
        )
    else:
        action_block = (
            '<p style="margin-top:16px;color:#888;font-size:12px;">'
            "To resubmit, run the DLQ resubmit CLI with a valid API key.</p>"
        )
    return (
        '<html><body style="font-family:Helvetica,Arial,sans-serif;color:#333;">'
        f"<h2>Dead-letter queue digest ({len(entries)} messages)</h2>"
        '<table style="border-collapse:collapse;border:1px solid #ddd;width:100%;">'
        "<thead><tr>"
        '<th style="padding:4px 8px;border:1px solid #ddd;">Attachment</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Sender</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">DLQ&rsquo;d at</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Reason</th>'
        '<th style="padding:4px 8px;border:1px solid #ddd;">Detail</th>'
        "</tr></thead><tbody>" + table_body + "</tbody></table>" + action_block + "</body></html>"
    )

run_dlq_digest 

run_dlq_digest(service_bus_repo: Any, email_repo: EmailRepoProtocol, *, dev_recipients: Iterable[str], api_key: str, public_base_url: str, max_peek: int = 50, subject_prefix: str = '[DLQ digest]') -> dict[str, Any]

Daily DLQ digest + pending-alerts summary email.

Source code in azure_bootstrap/servicebus/dlq_digest.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@traced(operation="dlq_digest.run", alert_on_error="error")
def run_dlq_digest(
    service_bus_repo: Any,
    email_repo: EmailRepoProtocol,
    *,
    dev_recipients: Iterable[str],
    api_key: str,
    public_base_url: str,
    max_peek: int = 50,
    subject_prefix: str = "[DLQ digest]",
) -> dict[str, Any]:
    """Daily DLQ digest + pending-alerts summary email."""
    from azure_bootstrap.alerts import (
        drain_pending_alerts,
        render_pending_alerts_html,
    )

    try:
        entries = service_bus_repo.peek_dead_letter_messages(max_count=max_peek) or []
    except Exception:
        _logger.exception("dlq_digest: peek failed")
        entries = []

    pending_alerts = drain_pending_alerts()
    recipients = [r.strip() for r in dev_recipients if r and r.strip()]

    if not entries and not pending_alerts:
        return {
            "dlq_count": 0,
            "email_sent": False,
            "skipped_reason": "empty",
            "pending_alert_count": 0,
        }
    if not recipients:
        _logger.warning("dlq_digest: no recipients, skipping")
        return {
            "dlq_count": len(entries),
            "email_sent": False,
            "skipped_reason": "no_recipients",
            "pending_alert_count": len(pending_alerts),
        }

    resubmit_url: str | None = None
    if api_key and public_base_url:
        token = issue_resubmit_token(api_key)
        resubmit_url = f"{public_base_url.rstrip('/')}/dlq/resubmit?token={token}"
    body = build_dlq_digest_body(entries, resubmit_url)
    if pending_alerts:
        body += render_pending_alerts_html(pending_alerts)

    subject = f"{subject_prefix} {len(entries)} DLQ · {len(pending_alerts)} batched"
    email_repo.send_email(recipients, subject, body)
    return {
        "dlq_count": len(entries),
        "email_sent": True,
        "skipped_reason": None,
        "pending_alert_count": len(pending_alerts),
    }