Observability

OpenTelemetry helpers and structlog setup.

obs

Vendor-neutral observability bootstrap.

The library never imports vendor SDKs (Sentry, Datadog, PostHog, App Insights). Instead, it emits OpenTelemetry spans and metrics and lets operators wire any OTLP-compatible backend by configuring environment variables (or by passing an already-configured TracerProvider / MeterProvider).

Logs are NOT an OTel signal here: the library attaches no LoggingHandler and emits no log records through a LoggerProvider. :func:setup_logging configures structlog over the stdlib logging root logger, so log lines reach a telemetry backend only if the operator has attached a handler to that root logger themselves -- which is what configure_azure_monitor() does. That is incidental wiring, not an emission path this library owns, and it is why exception text has to be scrubbed inside the processor chain rather than at an exporter (see _redact_exc). (:func:configure_exporters lets the SDK's configurator install a LoggerProvider alongside the tracer and meter providers, as opentelemetry-instrument would; nothing in the library writes to it.)

Under the taskq worker CLI the standard OTEL_* exporter variables are enough: :func:configure_exporters installs SDK providers from them at startup when the [otel] extra is installed and no provider is set yet (see _exporter). Embedding applications call it themselves or configure the SDK directly.

Common deployment shapes:

• Datadog Agent, accepts OTLP on localhost:4317. Set OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317.
• Sentry, Sentry Spotlight / Sentry OTel ingest, ditto.
• App Insights, set OTEL_EXPORTER_OTLP_ENDPOINT to the Azure Monitor connection string-derived OTLP URL (typically via the App Insights agent).
• PostHog, currently via PostHog Cloud OTLP endpoint, same env var.

For error reporting that doesn't fit OTel exception events (e.g., DLQ routing to Sentry), users implement the ErrorReporter Protocol as a DI provider, vendor-neutral and added with the observability surface.

The library depends only on opentelemetry-api at runtime. Operators who want to configure providers programmatically (or use the in-process testing utilities in taskq.testing) install the [otel] extra::

pip install "taskq-py[otel]"

which pulls in opentelemetry-sdk and opentelemetry-exporter-otlp. For Prometheus scrapes, use the [prometheus] extra instead.

Semconv compliance: the library uses spec-compliant messaging semconv attribute names (messaging.operation.type=publish, messaging.operation.type=process, messaging.consumer.group.name, etc.) so that operators who set OTEL_SEMCONV_STABILITY_OPT_IN=messaging get consistent behavior. No runtime conditional branching on this env var is needed, the attribute values are correct by construction.

INSTRUMENTATION_NAME module-attribute

INSTRUMENTATION_NAME: str = 'taskq'

__all__ module-attribute

__all__ = [
    "INSTRUMENTATION_NAME",
    "ConsumedOutcome",
    "ErrorReporter",
    "ErrorReporterType",
    "ExceptionText",
    "ExporterWiring",
    "NullErrorReporter",
    "OtelExporterConfigurationError",
    "ScrubbedText",
    "StrandedReason",
    "TimeoutKind",
    "bind_job_context",
    "claim_health_snapshot",
    "configure_exporters",
    "get_logger",
    "get_meter",
    "get_tracer",
    "invoke_error_reporter",
    "log_cancel_phase_change",
    "log_state_change",
    "otel_enabled",
    "reconcile_cron_failures",
    "record_archived_jobs",
    "record_attempt_failure",
    "record_backpressure_error",
    "record_cancel_actored",
    "record_cancel_requested",
    "record_capacity_refresh_failure",
    "record_claim_latency",
    "record_consumed_message",
    "record_corrupt_dispatch_row",
    "record_cron_budget_deferral",
    "record_cron_failure",
    "record_cron_lock_contention",
    "record_cron_skipped_slots",
    "record_deadline_exceeded_swept",
    "record_dispatch_duration",
    "record_dispatch_failure",
    "record_election_attempt",
    "record_enqueue_dedup",
    "record_error_reporter_failure",
    "record_exception_safe",
    "record_exception_text",
    "record_expired_archive_jobs",
    "record_heartbeat_miss",
    "record_job_abandoned",
    "record_job_interrupted",
    "record_job_interrupted_noop",
    "record_job_timeout",
    "record_leader_lease_expires_in_seconds",
    "record_lock_contention",
    "record_lock_expires_in_seconds",
    "record_loop_stall_attribution",
    "record_pool_acquire_duration",
    "record_process_duration",
    "record_progress_flush_failure",
    "record_progress_publish_failure",
    "record_pruned_jobs",
    "record_published_message",
    "record_queue_wait",
    "record_ratelimit_acquire_dependency_failure",
    "record_ratelimit_denial",
    "record_ratelimit_refund_failure",
    "record_reclaimed_jobs",
    "record_reservation_denial",
    "record_reservation_reclaim_drain_duration",
    "record_reservation_reclaim_drain_failure",
    "record_reservation_reclaim_drain_rows",
    "record_reservation_reclaim_heal_failure",
    "record_slot_pool_acquire_failure",
    "record_sse_connection_closed",
    "record_sse_connection_opened",
    "record_sse_rejection",
    "record_sub_enqueue_failure",
    "record_sweep_batch_size",
    "record_sweep_batch_size_configured",
    "record_sweep_success",
    "record_sweep_timeout",
    "record_sweep_unexpected_error",
    "redact_payload",
    "render_exception",
    "reset_claim_health_state",
    "safe_exception_message",
    "safe_repr",
    "safe_start_span",
    "safe_str",
    "set_exception_message_max_chars",
    "set_exception_redaction_enabled",
    "set_otel_enabled",
    "set_slot_pool_occupancy_source",
    "set_worker_capacity_source",
    "setup_logging",
    "update_actor_backlog_cache",
    "update_actor_oldest_pending_age_cache",
    "update_actor_oldest_running_age_cache",
    "update_cancel_pending_cache",
    "update_cron_slots_behind",
    "update_disabled_schedules_count",
    "update_heartbeat_consecutive_failures",
    "update_jobs_by_status_cache",
    "update_jobs_retry_headroom_cache",
    "update_jobs_retrying_cache",
    "update_jobs_running_cache",
    "update_keyed_reclaim_pending",
    "update_oldest_due_age_cache",
    "update_queue_depth_by_status_cache",
    "update_queue_depth_cache",
    "update_queue_live_workers_cache",
    "update_queue_utilization_cache",
    "update_ratelimit_bucket_tokens_cache",
    "update_reservation_slots_cache",
    "update_running_lease_expired_cache",
    "update_scheduled_count_cache",
    "update_scheduled_horizon_cache",
    "update_stranded_jobs_cache",
    "update_sweep_batch_size_cache",
]

ExporterWiring

ExporterWiring = Literal[
    "configured",
    "preconfigured",
    "none",
    "disabled",
    "sdk_disabled",
    "sdk_missing",
]

What :func:configure_exporters did about the process-global providers.

• configured: SDK providers were installed from the environment (and metrics_port); otel-exporter-configured names the exporters.
• preconfigured: a real provider was already set, an embedding application, a vendor distro or opentelemetry-instrument owns the process, so nothing was touched.
• none: nothing asked for an exporter; the API's no-op proxies stay.
• disabled: TASKQ_OTEL_AUTOCONFIGURE=false.
• sdk_disabled: OTEL_SDK_DISABLED=true.
• sdk_missing: an exporter was requested but the package that provides it is not installed; otel-exporter-unavailable names the extra.

ConsumedOutcome

ConsumedOutcome = Literal[
    "succeeded", "failed", "cancelled", "scheduled"
]

The outcome label set of messaging.client.consumed.messages.

Every value has a producer: the three terminal outcomes, and scheduled for an attempt that ended with the row released back to the queue, a retryable failure, a Snooze / RetryAfter, or an admission denial. taskq.jobs.attempt_failures separates the failure share of scheduled; taskq.jobs.abandoned counts real abandonment, which is an operator cancel outlasting its graces and never a consumer outcome.

StrandedReason

StrandedReason = Literal[
    "no_actor_config", "unserved_queue"
]

Why a pending/scheduled row can never dispatch, the closed reason label set of taskq.jobs.stranded.

TimeoutKind

TimeoutKind = Literal["start_to_close", "schedule_to_close"]

Which budget a job exceeded, the closed kind label set of taskq.jobs.timeouts.

ErrorReporterType

ErrorReporterType = ErrorReporter

Structural type alias for DI registration and parameter annotations.

Use :class:ErrorReporter directly in most contexts; this alias is provided for register_value(ErrorReporterType, ...) calls where a distinct type object is needed for the registry key.

OtelExporterConfigurationError

Bases: RuntimeError

The operator asked for an exporter the SDK could not build.

Raised at worker startup, the earliest stage that can catch a misspelled exporter name or a protocol the installed exporters do not speak, so the worker refuses to start rather than run with a telemetry pipeline that looks configured and exports nothing.

ExceptionText dataclass

ExceptionText(
    type_name: str,
    raw_stacktrace: str,
    message: ScrubbedText,
    stacktrace: ScrubbedText,
)

One rendering of an exception, shared by every sink that reports it.

A failed attempt is reported on the attempt.N span, on the job_exception and job-failed log lines and in the durable ErrorInfo. Rendering the traceback and scrubbing it are the dominant CPU cost of a failed job (a 27-frame traceback is ~0.8 ms to render and ~0.4-0.8 ms to scrub, GIL-held), so :func:render_exception does each once and the sinks are handed this value rather than the exception.

raw_stacktrace is unscrubbed: the durable row lives inside the trust boundary and keeps the DETAIL row values the operator needs; only the telemetry-bound text is scrubbed.

type_name instance-attribute

type_name: str

__qualname__ of the exception class, as OTel's exception.type.

raw_stacktrace instance-attribute

raw_stacktrace: str

traceback.format_exception output, verbatim.

message instance-attribute

message: ScrubbedText

str(exc) scrubbed and length-bounded -- see :func:safe_exception_message.

stacktrace instance-attribute

stacktrace: ScrubbedText

raw_stacktrace scrubbed line-wise, never length-bounded, so it stays diagnostic.

ScrubbedText

Bases: str

Exception text that has been through :func:_scrub_text.

A str in every other respect, so every renderer and serializer treats it as plain text. The type is the invariant: the log processor (_scrub_exception_fields) passes a ScrubbedText field through untouched, which is what lets a handler scrub a traceback once and emit it on several log lines. Only this module constructs one from freshly rendered text; a caller that needs one holds a value that already is one.

__slots__ class-attribute instance-attribute

__slots__ = ()

nul_escaped

nul_escaped() -> ScrubbedText

The same text with NUL codepoints rewritten to the visible \x00.

Scrubbing must run BEFORE the escape, never after: the credential mask's keyword boundary is a lookbehind for a non-word character, and the escape's trailing 0 satisfies it in the wrong direction, so \x00password=… escaped first ships the password. The escape itself introduces only backslash, x and 0 -- it cannot reassemble a DETAIL line or a credential the scrub removed, so the result keeps its scrubbed standing.

Source code in src/taskq/obs/_redact_exc.py

def nul_escaped(self) -> "ScrubbedText":
    """The same text with NUL codepoints rewritten to the visible ``\\x00``.

    Scrubbing must run BEFORE the escape, never after: the credential
    mask's keyword boundary is a lookbehind for a non-word character, and
    the escape's trailing ``0`` satisfies it in the wrong direction, so
    ``\\x00password=…`` escaped first ships the password. The escape
    itself introduces only backslash, ``x`` and ``0`` -- it cannot
    reassemble a DETAIL line or a credential the scrub removed, so the
    result keeps its scrubbed standing.
    """
    return ScrubbedText(sanitize_nul_str(self))

ErrorReporter

Bases: Protocol

Vendor-neutral hook for routing terminal job failures to external systems.

Implementations capture the error and job row, then forward to a vendor-specific backend (Sentry, Datadog, a DLQ, etc.). The library calls :meth:report when a job reaches a terminal failure state , either because retries were exhausted or because the error was non-retryable.

The call is wrapped in a try/except by :func:invoke_error_reporter; a failing reporter never crashes the worker. Reporter failures are counted on the taskq.error_reporter.failures counter with a reporter_type attribute.

Register an :class:ErrorReporter instance as a DI provider (registry.register_value(ErrorReporter, Scope.PROCESS, my_reporter)) or pass it directly to the worker bootstrap.

report async

report(job: JobRow, exception: BaseException) -> None

Source code in src/taskq/obs/error_reporter.py

async def report(self, job: JobRow, exception: BaseException) -> None: ...

NullErrorReporter

Default no-op :class:ErrorReporter, silently drops all reports.

Used when no vendor-specific error routing is configured. Instances are stateless and safe to share.

report async

report(job: JobRow, exception: BaseException) -> None

Source code in src/taskq/obs/error_reporter.py

async def report(self, job: JobRow, exception: BaseException) -> None:
    return None

claim_health_snapshot

claim_health_snapshot(
    *, now: float | None = None
) -> dict[str, ClaimQueueHealth]

Per-queue claim health over the recent window, at now.

Pure read: trims nothing the next record wouldn't, yields NOTHING for a queue with no window entries (the empty-not-zero discipline - a scrape of an idle worker produces no data points, never zeros).

Source code in src/taskq/obs/_claim_health.py

def claim_health_snapshot(*, now: float | None = None) -> dict[str, ClaimQueueHealth]:
    """Per-queue claim health over the recent window, at *now*.

    Pure read: trims nothing the next record wouldn't, yields NOTHING for
    a queue with no window entries (the empty-not-zero discipline - a
    scrape of an idle worker produces no data points, never zeros).
    """
    ts = now if now is not None else _clock()
    cutoff = ts - CLAIM_WINDOW_SECONDS
    by_queue: dict[str, list[float]] = {}
    oldest: dict[str, float] = {}
    newest: dict[str, float] = {}
    for entry_ts, q, seconds in _window:
        if entry_ts < cutoff:
            continue
        by_queue.setdefault(q, []).append(seconds)
        if q not in oldest or entry_ts < oldest[q]:
            oldest[q] = entry_ts
        if q not in newest or entry_ts > newest[q]:
            newest[q] = entry_ts

    # Freeze any baseline buckets the read's clock has completed (a
    # quiet worker's last bucket rolls on the read, not the next write).
    _roll_baseline_buckets(int(ts // 60.0))

    out: dict[str, ClaimQueueHealth] = {}
    for q, lat in by_queue.items():
        p50 = _percentile(lat, 50.0)
        p95 = _percentile(lat, 95.0)
        p99 = _percentile(lat, 99.0)
        span = max(newest[q] - oldest[q], 1.0)
        out[q] = ClaimQueueHealth(
            p50_seconds=p50,
            p95_seconds=p95,
            p99_seconds=p99,
            claims_per_second=len(lat) / span,
            degradation_ratio=_degradation_ratio(q, p99, ts),
        )
    return out

record_claim_latency

record_claim_latency(
    queue: str, seconds: float, *, now: float | None = None
) -> None

Record ONE claim statement's latency (the wiring site is the dispatch helper, success or failure - a failed claim round is still a latency the queue paid).

The queue label is bounded HERE, at record time - the same _bounded_queue discipline every taskq.* metric uses; a payload-shaped string can never become a label.

Source code in src/taskq/obs/_claim_health.py

def record_claim_latency(queue: str, seconds: float, *, now: float | None = None) -> None:
    """Record ONE claim statement's latency (the wiring site is the
    dispatch helper, success or failure - a failed claim round is still a
    latency the queue paid).

    The queue label is bounded HERE, at record time - the same
    ``_bounded_queue`` discipline every ``taskq.*`` metric uses; a
    payload-shaped string can never become a label.
    """
    if not otel_enabled():
        return
    ts = now if now is not None else _clock()
    q = bounded_queue_label(queue)
    _window.append((ts, q, seconds))
    # The count cap trims oldest-first; the time trim happens at read.
    while len(_window) > CLAIM_WINDOW_CAP_ENTRIES:
        _window.popleft()
    # Baseline bucket fill: minute buckets keyed on the recording clock.
    bucket = int(ts // 60.0)
    lat = _current_bucket.get((bucket, q))
    if lat is None:
        _roll_baseline_buckets(bucket)
        lat = _current_bucket.setdefault((bucket, q), [])
    lat.append(seconds)

reset_claim_health_state

reset_claim_health_state() -> None

Drop every recorded latency (test isolation).

Source code in src/taskq/obs/_claim_health.py

def reset_claim_health_state() -> None:
    """Drop every recorded latency (test isolation)."""
    global _clock
    _window.clear()
    _current_bucket.clear()
    _baseline.clear()
    _clock = time.monotonic

configure_exporters

configure_exporters(
    settings: _ExporterSettings,
) -> ExporterWiring

Install SDK tracer and meter providers from the OTel environment.

Called once by the taskq worker CLI after settings load and before the worker records anything: measurements a proxy instrument takes before a provider exists are dropped, not replayed. Embedding applications that host a worker without the CLI call this themselves at process start, or configure the SDK directly, a provider that is already set is never replaced.

Raises :class:OtelExporterConfigurationError when the environment names an exporter the SDK cannot build; every other outcome is a return value plus one log line.

Source code in src/taskq/obs/_exporter.py

def configure_exporters(settings: _ExporterSettings) -> ExporterWiring:
    """Install SDK tracer and meter providers from the OTel environment.

    Called once by the ``taskq worker`` CLI after settings load and before
    the worker records anything: measurements a proxy instrument takes
    before a provider exists are dropped, not replayed. Embedding
    applications that host a worker without the CLI call this themselves
    at process start, or configure the SDK directly, a provider that is
    already set is never replaced.

    Raises :class:`OtelExporterConfigurationError` when the environment
    names an exporter the SDK cannot build; every other outcome is a
    return value plus one log line.
    """
    if not settings.otel_autoconfigure:
        _log.info("otel-exporter-autoconfigure-disabled", source="TASKQ_OTEL_AUTOCONFIGURE")
        return "disabled"
    if os.environ.get(_SDK_DISABLED_ENV, "").strip().lower() == "true":
        _log.info("otel-exporter-sdk-disabled", source=_SDK_DISABLED_ENV)
        return "sdk_disabled"

    prometheus_available = _installed(_PROMETHEUS_MODULE)
    if settings.metrics_port is not None and not prometheus_available:
        _warn_unavailable(
            "prometheus",
            source=_PROMETHEUS_EXPORTER,
            detail=(
                "TASKQ_METRICS_PORT is set but opentelemetry-exporter-prometheus is "
                "not installed, so no scrape listener will be bound; install "
                "taskq-py[prometheus]"
            ),
        )
    plan = _plan(settings, prometheus_available=prometheus_available)
    if not plan.requested:
        if settings.metrics_port is not None:
            return "sdk_missing"
        _log.info("otel-exporter-configured", traces="", metrics="", logs="", source="none")
        return "none"

    if _provider_already_set():
        if settings.metrics_port is not None:
            _log.warning(
                "otel-exporter-scrape-not-served",
                port=settings.metrics_port,
                reason=(
                    "a tracer or meter provider is already installed, so this "
                    "worker wires no exporters and the requested scrape port "
                    "stays unbound"
                ),
            )
        _log.info(
            "otel-exporter-preconfigured",
            detail=(
                "a tracer or meter provider is already installed (an embedding "
                "application, a vendor distro, or opentelemetry-instrument); "
                "the worker leaves it untouched"
            ),
        )
        return "preconfigured"

    try:
        from opentelemetry.sdk._configuration import (
            _get_exporter_entry_point,  # pyright: ignore[reportPrivateUsage]  # Why: the SDK ships its env-var configurator as a private module; it is the exact code path opentelemetry-instrument runs, and re-implementing it would drift from the SDK's own parsing of the same variables.
            _initialize_components,  # pyright: ignore[reportPrivateUsage]  # Why: see above -- same module, same reasoning.
        )
    except ImportError:
        _warn_unavailable(
            "otel",
            source=",".join(plan.sources),
            detail=(
                "OTEL_* exporter variables are set but opentelemetry-sdk is not "
                "installed, so nothing will be exported; install taskq-py[otel]"
            ),
        )
        return "sdk_missing"

    def _make_prometheus_env_restore(saved: dict[str, str | None], /) -> Callable[[], None]:
        """Roll the two reader variables back to their pre-call state: the
        reader reads them once at construction, and a value left in
        ``os.environ`` would leak to child processes and override an
        operator's own ``OTEL_EXPORTER_PROMETHEUS_*`` choice there."""

        def _restore() -> None:
            for key, prior in saved.items():
                if prior is None:
                    del os.environ[key]
                else:
                    os.environ[key] = prior

        return _restore

    _restore_prometheus_env: Callable[[], None] | None = None
    if plan.prometheus_port is not None:
        # The SDK's Prometheus reader binds its listener from these two
        # variables and nothing else. TASKQ_METRICS_PORT is the explicit
        # opt-in, so it is authoritative for the port; the host follows
        # TASKQ_METRICS_HOST, then the worker's other TCP listener, unless
        # the operator addressed the reader directly. The write is scoped
        # to the _initialize_components call below and rolled back after
        # it (see _make_prometheus_env_restore).
        _saved_prometheus_env = {
            key: os.environ.get(key) for key in (_PROMETHEUS_PORT_ENV, _PROMETHEUS_HOST_ENV)
        }
        os.environ[_PROMETHEUS_PORT_ENV] = str(plan.prometheus_port)
        if _PROMETHEUS_HOST_ENV not in os.environ:
            os.environ[_PROMETHEUS_HOST_ENV] = settings.metrics_host or settings.health_host
        _restore_prometheus_env = _make_prometheus_env_restore(_saved_prometheus_env)

    def _resolve(names: list[str], signal: Literal["traces", "metrics", "logs"]) -> list[str]:
        # Why the SDK's own resolver and not a local map: the bare "otlp"
        # entry point is the gRPC exporter in every signal, and the protocol
        # variables (OTEL_EXPORTER_OTLP_PROTOCOL and its per-signal forms)
        # select the HTTP one only when the name is resolved through this
        # mapping. Names handed to _initialize_components directly SKIP it
        # (it applies only to its own environment parse), so without this
        # step a worker configured with http/protobuf silently exported gRPC.
        return [_get_exporter_entry_point(name, signal) for name in names]

    _bound_host: str | None = None
    try:
        _initialize_components(
            trace_exporter_names=_resolve(plan.extras("traces"), "traces"),
            metric_exporter_names=_resolve(plan.extras("metrics"), "metrics"),
            log_exporter_names=_resolve(plan.extras("logs"), "logs"),
        )
        _bound_host = (
            os.environ.get(_PROMETHEUS_HOST_ENV) if plan.prometheus_port is not None else None
        )
    except Exception as exc:
        raise OtelExporterConfigurationError(
            "OpenTelemetry exporter configuration failed: "
            f"{type(exc).__name__}: {exc}. Check OTEL_TRACES_EXPORTER, "
            "OTEL_METRICS_EXPORTER, OTEL_LOGS_EXPORTER and OTEL_EXPORTER_OTLP_PROTOCOL "
            "against the exporters installed by taskq-py[otel] (otlp, console) and "
            "taskq-py[prometheus] (prometheus), or set TASKQ_OTEL_AUTOCONFIGURE=false "
            "to configure the SDK yourself."
        ) from exc
    finally:
        if _restore_prometheus_env is not None:
            _restore_prometheus_env()

    _log.info(
        "otel-exporter-configured",
        traces=",".join(plan.traces),
        metrics=",".join(plan.metrics),
        logs=",".join(plan.logs),
        source=",".join(plan.sources),
        prometheus_port=plan.prometheus_port,
        prometheus_host=_bound_host,
    )
    return "configured"

get_meter

get_meter() -> Meter

Return the library's meter. Honors any globally-configured provider.

The meter object is resolved on first use and memoized, exactly like :func:get_tracer: with no SDK set up, metrics.get_meter returns a _ProxyMeter, which rebinds to the real provider's meter when an SDK registers later (on_set_meter_provider notifies every proxy meter, and every proxy instrument on it), so memoization never pins the proxy/no-op behavior. And unlike the uncached call, it never asks the proxy provider for a second meter: _ProxyMeterProvider appends every get_meter result to a list with no cleanup path, so an unmemoized accessor grows that list on every lazy-instrument call , one entry per rate-limit denial, reservation denial, flush failure, and drain row-count, in exactly the default deployment (taskq never installs a provider itself).

Source code in src/taskq/obs/_otel.py

def get_meter() -> Meter:
    """Return the library's meter. Honors any globally-configured provider.

    The meter object is resolved on first use and memoized, exactly like
    :func:`get_tracer`: with no SDK set up, ``metrics.get_meter`` returns
    a ``_ProxyMeter``, which rebinds to the real provider's meter when an
    SDK registers later (``on_set_meter_provider`` notifies every proxy
    meter, and every proxy instrument on it), so memoization never pins
    the proxy/no-op behavior. And unlike the uncached call, it never asks
    the proxy provider for a second meter: ``_ProxyMeterProvider`` appends
    every ``get_meter`` result to a list with no cleanup path, so an
    unmemoized accessor grows that list on every lazy-instrument call ,
    one entry per rate-limit denial, reservation denial, flush failure,
    and drain row-count, in exactly the default deployment (taskq never
    installs a provider itself).
    """
    global _library_meter
    if _library_meter is None:
        _library_meter = metrics.get_meter(INSTRUMENTATION_NAME, _version())
    return _library_meter

get_tracer

get_tracer() -> Tracer

Return the library's tracer. Honors any globally-configured provider.

The tracer object is resolved on first use and memoized. That is safe across the no-provider → real-provider transition: with no SDK set up, trace.get_tracer returns a ProxyTracer, which re-checks the global provider on every span start and rebinds to the real one when an SDK registers later -- so memoization never pins the proxy/no-op behavior.

Source code in src/taskq/obs/_otel.py

def get_tracer() -> Tracer:
    """Return the library's tracer. Honors any globally-configured provider.

    The tracer object is resolved on first use and memoized.  That is safe
    across the no-provider → real-provider transition: with no SDK set up,
    ``trace.get_tracer`` returns a ``ProxyTracer``, which re-checks the
    global provider on every span start and rebinds to the real one when an
    SDK registers later -- so memoization never pins the proxy/no-op
    behavior.
    """
    global _library_tracer
    if _library_tracer is None:
        _library_tracer = trace.get_tracer(INSTRUMENTATION_NAME, _version())
    return _library_tracer

otel_enabled

otel_enabled() -> bool

Whether telemetry emission is currently enabled.

Read at call time, never imported by value: the flag is module state flipped once by worker startup (see :func:set_otel_enabled), and a reader checks it to skip WORK whose only consumer is a metric, a query whose result only a reconcile would export is a wasted round trip when the flag is off. The emitters themselves stay gated individually regardless.

Source code in src/taskq/obs/_otel.py

def otel_enabled() -> bool:
    """Whether telemetry emission is currently enabled.

    Read at call time, never imported by value: the flag is module state
    flipped once by worker startup (see :func:`set_otel_enabled`), and a
    reader checks it to skip WORK whose only consumer is a metric, a
    query whose result only a reconcile would export is a wasted round
    trip when the flag is off.  The emitters themselves stay gated
    individually regardless.
    """
    return _otel_enabled

reconcile_cron_failures

reconcile_cron_failures(totals: Mapping[str, int]) -> None

Move each actor's failure series onto totals, the database's own summed cron_schedules.consecutive_failures per actor.

The series answers "is any schedule failing right now", and only the database knows. Schedules are enabled, disabled and deleted by clients, by the CLI and by the admin UI, every one of them outside the process that emits this metric, and each of those actions clears or removes a count some worker counted up. A worker cannot emit another process's delta, so a balance built from this process's deltas alone can only drift upward: an actor whose last failing schedule was deleted a month ago would report a failure level forever, and an alert on the series becomes unreadable exactly when an operator needs it. Reconciling against the database each tick makes every out-of-process change self-correct on the next tick with due work.

totals must be the complete per-actor truth, every actor with a failing schedule anywhere in the table (see _actor_failure_totals), not the slice one tick's batch happened to touch. A batch covers only DUE schedules, so an actor whose failing schedule was deleted or disabled with nothing left due never appears in a batch again; under a batch-scoped totals its level would strand at its last value. An actor ABSENT from totals therefore reads as a true zero, the database holds no failing schedule for it, and its series is returned to zero here.

Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def reconcile_cron_failures(totals: Mapping[str, int]) -> None:
    """Move each actor's failure series onto *totals*, the database's own
    summed ``cron_schedules.consecutive_failures`` per actor.

    The series answers "is any schedule failing right now", and only the
    database knows.  Schedules are enabled, disabled and deleted by
    clients, by the CLI and by the admin UI, every one of them outside
    the process that emits this metric, and each of those actions clears
    or removes a count some worker counted up.  A worker cannot emit
    another process's delta, so a balance built from this process's
    deltas alone can only drift upward: an actor whose last failing
    schedule was deleted a month ago would report a failure level
    forever, and an alert on the series becomes unreadable exactly when
    an operator needs it.  Reconciling against the database each tick
    makes every out-of-process change self-correct on the next tick with
    due work.

    *totals* must be the complete per-actor truth, every actor with a
    failing schedule anywhere in the table (see ``_actor_failure_totals``),
    not the slice one tick's batch happened to touch.  A batch covers only
    DUE schedules, so an actor whose failing schedule was deleted or
    disabled with nothing left due never appears in a batch again; under a
    batch-scoped *totals* its level would strand at its last value.  An
    actor ABSENT from *totals* therefore reads as a true zero, the
    database holds no failing schedule for it, and its series is returned
    to zero here.

    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    wanted: dict[str, int] = {}
    for actor, total in totals.items():
        label = _bounded_cron_actor(actor)
        # Distinct actors can share the overflow label, so their totals
        # sum onto the one series they share.
        wanted[label] = wanted.get(label, 0) + total
    level = _cron_failure_level()
    for label, target in wanted.items():
        delta = target - level.get(label, 0)
        if delta:
            _cron_consecutive_failures.add(delta, {"actor": label})
        level[label] = target
    # *totals* is the complete truth, so a label the database no longer
    # reports is a schedule set that stopped failing somewhere else ,
    # return it to zero rather than stranding its last level.
    for label in list(level):
        if label not in wanted:
            stranded = level.pop(label)
            if stranded:
                _cron_consecutive_failures.add(-stranded, {"actor": label})

record_archived_jobs

record_archived_jobs(status: str, count: int = 1) -> None

Bump the archived.jobs counter.

Called alongside record_pruned_jobs at the prune sweep call site in leader.py. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_archived_jobs(status: str, count: int = 1) -> None:
    """Bump the archived.jobs counter.

    Called alongside record_pruned_jobs at the prune sweep call site in leader.py.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _archived_jobs.add(count, {"status": status})

record_attempt_failure

record_attempt_failure(
    actor: str,
    error_type: str | None = None,
    *,
    retryable: bool,
) -> None

Count one attempt that ended in an actor failure, retried or terminal.

Called from the failure handlers (worker/_handlers.py) once per handled exception, after the retry decision and before the terminal write: an attempt that raised failed whether or not the row write that follows lands, and the consumed-messages outcome says what happened to the row. retryable is the classifier's decision, true when the attempt is rescheduled for another try, false when the failure is terminal (a non-retryable class, or the attempt budget exhausted) , which is what a retry-rate alert reads. error_type is the exception class name; omitted, it derives from the exception being handled (_resolve_error_type): a closed set, never caller text. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_attempt_failure(actor: str, error_type: str | None = None, *, retryable: bool) -> None:
    """Count one attempt that ended in an actor failure, retried or terminal.

    Called from the failure handlers (``worker/_handlers.py``) once per
    handled exception, after the retry decision and before the terminal
    write: an attempt that raised failed whether or not the row write that
    follows lands, and the consumed-messages ``outcome`` says what happened
    to the row. ``retryable`` is the classifier's decision, ``true`` when
    the attempt is rescheduled for another try, ``false`` when the failure
    is terminal (a non-retryable class, or the attempt budget exhausted) ,
    which is what a retry-rate alert reads. ``error_type`` is the exception
    class name; omitted, it derives from the exception being handled
    (``_resolve_error_type``): a closed set, never caller text.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.jobs.attempt_failures",
        description=(
            "Attempts that ended in an actor failure. Attributes: actor, "
            "error_type (exception class name, a closed set; see "
            "_resolve_error_type), retryable ('true' when the attempt is "
            "rescheduled for another try, 'false' when the failure is "
            "terminal). The failure share of consumed outcome='scheduled'."
        ),
    ).add(
        1,
        {
            "actor": actor,
            "error_type": _resolve_error_type(error_type),
            "retryable": "true" if retryable else "false",
        },
    )

record_backpressure_error

record_backpressure_error(
    actor: str, *, kind: str = "max_pending"
) -> None

Bump the backpressure.errors counter.

Unconditional (not gated by _otel_enabled): backpressure errors are safety-critical signals that must be counted even when OTel is disabled, so operators always have visibility into enqueue rejections. kind is the bounded enum named on the counter: the two capacity kinds, and the two identity-serialization lock-timeout kinds that count their refusals beside the typed errors' warning logs.

Source code in src/taskq/obs/_otel.py

def record_backpressure_error(actor: str, *, kind: str = "max_pending") -> None:
    """Bump the backpressure.errors counter.

    Unconditional (not gated by ``_otel_enabled``): backpressure errors are
    safety-critical signals that must be counted even when OTel is disabled,
    so operators always have visibility into enqueue rejections. ``kind``
    is the bounded enum named on the counter: the two capacity kinds, and
    the two identity-serialization lock-timeout kinds that count their
    refusals beside the typed errors' warning logs.
    """
    try:
        _backpressure_errors.add(1, {"actor": actor, "kind": kind})
    except Exception:
        _log.warning("otel-metric-record-failed", instrument_name="taskq.backpressure.errors")

record_cancel_actored

record_cancel_actored(actor: str) -> None

Count one operator cancel the worker began acting on, labeled by actor.

Called from the cancel controller's two FIRST-observation arms only (worker/cancel.py): the phase-1 cooperative observation, and the PG-observation fast-advance when the local phase was still NONE. A re-issue arm or a later tick walking an already-observed row records nothing, so one cancel request is one increment no matter how many heartbeat ticks it spans.

Unconditional (not gated by _otel_enabled), like the other cancel-plane counters: a cancel the operator asked for and the fleet cannot see acting on is the exact defect this counter exists to expose, and it must survive an exporter misconfiguration.

Source code in src/taskq/obs/_otel.py

def record_cancel_actored(actor: str) -> None:
    """Count one operator cancel the worker began acting on, labeled by actor.

    Called from the cancel controller's two FIRST-observation arms only
    (``worker/cancel.py``): the phase-1 cooperative observation, and the
    PG-observation fast-advance when the local phase was still NONE. A
    re-issue arm or a later tick walking an already-observed row records
    nothing, so one cancel request is one increment no matter how many
    heartbeat ticks it spans.

    Unconditional (not gated by ``_otel_enabled``), like the other
    cancel-plane counters: a cancel the operator asked for and the fleet
    cannot see acting on is the exact defect this counter exists to
    expose, and it must survive an exporter misconfiguration.
    """
    try:
        _cancels_actored.add(1, {"actor": actor})
    except Exception:
        _log.warning("otel-metric-record-failed", instrument_name="taskq.jobs.cancels_actored")

record_cancel_requested

record_cancel_requested() -> None

Bump the cancel-requested counter. This counter is unconditional: incremented once per JobsClient.cancel() call regardless of cancellation_initiated outcome.

Source code in src/taskq/obs/_otel.py

def record_cancel_requested() -> None:
    """Bump the cancel-requested counter.
    This counter is unconditional:
    incremented once per ``JobsClient.cancel()`` call regardless of
    ``cancellation_initiated`` outcome.
    """
    try:
        _cancellation_requested.add(1)
    except Exception:
        _log.warning("otel-metric-record-failed", instrument_name="taskq.cancellation.requested")

record_capacity_refresh_failure

record_capacity_refresh_failure(
    *, has_snapshot: bool, error_type: str | None = None
) -> None

Count a failed capacity-cache refresh.

The cache fails OPEN by design -- it keeps the last snapshot, or falls back to the @actor literal, and stamps refreshed_at so a sick backend is not re-queried on every enqueue. That reasoning is sound; the problem was that a load-shedding gate which relaxes precisely when the backend is degraded had no signal an operator could alert on, only a warning log.

has_snapshot=False is the materially worse case: the first refresh at process start failed, so there is no stored data at all and every enqueue enforces the code literal rather than the operator's tightened max_pending -- for a full TTL at a time, indefinitely while the backend stays sick.

error_type is the exception class name; omitted, it derives from the exception being handled (_resolve_error_type), the call site records from the refresh read's except block.

Unconditional (not gated by _otel_enabled) for the same reason as record_backpressure_error: this is a safety-critical signal.

Source code in src/taskq/obs/_otel.py

def record_capacity_refresh_failure(*, has_snapshot: bool, error_type: str | None = None) -> None:
    """Count a failed capacity-cache refresh.

    The cache fails OPEN by design -- it keeps the last snapshot, or falls back
    to the ``@actor`` literal, and stamps ``refreshed_at`` so a sick backend is
    not re-queried on every enqueue. That reasoning is sound; the problem was
    that a load-shedding gate which relaxes precisely when the backend is
    degraded had no signal an operator could alert on, only a warning log.

    ``has_snapshot=False`` is the materially worse case: the first refresh at
    process start failed, so there is no stored data at all and every enqueue
    enforces the code literal rather than the operator's tightened
    ``max_pending`` -- for a full TTL at a time, indefinitely while the backend
    stays sick.

    ``error_type`` is the exception class name; omitted, it derives from the
    exception being handled (``_resolve_error_type``), the call site records
    from the refresh read's ``except`` block.

    Unconditional (not gated by ``_otel_enabled``) for the same reason as
    ``record_backpressure_error``: this is a safety-critical signal.
    """
    try:
        _capacity_refresh_failures.add(
            1,
            {
                "degraded": "stale_snapshot" if has_snapshot else "no_snapshot",
                "error_type": _resolve_error_type(error_type),
            },
        )
    except Exception:
        _log.warning(
            "otel-metric-record-failed",
            instrument_name="taskq.backpressure.capacity_refresh_failures",
        )

record_consumed_message

record_consumed_message(
    actor: str, queue: str, *, outcome: ConsumedOutcome
) -> None

Bump the consumed-messages counter.

Called after job completion, outside the CONSUMER span body, to ensure sampling independence. Respects _otel_enabled, no-op when False.

outcome is the closed :data:ConsumedOutcome set. A consumer-path AttemptOutcome of "scheduled" (retry, snooze, admission denial) is recorded as exactly that, the row went back to the queue, never as abandoned, which is the operator-cancel outcome and has its own counter (:func:record_job_abandoned). A "noop" attempt consumed nothing and must not reach this recorder.

Source code in src/taskq/obs/_otel.py

def record_consumed_message(actor: str, queue: str, *, outcome: ConsumedOutcome) -> None:
    """Bump the consumed-messages counter.

    Called after job completion, outside the CONSUMER span body,
    to ensure sampling independence.
    Respects ``_otel_enabled``, no-op when False.

    ``outcome`` is the closed :data:`ConsumedOutcome` set. A consumer-path
    ``AttemptOutcome`` of ``"scheduled"`` (retry, snooze, admission denial)
    is recorded as exactly that, the row went back to the queue, never
    as ``abandoned``, which is the operator-cancel outcome and has its own
    counter (:func:`record_job_abandoned`). A ``"noop"`` attempt consumed
    nothing and must not reach this recorder.
    """
    if not _otel_enabled:
        return
    _consumed_messages.add(1, {"actor": actor, "queue": _bounded_queue(queue), "outcome": outcome})

record_corrupt_dispatch_row

record_corrupt_dispatch_row(
    queue: str, column: str
) -> None

Bump the corrupt-claimed-row counter.

Called from the dispatch round's per-row decode path. column is the jsonb column that refused to decode (a closed set). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_corrupt_dispatch_row(queue: str, column: str) -> None:
    """Bump the corrupt-claimed-row counter.

    Called from the dispatch round's per-row decode path. ``column`` is
    the jsonb column that refused to decode (a closed set). Respects
    ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _corrupt_dispatch_rows.add(1, {"queue": _bounded_queue(queue), "column": column})

record_cron_budget_deferral

record_cron_budget_deferral(actor: str) -> None

Count one cron fire deferred by the tick's factory budget.

The alertable face of the budget-deferral path: the log event (cron-fire-budget-deferred) is unconditional and carries the per-schedule attribution; this counter is the series an operator alerts on when deferrals stop being transient. See the counter's description for the sustained-rate diagnosis and the operator knobs. Labeled by actor, admitted through the same cap as :func:record_cron_failure (schedule rows accept any string at creation time, so the label is bounded). Respects _otel_enabled: no-op when False.

Source code in src/taskq/obs/_otel.py

def record_cron_budget_deferral(actor: str) -> None:
    """Count one cron fire deferred by the tick's factory budget.

    The alertable face of the budget-deferral path: the log event
    (``cron-fire-budget-deferred``) is unconditional and carries the
    per-schedule attribution; this counter is the series an operator
    alerts on when deferrals stop being transient.  See the counter's
    description for the sustained-rate diagnosis and the operator knobs.
    Labeled by ``actor``, admitted through the same cap as
    :func:`record_cron_failure` (schedule rows accept any string at
    creation time, so the label is bounded).
    Respects ``_otel_enabled``: no-op when False.
    """
    if not _otel_enabled:
        return
    label = _bounded_cron_actor(actor)
    _cron_budget_deferrals.add(1, {"actor": label})

record_cron_failure

record_cron_failure(actor: str, delta: int) -> None

Record a cron failure delta on the UpDownCounter.

On failure, callers add +1 per failure. On success, callers add -current_count for that schedule to reset the counter to zero , a simple add(-1) would leave a non-zero cumulative value if there were multiple consecutive failures.

Labeled by actor, admitted through the same cap as queue: the failure path emits the raw cron_schedules.actor string and schedule rows accept any string at creation time, so the first _MAX_ACTOR_LABEL_VALUES distinct names keep their series and later names collapse onto _other_ (see the cardinality note above _bounded_cron_actor). Schedules on one actor share a series, and the per-schedule attribution the caller already holds rides on the cron fired / cron fire failed log lines and the cron-fire span instead, schedule_id is a per-row, runtime-minted UUID and identity-like (see the cardinality note above _lock_expires_in_seconds). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_cron_failure(actor: str, delta: int) -> None:
    """Record a cron failure delta on the UpDownCounter.

    On failure, callers add ``+1`` per failure. On success, callers add
    ``-current_count`` for that schedule to reset the counter to zero ,
    a simple ``add(-1)`` would leave a non-zero cumulative value if
    there were multiple consecutive failures.

    Labeled by ``actor``, admitted through the same cap as ``queue``:
    the failure path emits the raw ``cron_schedules.actor`` string and
    schedule rows accept any string at creation time, so the first
    ``_MAX_ACTOR_LABEL_VALUES`` distinct names keep their series and
    later names collapse onto ``_other_`` (see the cardinality note
    above ``_bounded_cron_actor``).  Schedules on one actor share a
    series, and the per-schedule attribution the caller already holds
    rides on the ``cron fired`` / ``cron fire failed`` log lines and the
    cron-fire span instead, ``schedule_id`` is a per-row,
    runtime-minted UUID and identity-like (see the cardinality note
    above ``_lock_expires_in_seconds``).
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    label = _bounded_cron_actor(actor)
    _cron_consecutive_failures.add(delta, {"actor": label})
    level = _cron_failure_level()
    level[label] = level.get(label, 0) + delta

record_cron_lock_contention

record_cron_lock_contention(worker_id: str) -> None

Count a cron tick skipped because the advisory lock was held.

A steady low rate is the benign leader-handover overlap. A rate equal to the tick rate, sustained, means cron is not running anywhere: the lock is transaction-scoped and releases on COMMIT/ROLLBACK, which never happens if the holding session was partitioned without a FIN. Before this counter the two were indistinguishable, because the contended branch returned in silence. Respects _otel_enabled -- no-op when False.

Source code in src/taskq/obs/_otel.py

def record_cron_lock_contention(worker_id: str) -> None:
    """Count a cron tick skipped because the advisory lock was held.

    A steady low rate is the benign leader-handover overlap. A rate equal to
    the tick rate, sustained, means cron is not running anywhere: the lock is
    transaction-scoped and releases on COMMIT/ROLLBACK, which never happens if
    the holding session was partitioned without a FIN. Before this counter the
    two were indistinguishable, because the contended branch returned in
    silence.
    Respects ``_otel_enabled`` -- no-op when False.
    """
    if not _otel_enabled:
        return
    del worker_id  # Why: not a dimension -- see the cardinality note above.
    _cron_lock_contention.add(1)

record_cron_skipped_slots

record_cron_skipped_slots(actor: str, count: int) -> None

Count count cron slots a fire attempt dropped, labeled by actor.

The catch-up window's skip branch (cron_loop._plan_fire) is the system ADMITTING it is losing the race: a schedule's next_fire_at fell behind server_now - cron_catch_up_window, and every schedule occurrence between the owed slot and the recomputed one is a fire the system will not even attempt. Until this counter existed that admission was log-only (the cron missed slots skipped warning), so "jobs clearing slower than the cron period" -- the runaway fan-out -- was invisible on the metrics plane.

count is in FIRE UNITS, not seconds: a */5 schedule reported as 3 is three dropped fires (about 15 minutes of schedule time), the unit the runaway predicate compares against the cron period. The count is emitted from the tick's commit-gated emission, so a rolled back tick (whose owed slots are re-attempted, not dropped) reports nothing; it covers both committed successes and committed suppressions (a budget-deferred fire of an overdue schedule durably advances next_fire_at past the owed slots, so its skips count too). The clearance comparison itself -- fires vs. job completion rate, the runaway PREDICATE -- lives in taskq.insights; this counter is one of the producer-side facts it reads.

A floor, not always the depth: the count walk is bounded by a slice of the tick's own deadline (cron_loop._SKIP_WALK_BUDGET_FRACTION) so a fleet-down backlog cannot outlive the tick and livelock the fleet. A walk its budget cut short reports the count it reached and flags it skipped_slots_partial=True on the cron missed slots skipped warning, the cron fired line and the fire span; the counter and the depth gauge then read that lower bound. Every consumer tolerates it -- the runaway predicate compares the depth against the cron period, and a backlog deeper than the budget's reach is deeper than any period.

Labeled by actor, admitted through the same cap as :func:record_cron_failure (schedule rows accept any string at creation time, so the label is bounded). Per-schedule attribution rides the cron missed slots skipped warning, the cron fired log line and the cron-fire span's taskq.cron.skipped_slots attribute, where cardinality is free. The depth companion (how far behind, not just that a drop happened) is the taskq.cron.slots_behind gauge. Respects _otel_enabled: no-op when False.

Source code in src/taskq/obs/_otel.py

def record_cron_skipped_slots(actor: str, count: int) -> None:
    """Count *count* cron slots a fire attempt dropped, labeled by actor.

    The catch-up window's skip branch (``cron_loop._plan_fire``) is the
    system ADMITTING it is losing the race: a schedule's ``next_fire_at``
    fell behind ``server_now - cron_catch_up_window``, and every schedule
    occurrence between the owed slot and the recomputed one is a fire the
    system will not even attempt. Until this counter existed that
    admission was log-only (the ``cron missed slots skipped`` warning),
    so "jobs clearing slower than the cron period" -- the runaway
    fan-out -- was invisible on the metrics plane.

    *count* is in FIRE UNITS, not seconds: a ``*/5`` schedule reported as
    3 is three dropped fires (about 15 minutes of schedule time), the
    unit the runaway predicate compares against the cron period. The
    count is emitted from the tick's commit-gated emission, so a rolled
    back tick (whose owed slots are re-attempted, not dropped) reports
    nothing; it covers both committed successes and committed
    suppressions (a budget-deferred fire of an overdue schedule durably
    advances ``next_fire_at`` past the owed slots, so its skips count
    too). The clearance comparison itself -- fires vs. job completion
    rate, the runaway PREDICATE -- lives in ``taskq.insights``; this
    counter is one of the producer-side facts it reads.

    A floor, not always the depth: the count walk is bounded by a slice
    of the tick's own deadline (``cron_loop._SKIP_WALK_BUDGET_FRACTION``)
    so a fleet-down backlog cannot outlive the tick and livelock the
    fleet. A walk its budget cut short reports the count it reached and
    flags it ``skipped_slots_partial=True`` on the ``cron missed slots
    skipped`` warning, the ``cron fired`` line and the fire span; the
    counter and the depth gauge then read that lower bound. Every
    consumer tolerates it -- the runaway predicate compares the depth
    against the cron period, and a backlog deeper than the budget's
    reach is deeper than any period.

    Labeled by ``actor``, admitted through the same cap as
    :func:`record_cron_failure` (schedule rows accept any string at
    creation time, so the label is bounded). Per-schedule attribution
    rides the ``cron missed slots skipped`` warning, the ``cron fired``
    log line and the cron-fire span's ``taskq.cron.skipped_slots``
    attribute, where cardinality is free. The depth companion (how far
    behind, not just that a drop happened) is the
    ``taskq.cron.slots_behind`` gauge.
    Respects ``_otel_enabled``: no-op when False.
    """
    if not _otel_enabled:
        return
    label = _bounded_cron_actor(actor)
    _lazy_counter(
        "taskq.cron.skipped_slots",
        description=(
            "Cron schedule occurrences a fire attempt DROPPED because the "
            "schedule's next_fire_at fell behind the catch-up window "
            "(server_now - TASKQ_CRON_CATCH_UP_WINDOW): fires the system "
            "could not even attempt, counted in fire units (a */5 schedule "
            "reported as 3 is three dropped fires). One increment per fire "
            "attempt, carrying that attempt's full drop count; a non-zero "
            "rate is the leader losing the race -- the cron period, the "
            "tick cadence or the catch-up window no longer fits the "
            "schedule's real delivery time. Delayed-not-lost does not "
            "apply: skipped slots are never replayed (the recompute lands "
            "next_fire_at in the future), the count IS the record of what "
            "was dropped -- except when the count walk's tick-budget slice "
            "expired mid-backlog, which the cron missed slots skipped "
            "warning flags as skipped_slots_partial: the count is then a "
            "floor (a backlog that deep is past any runaway predicate "
            "anyway). Read beside taskq.cron.slots_behind (the "
            "per-attempt depth) and taskq.cron.budget_deferrals (a "
            "different, budget-shaped lag). The actor label is capped like "
            "taskq.cron.consecutive_failures (first 100 distinct names, "
            "overflow collapses to '_other_'); per-schedule attribution is "
            "on the cron missed slots skipped / cron fired log lines and "
            "the cron-fire span."
        ),
    ).add(count, {"actor": label})

record_deadline_exceeded_swept

record_deadline_exceeded_swept(
    actor: str, count: int = 1
) -> None

Bump the deadline-exceeded sweep counter.

Unconditional (not gated by _otel_enabled): deadline-exceeded sweeps indicate jobs that violated their execution budget, a correctness signal that must be counted even when OTel is disabled, so operators always have visibility into sweep activity.

Source code in src/taskq/obs/_otel.py

def record_deadline_exceeded_swept(actor: str, count: int = 1) -> None:
    """Bump the deadline-exceeded sweep counter.

    Unconditional (not gated by ``_otel_enabled``): deadline-exceeded sweeps
    indicate jobs that violated their execution budget, a correctness signal
    that must be counted even when OTel is disabled, so operators always have
    visibility into sweep activity.
    """
    try:
        _deadline_exceeded_sweep_jobs_failed.add(count, {"actor": actor})
    except Exception:
        _log.warning(
            "otel-metric-record-failed", instrument_name="taskq.deadline_exceeded_sweep.jobs_failed"
        )
    # The sweep's arm of the whole-job deadline, on the timeouts family
    # beside the handler arms (gated, unlike the sweep counter above).
    record_job_timeout(actor, kind="schedule_to_close", count=count)

record_dispatch_duration

record_dispatch_duration(
    queue: str, elapsed: float
) -> None

Record dispatch query latency on the histogram.

Called outside the dispatch span body for sampling independence. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_dispatch_duration(queue: str, elapsed: float) -> None:
    """Record dispatch query latency on the histogram.

    Called outside the ``dispatch`` span body for sampling independence.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _dispatch_duration.record(elapsed, {"queue": _bounded_queue(queue)})

record_dispatch_failure

record_dispatch_failure(
    queue: str, error_type: str | None = None
) -> None

Bump the dispatch-failure counter.

Called from the dispatch round's exception path, outside the span body for sampling independence. error_type is the exception class name; omitted, it derives from the exception being handled (_resolve_error_type), every call site records from an except block, so the derived value is the caught exception's class. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_dispatch_failure(queue: str, error_type: str | None = None) -> None:
    """Bump the dispatch-failure counter.

    Called from the dispatch round's exception path, outside the span body
    for sampling independence.  ``error_type`` is the exception class name;
    omitted, it derives from the exception being handled
    (``_resolve_error_type``), every call site records from an ``except``
    block, so the derived value is the caught exception's class.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _dispatch_failures.add(
        1, {"queue": _bounded_queue(queue), "error_type": _resolve_error_type(error_type)}
    )

record_election_attempt

record_election_attempt(
    worker_id: str, *, won: bool
) -> None

Record a leader election attempt.

Always increments election_attempts; increments election_failures only when the attempt did not win the lock. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_election_attempt(worker_id: str, *, won: bool) -> None:
    """Record a leader election attempt.

    Always increments ``election_attempts``; increments ``election_failures``
    only when the attempt did not win the lock.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    del worker_id  # Why: not a dimension -- see the cardinality note above.
    _leader_election_attempts.add(1)
    if not won:
        _leader_election_failures.add(1)

record_enqueue_dedup

record_enqueue_dedup(dedup_reason: str) -> None

Bump the enqueue.dedups counter.

Called from the shared dedup-report helper (backend/_enqueue.py::_log_enqueue_dedup) at every dedup hit, on both backends, the log lines are per-hit observability, and the per-hit terminal-target WARNING arm is budget-bounded at batch scale, so the RATE a stampede produces has no log channel left to ride on; this counter is that rate. dedup_reason is the bounded enum of reasons a hit can occur (unique_for | idempotency_key | zombie_ack_lost), the same value the helper logs, never a caller-controlled string. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_enqueue_dedup(dedup_reason: str) -> None:
    """Bump the enqueue.dedups counter.

    Called from the shared dedup-report helper
    (``backend/_enqueue.py::_log_enqueue_dedup``) at every dedup hit, on
    both backends, the log lines are per-hit observability, and the
    per-hit terminal-target WARNING arm is budget-bounded at batch
    scale, so the RATE a stampede produces has no log channel left to
    ride on; this counter is that rate. ``dedup_reason`` is the bounded
    enum of reasons a hit can occur (``unique_for`` |
    ``idempotency_key`` | ``zombie_ack_lost``), the same value the
    helper logs, never a caller-controlled string.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.enqueue.dedups",
        description=(
            "Enqueue dedup hits (an enqueue returned an existing row instead "
            "of writing one). Attributes: dedup_reason ('unique_for' | "
            "'idempotency_key' | 'zombie_ack_lost'). The per-hit log lines "
            "are budget-bounded at batch scale; this counter is the rate "
            "signal that survives the bound."
        ),
    ).add(1, {"dedup_reason": dedup_reason})

record_error_reporter_failure

record_error_reporter_failure(reporter_type: str) -> None

Bump the error_reporter.failures counter.

Called at the error-reporter catch site when report() raises. reporter_type is the exception-safe class name of the reporter instance (bounded cardinality, one per registered implementation). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_error_reporter_failure(reporter_type: str) -> None:
    """Bump the error_reporter.failures counter.

    Called at the error-reporter catch site when ``report()`` raises.
    ``reporter_type`` is the exception-safe class name of the reporter
    instance (bounded cardinality, one per registered implementation).
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _error_reporter_failures.add(1, {"reporter_type": reporter_type})

record_expired_archive_jobs

record_expired_archive_jobs(
    status: str, count: int = 1
) -> None

Bump the expired_archive.jobs counter.

Called at the archive expiry sweep call site in leader.py. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_expired_archive_jobs(status: str, count: int = 1) -> None:
    """Bump the expired_archive.jobs counter.

    Called at the archive expiry sweep call site in leader.py.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _expired_archive_jobs.add(count, {"status": status})

record_heartbeat_miss

record_heartbeat_miss(worker_id: str) -> None

Bump the heartbeat.misses counter.

Called in heartbeat.py on each heartbeat renewal failure. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_heartbeat_miss(worker_id: str) -> None:
    """Bump the heartbeat.misses counter.

    Called in heartbeat.py on each heartbeat renewal failure.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    del worker_id  # Why: not a dimension -- see the cardinality note above.
    _heartbeat_misses.add(1)

record_job_abandoned

record_job_abandoned(actor: str) -> None

Count one job abandoned by an operator cancel that outlasted its graces.

Called from mark_abandoned on both backends once the abandon write applied: the actor was asked to stop, then forced, and never exited, so the row is taken from it. Shutdowns never produce this, a deploy interrupts running attempts back to the fleet instead, which is why any non-zero rate is worth a page. Attributes: actor. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_job_abandoned(actor: str) -> None:
    """Count one job abandoned by an operator cancel that outlasted its graces.

    Called from ``mark_abandoned`` on both backends once the abandon write
    applied: the actor was asked to stop, then forced, and never exited, so
    the row is taken from it. Shutdowns never produce this, a deploy
    interrupts running attempts back to the fleet instead, which is why
    any non-zero rate is worth a page. Attributes: actor.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.jobs.abandoned",
        description=(
            "Jobs abandoned: an operator cancel outlasted the cooperative and "
            "forced grace periods and the running attempt was taken away. "
            "Never produced by a shutdown (those interrupt). Attributes: actor."
        ),
    ).add(1, {"actor": actor})

record_job_interrupted

record_job_interrupted(actor: str, *, held: bool) -> None

Bump the jobs.interrupted counter.

Called when a worker shutdown releases a running attempt back to the fleet (mark_interrupted landed). held buckets the release by whether the row was parked behind a hold (the actor was still running when the graces expired) or re-pended immediately, the split an operator reads to see whether deploys are interrupting responsive or unresponsive actors. Attributes: actor, hold ("0" | ">0"). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_job_interrupted(actor: str, *, held: bool) -> None:
    """Bump the jobs.interrupted counter.

    Called when a worker shutdown releases a running attempt back to the
    fleet (``mark_interrupted`` landed). ``held`` buckets the release by
    whether the row was parked behind a hold (the actor was still running
    when the graces expired) or re-pended immediately, the split an
    operator reads to see whether deploys are interrupting responsive or
    unresponsive actors. Attributes: actor, hold ("0" | ">0").
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.jobs.interrupted",
        description=(
            "Running attempts released back to the fleet by a worker "
            "shutdown (the spent attempt stands: the attempt did start "
            "executing, so its increment is not refunded and re-dispatch "
            "advances the attempt epoch). "
            "Attributes: actor, hold ('0' | '>0')."
        ),
    ).add(1, {"actor": actor, "hold": ">0" if held else "0"})

record_job_interrupted_noop

record_job_interrupted_noop(actor: str | None) -> None

Bump the jobs.interrupted_noop counter.

Called when mark_interrupted's fence declines the release (the row moved: a reclaim, a terminal write, or an operator cancel in flight owns it). A silent no-op here is the failure mode the project rule names, an interruption that looks released but never landed, so the fenced-out path is instrumented alongside the success path. actor is None when the fenced-out read cannot attribute one. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_job_interrupted_noop(actor: str | None) -> None:
    """Bump the jobs.interrupted_noop counter.

    Called when ``mark_interrupted``'s fence declines the release (the row
    moved: a reclaim, a terminal write, or an operator cancel in flight
    owns it). A silent no-op here is the failure mode the project rule
    names, an interruption that looks released but never landed, so the
    fenced-out path is instrumented alongside the success path.
    ``actor`` is None when the fenced-out read cannot attribute one.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.jobs.interrupted_noop",
        description=(
            "mark_interrupted calls declined by the fence (row not "
            "running-owned at the attempt epoch, or an operator cancel in "
            "flight). Attributes: actor (when attributable)."
        ),
    ).add(1, {"actor": actor if actor is not None else ""})

record_job_timeout

record_job_timeout(
    actor: str, *, kind: TimeoutKind, count: int = 1
) -> None

Count count jobs that exceeded a time budget.

start_to_close is recorded at the timeout handler once per attempt that hit its per-attempt budget, retried or not. schedule_to_close is recorded wherever the whole-job deadline is what ended it: the deadline sweep (which counts a batch at a time), and the handler arms where the backend's deadline arbitration refused a retry, a snooze or a denial's requeue with DeadlineExceeded. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_job_timeout(actor: str, *, kind: TimeoutKind, count: int = 1) -> None:
    """Count *count* jobs that exceeded a time budget.

    ``start_to_close`` is recorded at the timeout handler once per
    attempt that hit its per-attempt budget, retried or not.
    ``schedule_to_close`` is recorded wherever the whole-job deadline is
    what ended it: the deadline sweep (which counts a batch at a time),
    and the handler arms where the backend's deadline arbitration refused
    a retry, a snooze or a denial's requeue with ``DeadlineExceeded``.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.jobs.timeouts",
        description=(
            "Jobs that exceeded a time budget. Attributes: actor, kind "
            "('start_to_close', the per-attempt budget, at the timeout "
            "handler; 'schedule_to_close', the whole-job deadline, at the "
            "deadline sweep and the handler arms the backend refused with "
            "DeadlineExceeded)."
        ),
    ).add(count, {"actor": actor, "kind": kind})

record_leader_lease_expires_in_seconds

record_leader_lease_expires_in_seconds(
    worker_id: str, remaining_ttl: float
) -> None

Stamp the leader lease's TTL as of this process's last elect/renew.

The leader-side mirror of :func:record_lock_expires_in_seconds (heartbeat.py), but a gauge, not a histogram, because the contract that matters for the lease is FRESHNESS, not distribution: the series is present only on the pod that holds the lease, its value is the TTL the server just stamped (leader_lease seconds out), and a series that stops moving or vanishes is a leader that stopped renewing, the split-brain/no-leader detector's per-pod evidence. Called at election win and each successful renewal in worker/leader.py; cleared on demotion by :func:clear_leader_lease_expires_in_seconds so a demoted pod never keeps claiming a lease it no longer holds. Respects _otel_enabled, no-op when False.

Rebind, never write in place: the cache is read on the OTel reader thread while this runs on the event-loop thread (same discipline as :func:record_sweep_success).

Source code in src/taskq/obs/_otel.py

def record_leader_lease_expires_in_seconds(worker_id: str, remaining_ttl: float) -> None:
    """Stamp the leader lease's TTL as of this process's last elect/renew.

    The leader-side mirror of :func:`record_lock_expires_in_seconds`
    (heartbeat.py), but a gauge, not a histogram, because the contract
    that matters for the lease is FRESHNESS, not distribution: the series
    is present only on the pod that holds the lease, its value is the
    TTL the server just stamped (``leader_lease`` seconds out), and a
    series that stops moving or vanishes is a leader that stopped
    renewing, the split-brain/no-leader detector's per-pod evidence.
    Called at election win and each successful renewal in
    ``worker/leader.py``; cleared on demotion by
    :func:`clear_leader_lease_expires_in_seconds` so a demoted pod never
    keeps claiming a lease it no longer holds.
    Respects ``_otel_enabled``, no-op when False.

    Rebind, never write in place: the cache is read on the OTel reader
    thread while this runs on the event-loop thread (same discipline as
    :func:`record_sweep_success`).
    """
    if not _otel_enabled:
        return
    del worker_id  # Why: not a dimension -- see the cardinality note above.
    global _leader_lease_expires_in_seconds_cache
    _leader_lease_expires_in_seconds_cache = remaining_ttl

record_lock_contention

record_lock_contention(lock_name: str) -> None

Count an advisory-lock acquisition lost to another session.

Recorded by the LOSING side at every maintenance acquisition point (leader election, prune, archive-expiry). Contention is the signal that would have exposed two schemas in one database silently sharing one lock, and a sustained rate equal to the attempt rate means the loser never wins at all. Cron's contention stays on its own dedicated counter (taskq.cron.lock_contention). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_lock_contention(lock_name: str) -> None:
    """Count an advisory-lock acquisition lost to another session.

    Recorded by the LOSING side at every maintenance acquisition point
    (leader election, prune, archive-expiry). Contention is the signal that
    would have exposed two schemas in one database silently sharing one
    lock, and a sustained rate equal to the attempt rate means the loser
    never wins at all. Cron's contention stays on its own dedicated
    counter (``taskq.cron.lock_contention``).
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lock_contention.add(1, {"lock": lock_name})

record_lock_expires_in_seconds

record_lock_expires_in_seconds(
    worker_id: str, remaining_ttl: float
) -> None

Record the measured remaining lock TTL on the histogram.

Called in heartbeat.py at each successful renewal after the first, with the lease the previous renewal stamped minus the time elapsed since, a measurement, never the configured constant, so the lock-expiry alert can fire when renewals run late. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_lock_expires_in_seconds(worker_id: str, remaining_ttl: float) -> None:
    """Record the measured remaining lock TTL on the histogram.

    Called in heartbeat.py at each successful renewal after the first,
    with the lease the previous renewal stamped minus the time elapsed
    since, a measurement, never the configured constant, so the
    lock-expiry alert can fire when renewals run late.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    del worker_id  # Why: not a dimension -- see the cardinality note above.
    _lock_expires_in_seconds.record(remaining_ttl)

record_loop_stall_attribution

record_loop_stall_attribution(
    actor: str | None, *, kind: str
) -> None

Count one event-loop stall attributed to the frame holding the GIL.

Called from the loop-lag watchdog's daemon thread at the stall's warn and trip tiers: sampling the main thread's current frame from there is the only attribution that works mid-block, and it needs no loop cooperation. actor is None when no registered actor function appeared in the sampled stack (the block sits under taskq's own code or a non-actor coroutine). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_loop_stall_attribution(actor: str | None, *, kind: str) -> None:
    """Count one event-loop stall attributed to the frame holding the GIL.

    Called from the loop-lag watchdog's daemon thread at the stall's warn
    and trip tiers: sampling the main thread's current frame from there is
    the only attribution that works mid-block, and it needs no loop
    cooperation. ``actor`` is ``None`` when no registered actor function
    appeared in the sampled stack (the block sits under taskq's own code
    or a non-actor coroutine). Respects ``_otel_enabled``, no-op when
    False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.worker.loop_stall_attributions",
        description=(
            "Event-loop scheduling stalls attributed to the synchronous work "
            "holding the interpreter while the loop could not schedule. "
            "kind='blocking_call' marks a synchronous call that released the "
            "GIL (I/O wait, a subprocess); kind='gil_held' marks synchronous "
            "work that held it (a C extension without GIL release, or a hot "
            "pure-Python loop). The actor label names the registered actor "
            "function whose frame sat under the blocking call; run it down by "
            "moving the cited work off the event loop. Attributes: actor, kind."
        ),
    ).add(
        1,
        {
            "actor": (_bounded_cron_actor(actor) if actor is not None else _ACTOR_LABEL_OVERFLOW),
            "kind": kind,
        },
    )

record_pool_acquire_duration

record_pool_acquire_duration(
    queue: str, elapsed: float
) -> None

Record the dispatch round's pool-acquire wait on the histogram.

Called outside the dispatch span body for sampling independence. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_pool_acquire_duration(queue: str, elapsed: float) -> None:
    """Record the dispatch round's pool-acquire wait on the histogram.

    Called outside the ``dispatch`` span body for sampling independence.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _pool_acquire_duration.record(elapsed, {"queue": _bounded_queue(queue)})

record_process_duration

record_process_duration(
    actor: str,
    queue: str,
    elapsed: float,
    *,
    outcome: ConsumedOutcome,
) -> None

Record job execution duration on the histogram.

Called outside the CONSUMER span body for sampling independence, with the same outcome the consumed-messages counter records for the attempt: a start_to_close timeout lands at exactly the budget and a failure at whatever it took, and either would drag a success percentile if the distributions were shared. Respects _otel_enabled, no-op when False. Finer resolution is an operator's SDK View away (the shipped boundaries are seconds-scaled, not operator-tuned).

Source code in src/taskq/obs/_otel.py

def record_process_duration(
    actor: str, queue: str, elapsed: float, *, outcome: ConsumedOutcome
) -> None:
    """Record job execution duration on the histogram.

    Called outside the CONSUMER span body for sampling independence, with
    the same ``outcome`` the consumed-messages counter records for the
    attempt: a ``start_to_close`` timeout lands at exactly the budget and
    a failure at whatever it took, and either would drag a success
    percentile if the distributions were shared.
    Respects ``_otel_enabled``, no-op when False.
    Finer resolution is an operator's SDK View away (the shipped
    boundaries are seconds-scaled, not operator-tuned).
    """
    if not _otel_enabled:
        return
    _process_duration.record(
        elapsed, {"actor": actor, "queue": _bounded_queue(queue), "outcome": outcome}
    )

record_progress_flush_failure

record_progress_flush_failure(
    stage: str, error_type: str
) -> None

Bump the progress.flush_failures counter.

stage must be 'per_job' (one job's flush UPDATE failed, that job's progress since the last flush is lost) or 'pool' (a pool could not be obtained at all, the loop-level getter failed, or the per-job acquire failed/exhausted, so every job's progress is lost). The two are materially different incidents and must stay distinguishable in an alert rule, which is also why both pool-stage sites log a different kind than the per-job one. error_type is the exception class name. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_progress_flush_failure(stage: str, error_type: str) -> None:
    """Bump the progress.flush_failures counter.

    ``stage`` must be ``'per_job'`` (one job's flush UPDATE failed, that
    job's progress since the last flush is lost) or ``'pool'`` (a pool
    could not be obtained at all, the loop-level getter failed, or the
    per-job acquire failed/exhausted, so every job's progress is lost).
    The two are materially different incidents and must stay
    distinguishable in an alert rule, which is also why both pool-stage
    sites log a different kind than the per-job one.
    ``error_type`` is the exception class name.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.progress.flush_failures",
        description=(
            "Progress flush failures, by stage. Attributes: stage "
            "('per_job' | 'pool'), error_type (exception class name)."
        ),
    ).add(1, {"stage": stage, "error_type": error_type})

record_progress_publish_failure

record_progress_publish_failure(
    channel: str, error_type: str
) -> None

Bump the progress.publish_failures counter.

channel must be 'per_job' or 'global', bounded cardinality. error_type is the exception class name (e.g. 'ResponseError'). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_progress_publish_failure(channel: str, error_type: str) -> None:
    """Bump the progress.publish_failures counter.

    ``channel`` must be ``'per_job'`` or ``'global'``, bounded cardinality.
    ``error_type`` is the exception class name (e.g. ``'ResponseError'``).
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _progress_publish_failures.add(1, {"channel": channel, "error_type": error_type})

record_pruned_jobs

record_pruned_jobs(
    actor: str, status: str, count: int = 1
) -> None

Bump the pruned.jobs counter.

Called at the prune sweep call site in leader.py. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_pruned_jobs(actor: str, status: str, count: int = 1) -> None:
    """Bump the pruned.jobs counter.

    Called at the prune sweep call site in leader.py.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _pruned_jobs.add(count, {"actor": actor, "status": status})

record_published_message

record_published_message(actor: str, queue: str) -> None

Bump the published-messages counter.

Called after successful enqueue, outside the PRODUCER span body, to ensure sampling independence. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_published_message(actor: str, queue: str) -> None:
    """Bump the published-messages counter.

    Called after successful enqueue, outside the PRODUCER span body,
    to ensure sampling independence.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _published_messages.add(1, {"actor": actor, "queue": _bounded_queue(queue)})

record_queue_wait

record_queue_wait(
    actor: str, queue: str, waited_seconds: float
) -> None

Record how long a job waited from eligibility to claim.

Called once per dispatch from the claimed row's own server-clock stamps: started_at - scheduled_at, eligibility to claim. The per-job companion of the sampled oldest_pending_age_seconds: the gauge shows the head of the line, this histogram shows what every dispatched job actually waited, retries and re-pends included. Labels are the job-side pair (actor, queue capped as everywhere). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_queue_wait(actor: str, queue: str, waited_seconds: float) -> None:
    """Record how long a job waited from eligibility to claim.

    Called once per dispatch from the claimed row's own server-clock
    stamps: ``started_at - scheduled_at``, eligibility to claim. The
    per-job companion of the sampled ``oldest_pending_age_seconds``: the
    gauge shows the head of the line, this histogram shows what every
    dispatched job actually waited, retries and re-pends included. Labels
    are the job-side pair (actor, queue capped as everywhere).
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_histogram(
        "taskq.jobs.queue_wait_seconds",
        description=(
            "Seconds a job waited between becoming eligible (scheduled_at) "
            "and being claimed (started_at), both server-clock stamps on the "
            "dispatched row. Attributes: actor, queue (capped, see "
            "_bounded_queue)."
        ),
        unit="s",
        # Seconds-scaled: the ops playbook reads this histogram's p99
        # ("queue_wait p99 rises with flat depth is dispatch starvation");
        # sub-second waits need sub-second buckets.
        boundaries=(
            0.001,
            0.005,
            0.01,
            0.025,
            0.05,
            0.1,
            0.25,
            0.5,
            1.0,
            2.5,
            5.0,
            10.0,
            30.0,
            60.0,
            120.0,
            300.0,
            600.0,
            1200.0,
            3600.0,
        ),
    ).record(waited_seconds, {"actor": actor, "queue": _bounded_queue(queue)})

record_ratelimit_acquire_dependency_failure

record_ratelimit_acquire_dependency_failure(
    error_type: str,
) -> None

Bump the ratelimit.acquire_dependency_failures counter.

Called when a rate-limit acquire fails because the limiter's store , Redis, or the PG fallback behind it, could not answer, and the worker failed the acquire closed as a denial. An AVAILABILITY signal, distinct from taskq.reservation.denials, which counts admission decisions: a denial with this counter rising is an outage masquerading as contention, and an operator must read the two together before scaling a bucket. error_type is the exception class name. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_ratelimit_acquire_dependency_failure(error_type: str) -> None:
    """Bump the ratelimit.acquire_dependency_failures counter.

    Called when a rate-limit acquire fails because the limiter's store ,
    Redis, or the PG fallback behind it, could not answer, and the
    worker failed the acquire closed as a denial. An AVAILABILITY
    signal, distinct from ``taskq.reservation.denials``, which counts
    admission decisions: a denial with this counter rising is an outage
    masquerading as contention, and an operator must read the two
    together before scaling a bucket. ``error_type`` is the exception
    class name.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.ratelimit.acquire_dependency_failures",
        description=(
            "Rate-limit acquires that failed on a store dependency (Redis "
            "or the PG fallback) and were failed closed as denials, an "
            "availability signal, distinct from reservation.denials "
            "(admission decisions). Attributes: error_type (exception "
            "class name)."
        ),
    ).add(1, {"error_type": error_type})

record_ratelimit_denial

record_ratelimit_denial(backend: str) -> None

Bump the ratelimit.denials counter.

Called from the rate-limit decision logger whenever a decision denies admission. backend is the rate-limit backend enum (bounded). The bucket name is deliberately NOT a dimension: keyed bucket names are caller-derived and unbounded, so a bucket label would reintroduce the cardinality class the queue-label cap exists to prevent. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_ratelimit_denial(backend: str) -> None:
    """Bump the ratelimit.denials counter.

    Called from the rate-limit decision logger whenever a decision denies
    admission. ``backend`` is the rate-limit backend enum (bounded). The
    bucket name is deliberately NOT a dimension: keyed bucket names are
    caller-derived and unbounded, so a bucket label would reintroduce the
    cardinality class the queue-label cap exists to prevent.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.ratelimit.denials",
        description=(
            "Rate-limit decisions that denied admission. Attributes: "
            "backend (rate-limit backend). Bucket names are not a "
            "dimension (caller-controlled cardinality)."
        ),
    ).add(1, {"backend": backend})

record_ratelimit_refund_failure

record_ratelimit_refund_failure(
    bucket: str, backend: str, error_type: str | None = None
) -> None

Bump the ratelimit.refund_failures counter.

Called at the rate-limit refund failure catch site. bucket is the handle's concrete bucket name, admitted through the same cap as queue / the cron actor: a keyed bucket's name embeds a caller-controlled key (base_name:key), so carried as-is it would mint one never-released OTel series per distinct key ever seen (see the cardinality note above _bounded_bucket). error_type is the exception class name; omitted, it derives from the exception being handled (_resolve_error_type). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_ratelimit_refund_failure(
    bucket: str, backend: str, error_type: str | None = None
) -> None:
    """Bump the ratelimit.refund_failures counter.

    Called at the rate-limit refund failure catch site. ``bucket`` is the
    handle's concrete bucket name, admitted through the same cap as
    ``queue`` / the cron ``actor``: a keyed bucket's name embeds a
    caller-controlled key (``base_name:key``), so carried as-is it would
    mint one never-released OTel series per distinct key ever seen (see
    the cardinality note above ``_bounded_bucket``). ``error_type`` is
    the exception class name; omitted, it derives from the exception being
    handled (``_resolve_error_type``).
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _ratelimit_refund_failures.add(
        1,
        {
            "bucket": _bounded_bucket(bucket),
            "backend": backend,
            "error_type": _resolve_error_type(error_type),
        },
    )

record_reclaimed_jobs

record_reclaimed_jobs(
    actor: str, disposition: str, count: int = 1
) -> None

Bump the per-actor, per-disposition reclaimed-jobs counter.

Unconditional (not gated by _otel_enabled): a reclaim is a worker death the fleet recovered from -- the same class of fleet-health signal as :func:record_deadline_exceeded_swept, which must be counted even when OTel is disabled. Recorded aggregated per (actor, disposition) after the sweep's transaction, by the Postgres sweep (from its RETURNING) and by the in-memory twin alike, so the two backends' label sets cannot drift. The disposition values are the code-fixed enum repended / crashed / cancelled (taskq.backend._sweeps._RECLAIM_DISPOSITIONS); actor flows through registration like every other actor-labeled instrument.

Source code in src/taskq/obs/_otel.py

def record_reclaimed_jobs(actor: str, disposition: str, count: int = 1) -> None:
    """Bump the per-actor, per-disposition reclaimed-jobs counter.

    Unconditional (not gated by ``_otel_enabled``): a reclaim is a worker
    death the fleet recovered from -- the same class of fleet-health
    signal as :func:`record_deadline_exceeded_swept`, which must be
    counted even when OTel is disabled. Recorded aggregated per (actor,
    disposition) after the sweep's transaction, by the Postgres sweep
    (from its RETURNING) and by the in-memory twin alike, so the two
    backends' label sets cannot drift. The disposition values are the
    code-fixed enum ``repended`` / ``crashed`` / ``cancelled``
    (``taskq.backend._sweeps._RECLAIM_DISPOSITIONS``); ``actor`` flows
    through registration like every other actor-labeled instrument.
    """
    try:
        _reclaimed_jobs.add(count, {"actor": actor, "disposition": disposition})
    except Exception:
        _log.warning("otel-metric-record-failed", instrument_name="taskq.jobs.reclaimed")

record_reservation_denial

record_reservation_denial(
    bucket_name: str, source: str
) -> None

Bump the reservation.denials counter.

Called at the reservation-class denial handler for every ReservationUnavailable a worker fields. source is 'reservation' or 'rate_limit' (bounded). bucket_name is accepted so the call site reads naturally but is deliberately NOT a dimension: keyed bucket names are caller-derived and unbounded, so the label would reintroduce the cardinality class the queue-label cap exists to prevent. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_reservation_denial(bucket_name: str, source: str) -> None:
    """Bump the reservation.denials counter.

    Called at the reservation-class denial handler for every
    ``ReservationUnavailable`` a worker fields. ``source`` is
    ``'reservation'`` or ``'rate_limit'`` (bounded). ``bucket_name`` is
    accepted so the call site reads naturally but is deliberately NOT a
    dimension: keyed bucket names are caller-derived and unbounded, so
    the label would reintroduce the cardinality class the queue-label cap
    exists to prevent.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    del bucket_name  # Why: not a dimension -- see the cardinality note above.
    _lazy_counter(
        "taskq.reservation.denials",
        description=(
            "Reservation/rate-limit admission denials surfaced to a worker "
            "handler. Attributes: source ('reservation' | 'rate_limit'). "
            "Bucket names are not a dimension (caller-controlled "
            "cardinality)."
        ),
    ).add(1, {"source": source})

record_reservation_reclaim_drain_duration

record_reservation_reclaim_drain_duration(
    elapsed_seconds: float,
) -> None

Record the wall-clock duration of one reclaim drain statement.

Recorded on success and failure alike (callers pass it from a finally), so a timeout that aborted the drain still leaves a duration sample. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_reservation_reclaim_drain_duration(elapsed_seconds: float) -> None:
    """Record the wall-clock duration of one reclaim drain statement.

    Recorded on success and failure alike (callers pass it from a
    ``finally``), so a timeout that aborted the drain still leaves a
    duration sample.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_histogram(
        "taskq.ratelimit.reclaim_drain_duration",
        description="Wall-clock duration of one keyed-reservation reclaim drain statement.",
        unit="s",
    ).record(elapsed_seconds)

record_reservation_reclaim_drain_failure

record_reservation_reclaim_drain_failure(
    error_type: str,
) -> None

Bump the ratelimit.reclaim_drain_failures counter.

Called when the keyed-reservation slot-row reclaim drain fails. A persistently failing drain strands reservation_slots rows, a STORAGE signal, and the pending-depth gauge shows the backlog forming. Distinct from heal failures, which are an AVAILABILITY signal; the two must not share a counter. error_type is the exception class name. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_reservation_reclaim_drain_failure(error_type: str) -> None:
    """Bump the ratelimit.reclaim_drain_failures counter.

    Called when the keyed-reservation slot-row reclaim drain fails. A
    persistently failing drain strands ``reservation_slots`` rows, a
    STORAGE signal, and the pending-depth gauge shows the backlog
    forming. Distinct from heal failures, which are an AVAILABILITY
    signal; the two must not share a counter.
    ``error_type`` is the exception class name.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.ratelimit.reclaim_drain_failures",
        description=(
            "Failures of the keyed-reservation slot-row reclaim drain (a "
            "failed drain strands reservation_slots rows -- storage). "
            "Attributes: error_type. Distinct from reclaim_heal_failures "
            "(availability)."
        ),
    ).add(1, {"error_type": error_type})

record_reservation_reclaim_drain_rows

record_reservation_reclaim_drain_rows(rows: int) -> None

Count rows deleted by one keyed-reclaim drain, reservation_slots slot rows and rate_limit_buckets bucket rows alike.

Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_reservation_reclaim_drain_rows(rows: int) -> None:
    """Count rows deleted by one keyed-reclaim drain, ``reservation_slots``
    slot rows and ``rate_limit_buckets`` bucket rows alike.

    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.ratelimit.reclaim_drain_rows",
        description=(
            "Rows deleted by the keyed-reclaim drain, reservation_slots and "
            "rate_limit_buckets alike (RETURNING-confirmed)."
        ),
    ).add(rows)

record_reservation_reclaim_heal_failure

record_reservation_reclaim_heal_failure(
    error_type: str,
) -> None

Bump the ratelimit.reclaim_heal_failures counter.

Called when the acquire-path re-materialisation heal for a keyed bucket whose slot rows were deleted by a sibling worker's drain fails. A failing heal denies new admissions for that bucket, an AVAILABILITY signal, the opposite of a drain failure (storage); the two must not share a counter. error_type is the exception class name. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_reservation_reclaim_heal_failure(error_type: str) -> None:
    """Bump the ratelimit.reclaim_heal_failures counter.

    Called when the acquire-path re-materialisation heal for a keyed
    bucket whose slot rows were deleted by a sibling worker's drain
    fails. A failing heal denies new admissions for that bucket, an
    AVAILABILITY signal, the opposite of a drain failure (storage); the
    two must not share a counter.
    ``error_type`` is the exception class name.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.ratelimit.reclaim_heal_failures",
        description=(
            "Failures of the acquire-path re-materialisation heal for keyed "
            "buckets whose slot rows a sibling worker's drain deleted (a "
            "failed heal denies new admissions -- availability). "
            "Attributes: error_type. Distinct from reclaim_drain_failures "
            "(storage)."
        ),
    ).add(1, {"error_type": error_type})

record_slot_pool_acquire_failure

record_slot_pool_acquire_failure(
    error_type: str | None = None,
) -> None

Bump the worker.slot_pool.acquire_failures counter.

Called from the exception branch of the bounded per-job acquire in taskq.worker.dispatch, never the success path, matching record_sweep_timeout's contract. A rate here is what separates one transient timeout from every transactional job on a worker failing to acquire, and error_type is the exception class name , omitted, it derives from the exception being handled (_resolve_error_type). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_slot_pool_acquire_failure(error_type: str | None = None) -> None:
    """Bump the worker.slot_pool.acquire_failures counter.

    Called from the exception branch of the bounded per-job acquire in
    ``taskq.worker.dispatch``, never the success path, matching
    ``record_sweep_timeout``'s contract. A rate here is what separates
    one transient timeout from every transactional job on a worker
    failing to acquire, and ``error_type`` is the exception class name ,
    omitted, it derives from the exception being handled
    (``_resolve_error_type``).
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _slot_pool_acquire_failures.add(1, {"error_type": _resolve_error_type(error_type)})

record_sse_connection_closed

record_sse_connection_closed(
    surface: SseSurface, topic: str
) -> None

Record one SSE connection releasing its slot (the level gauge's -1).

Called from the exact finally/release sites the slot's semaphore is released at, so the gauge can never claim a connection whose slot was handed back.

Source code in src/taskq/obs/_otel.py

def record_sse_connection_closed(surface: SseSurface, topic: str) -> None:
    """Record one SSE connection releasing its slot (the level gauge's -1).

    Called from the exact finally/release sites the slot's semaphore is
    released at, so the gauge can never claim a connection whose slot
    was handed back.
    """
    global _sse_connections_cache
    _sse_connections_cache = _sse_cache_rebind(_sse_connections_cache, surface, topic, -1)

record_sse_connection_opened

record_sse_connection_opened(
    surface: SseSurface, topic: str
) -> None

Record one SSE connection taking a slot (the level gauge's +1).

Called at the acquire site AFTER the slot is held, so the gauge and the semaphore budget move together; the close is recorded at the same site the slot is released (record_sse_connection_closed). topic is the closed vocabulary the endpoints validate before the slot lookup (see the section note above).

Source code in src/taskq/obs/_otel.py

def record_sse_connection_opened(surface: SseSurface, topic: str) -> None:
    """Record one SSE connection taking a slot (the level gauge's +1).

    Called at the acquire site AFTER the slot is held, so the gauge and
    the semaphore budget move together; the close is recorded at the
    same site the slot is released (``record_sse_connection_closed``).
    ``topic`` is the closed vocabulary the endpoints validate before the
    slot lookup (see the section note above).
    """
    global _sse_connections_cache
    _sse_connections_cache = _sse_cache_rebind(_sse_connections_cache, surface, topic, +1)

record_sse_rejection

record_sse_rejection(
    surface: SseSurface, topic: str
) -> None

Count one SSE connection refused at the cap (the 429 site).

Called from the two 429 sites (the admin topic endpoint's inline cap check, and the shared acquire_sse_slot helper the progress stream uses). Same closed label vocabulary as the connections gauge, so the two join: rejections rising beside connections pinned at the cap is saturation; rejections alone is a burst the cap absorbed. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_sse_rejection(surface: SseSurface, topic: str) -> None:
    """Count one SSE connection refused at the cap (the 429 site).

    Called from the two 429 sites (the admin topic endpoint's inline
    cap check, and the shared ``acquire_sse_slot`` helper the progress
    stream uses). Same closed label vocabulary as the connections
    gauge, so the two join: rejections rising beside connections pinned
    at the cap is saturation; rejections alone is a burst the cap
    absorbed.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.admin.sse.rejections",
        description=(
            "SSE connection attempts refused because the topic's "
            "connection cap was already held (HTTP 429). Attributes: "
            "topic, surface (the same closed enums as "
            "taskq.admin.sse.connections, so the two join). The other "
            "half of the saturation signature: connections pinned at "
            "the configured cap beside a rising rejection rate."
        ),
    ).add(1, {"topic": _bounded_sse_topic(topic), "surface": surface})

record_sub_enqueue_failure

record_sub_enqueue_failure(
    actor: str, count: int, error_type: str | None = None
) -> None

Bump the sub_enqueue.failures counter by count failed child enqueues.

Called at the post-commit flush catch site: the parent job has already been reported as succeeded, so every failed child enqueue is a job the caller believes exists but does not. actor is the parent's actor (bounded by the registered actor set). count is the number of child enqueues that failed, so one incident with N lost children records N, not 1. error_type is the exception class name; omitted, it derives from the exception being handled (_resolve_error_type). Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_sub_enqueue_failure(actor: str, count: int, error_type: str | None = None) -> None:
    """Bump the sub_enqueue.failures counter by *count* failed child enqueues.

    Called at the post-commit flush catch site: the parent job has already
    been reported as succeeded, so every failed child enqueue is a job the
    caller believes exists but does not. ``actor`` is the parent's actor
    (bounded by the registered actor set). ``count`` is the number of
    child enqueues that failed, so one incident with N lost children
    records N, not 1. ``error_type`` is the exception class name; omitted,
    it derives from the exception being handled (``_resolve_error_type``).
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _lazy_counter(
        "taskq.sub_enqueue.failures",
        description=(
            "Sub-enqueue flush failures after the parent job committed; "
            "each counted unit is one child job that was reported as "
            "enqueued but was not. Attributes: actor (parent job's actor), "
            "error_type (exception class name, a closed set; see "
            "_resolve_error_type)."
        ),
    ).add(count, {"actor": actor, "error_type": _resolve_error_type(error_type)})

record_sweep_batch_size

record_sweep_batch_size(
    sweep_name: str, batch_size: int
) -> None

Record the batch size a sweep call actually used.

The maintenance sweeps degrade to a reduced batch after repeated cancellations; a worker reporting the reduced tier is reporting an unhealthy database and must not be silent about it. Feeds BOTH the batch-size gauge below AND maintenance_health's reduced-tier view (health.py reads the cache directly), so it is deliberately NOT gated on _otel_enabled: the health body must keep reporting a latched reduced tier when an operator has exported telemetry off, because that body is the Prometheus-free surface the degraded signal exists for.

Source code in src/taskq/obs/_otel.py

def record_sweep_batch_size(sweep_name: str, batch_size: int) -> None:
    """Record the batch size a sweep call actually used.

    The maintenance sweeps degrade to a reduced batch after repeated
    cancellations; a worker reporting the reduced tier is reporting an
    unhealthy database and must not be silent about it. Feeds BOTH the
    batch-size gauge below AND ``maintenance_health``'s reduced-tier view
    (health.py reads the cache directly), so it is deliberately NOT gated
    on ``_otel_enabled``: the health body must keep reporting a latched
    reduced tier when an operator has exported telemetry off, because that
    body is the Prometheus-free surface the degraded signal exists for.
    """
    update_sweep_batch_size_cache(sweep_name, batch_size)

record_sweep_batch_size_configured

record_sweep_batch_size_configured(
    sweep_name: str, configured_size: int
) -> None

Record the batch size this worker is CONFIGURED to use for sweep_name.

Emitted at the same call site as :func:record_sweep_batch_size so the two gauges carry the same sweep_name label set and a gauge-to-gauge comparison is always label-matched. The alert pair (used vs configured) is what makes the sweep-degraded signal track per-worker event_writer_batch_size: a literal threshold is blind on every deployment whose configured size is not the default. Respects _otel_enabled, no-op when False (its only consumer is the OTel gauge, unlike :func:record_sweep_batch_size).

Rebind, never write in place, same cross-thread reader discipline as :func:record_sweep_success.

Source code in src/taskq/obs/_otel.py

def record_sweep_batch_size_configured(sweep_name: str, configured_size: int) -> None:
    """Record the batch size this worker is CONFIGURED to use for *sweep_name*.

    Emitted at the same call site as :func:`record_sweep_batch_size` so
    the two gauges carry the same ``sweep_name`` label set and a
    gauge-to-gauge comparison is always label-matched. The alert pair
    (used vs configured) is what makes the sweep-degraded signal track
    per-worker ``event_writer_batch_size``: a literal threshold is blind
    on every deployment whose configured size is not the default.
    Respects ``_otel_enabled``, no-op when False (its only consumer is
    the OTel gauge, unlike :func:`record_sweep_batch_size`).

    Rebind, never write in place, same cross-thread reader discipline as
    :func:`record_sweep_success`.
    """
    if not _otel_enabled:
        return
    global _sweep_batch_size_configured_cache
    _sweep_batch_size_configured_cache = {
        **_sweep_batch_size_configured_cache,
        sweep_name: configured_size,
    }

record_sweep_success

record_sweep_success(sweep_name: str) -> None

Stamp a sweep call's success in BOTH clock domains.

Feeds the staleness gauge below (the wall ledger, an absolute Unix timestamp) and maintenance_health's stalled view (the monotonic ledger, an elapsed-time anchor, see _sweep_success_monotonic_cache for why the health view must not read the wall ledger): time() - last_success answers "is this sweep still making progress?" independently of row counts, so a sweep that finds zero eligible rows every tick (healthy) is distinguishable from one that never completes (stalled). The row-count counter and the duration histogram share this call site but NOT this sample population: a timed-out sweep records duration and a timeout but no row sample, so rows and duration must be read as different populations, which the sweep_timeouts counter reconciles.

Rebind, never write in place: the caches are iterated on the OTel reader thread while this runs on the event-loop thread (see the section comment above).

Source code in src/taskq/obs/_otel.py

def record_sweep_success(sweep_name: str) -> None:
    """Stamp a sweep call's success in BOTH clock domains.

    Feeds the staleness gauge below (the wall ledger, an absolute Unix
    timestamp) and ``maintenance_health``'s stalled view (the monotonic
    ledger, an elapsed-time anchor, see
    ``_sweep_success_monotonic_cache`` for why the health view must not
    read the wall ledger): ``time() - last_success`` answers "is this
    sweep still making progress?" independently of row counts, so a
    sweep that finds zero eligible rows every tick (healthy) is
    distinguishable from one that never completes (stalled). The
    row-count counter and the duration histogram share this call site
    but NOT this sample population: a timed-out sweep records duration
    and a timeout but no row sample, so rows and duration must be read
    as different populations, which the sweep_timeouts counter
    reconciles.

    Rebind, never write in place: the caches are iterated on the OTel
    reader thread while this runs on the event-loop thread (see the
    section comment above).
    """
    global _sweep_success_cache, _sweep_success_monotonic_cache
    _sweep_success_cache = {**_sweep_success_cache, sweep_name: time.time()}
    _sweep_success_monotonic_cache = {
        **_sweep_success_monotonic_cache,
        sweep_name: time.monotonic(),
    }

record_sweep_timeout

record_sweep_timeout(sweep_name: str) -> None

Count a sweep or sampler read that did not complete.

Called on the failure path, never the success path. A sweep reports the deadline family (TimeoutError / QueryCanceledError), where the distinction between aborted and merely slow is the actionable one; a gauge sampler reports every failure, because its gauge keeps serving its last value either way and the read not happening is the whole fault. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_sweep_timeout(sweep_name: str) -> None:
    """Count a sweep or sampler read that did not complete.

    Called on the failure path, never the success path. A sweep reports the
    deadline family (``TimeoutError`` / ``QueryCanceledError``), where the
    distinction between aborted and merely slow is the actionable one; a
    gauge sampler reports every failure, because its gauge keeps serving its
    last value either way and the read not happening is the whole fault.
    Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _sweep_timeouts.add(1, {"sweep_name": sweep_name})

record_sweep_unexpected_error

record_sweep_unexpected_error(sweep_name: str) -> None

Count a sweep batch that failed with an error OUTSIDE the deadline family.

Called on the failure path, never the success path, from the prune family's batch machinery: a batch aborted by an archive UniqueViolation-class error (or any other transient fault the deadline family does not name) never reaches record_sweep_timeout, and the sweep's own row counters and success stamps only move on success, so without this counter a stopped drain reads healthy on every metric plane. Respects _otel_enabled, no-op when False.

Source code in src/taskq/obs/_otel.py

def record_sweep_unexpected_error(sweep_name: str) -> None:
    """Count a sweep batch that failed with an error OUTSIDE the deadline
    family.

    Called on the failure path, never the success path, from the prune
    family's batch machinery: a batch aborted by an archive
    UniqueViolation-class error (or any other transient fault the deadline
    family does not name) never reaches ``record_sweep_timeout``, and the
    sweep's own row counters and success stamps only move on success, so
    without this counter a stopped drain reads healthy on every metric
    plane. Respects ``_otel_enabled``, no-op when False.
    """
    if not _otel_enabled:
        return
    _sweep_unexpected_errors.add(1, {"sweep_name": sweep_name})

safe_start_span

safe_start_span(
    name: str,
    *,
    kind: SpanKind | None = None,
    attributes: Attributes = None,
    links: Sequence[Link] | None = None,
    new_root: bool = False,
) -> Generator[Span, None, None]

Start a span safely, never propagates exceptions from OTel API calls.

Checks _otel_enabled first; when False, yields a no-op NonRecordingSpan. When True, delegates to get_tracer().start_as_current_span with a try/except around span creation only. Exceptions from code inside the with block propagate normally, only OTel API failures (misconfiguration, exporter unavailability) are suppressed.

When new_root=True, passes an empty Context() so the span has no parent, it is a root span linked (not parented) to the ambient trace. This satisfies the "linked, not parented" requirement for PRODUCER spans in the cron loop.

The SDK's own exception handling is switched OFF and replaced by :func:_record_scrubbed_error, see the comment at the call below.

Source code in src/taskq/obs/_otel.py

@contextlib.contextmanager
def safe_start_span(
    name: str,
    *,
    kind: trace.SpanKind | None = None,
    attributes: Attributes = None,
    links: Sequence[trace.Link] | None = None,
    new_root: bool = False,
) -> Generator[Span, None, None]:
    """Start a span safely, never propagates exceptions from OTel API calls.

    Checks ``_otel_enabled`` first; when ``False``, yields a no-op
    ``NonRecordingSpan``. When ``True``, delegates to
    ``get_tracer().start_as_current_span`` with a ``try/except`` around
    span *creation* only. Exceptions from code inside the ``with`` block
    propagate normally, only OTel API failures (misconfiguration,
    exporter unavailability) are suppressed.

    When ``new_root=True``, passes an empty ``Context()`` so the span
    has no parent, it is a root span linked (not parented) to the
    ambient trace. This satisfies the "linked, not parented"
    requirement for PRODUCER spans in the cron loop.

    The SDK's own exception handling is switched OFF and replaced by
    :func:`_record_scrubbed_error`, see the comment at the call below.
    """
    if not _otel_enabled:
        yield trace.NonRecordingSpan(_NOOP_SPAN_CONTEXT)
        return

    ctx: Context | None = Context() if new_root else None

    try:
        span_cm = get_tracer().start_as_current_span(
            name,
            context=ctx,
            kind=kind if kind is not None else trace.SpanKind.INTERNAL,
            attributes=attributes,
            links=links,
            # Why: both default to True, and the SDK derives its event and its
            # status description from the RAW ``str(exc)`` with no hook to
            # override it. At every call site that scrubs and re-raises, that
            # emitted a SECOND, unscrubbed ``exception`` event and overwrote
            # the scrubbed status description -- shipping the Postgres DETAIL
            # row values (idempotency_key / identity_key / fairness_key, all
            # caller-supplied) straight to the telemetry backend, undoing
            # ``_redact_exc`` entirely. No source-level guard can see this:
            # the leaking call is made by the SDK, not by TaskQ.
            record_exception=False,
            set_status_on_exception=False,
        )
    except Exception:
        _log.warning("otel-span-creation-failed", span_name=name)
        yield trace.NonRecordingSpan(_NOOP_SPAN_CONTEXT)
        return

    with span_cm as span:
        try:
            yield span
        except Exception as exc:
            # ``Exception``, not ``BaseException``: mirrors what the SDK's own
            # ``use_span`` catches, so cancellation semantics are unchanged.
            _record_scrubbed_error(span, exc)
            raise

set_otel_enabled

set_otel_enabled(enabled: bool) -> None

Set the module-level OTel enabled flag.

Called by worker startup code after loading WorkerSettings so that all safe helpers check the flag without requiring a WorkerSettings import at every call site. Avoids circular imports (modules like dispatch.py import from obs and should not import from settings.py in a circular path).

Source code in src/taskq/obs/_otel.py

def set_otel_enabled(enabled: bool) -> None:
    """Set the module-level OTel enabled flag.

    Called by worker startup code after loading ``WorkerSettings`` so that
    all safe helpers check the flag without requiring a ``WorkerSettings``
    import at every call site. Avoids circular imports (modules like
    ``dispatch.py`` import from ``obs`` and should not import from
    ``settings.py`` in a circular path).
    """
    global _otel_enabled
    _otel_enabled = enabled

set_slot_pool_occupancy_source

set_slot_pool_occupancy_source(
    pool: _PoolOccupancySource | None,
) -> None

Point the slot-pool occupancy gauge at pool.

Called at slot-pool bootstrap open and whenever a credential reload swaps the pool, so the gauge always reads the live one. None clears the source, the gauge reports nothing, matching the pool-not-open state.

Source code in src/taskq/obs/_otel.py

def set_slot_pool_occupancy_source(pool: _PoolOccupancySource | None) -> None:
    """Point the slot-pool occupancy gauge at *pool*.

    Called at slot-pool bootstrap open and whenever a credential reload
    swaps the pool, so the gauge always reads the live one. ``None``
    clears the source, the gauge reports nothing, matching the
    pool-not-open state.
    """

    global _slot_pool_occupancy_source
    _slot_pool_occupancy_source = pool

set_worker_capacity_source

set_worker_capacity_source(
    active_jobs: _ActiveJobsSource | None,
    max_concurrency: int,
) -> None

Point the worker capacity gauges at active_jobs and max_concurrency.

Called once at worker bootstrap, after the deps that own the registry exist. None clears the source, both gauges report nothing, matching a process that hosts no worker.

Source code in src/taskq/obs/_otel.py

def set_worker_capacity_source(active_jobs: _ActiveJobsSource | None, max_concurrency: int) -> None:
    """Point the worker capacity gauges at *active_jobs* and *max_concurrency*.

    Called once at worker bootstrap, after the deps that own the
    registry exist. ``None`` clears the source, both gauges report
    nothing, matching a process that hosts no worker.
    """
    global _worker_capacity_source
    _worker_capacity_source = (active_jobs, max_concurrency) if active_jobs is not None else None

update_actor_backlog_cache

update_actor_backlog_cache(
    data: dict[tuple[str, str], int],
) -> None

Replace the per-(actor, queue) pending-depth cache with fresh data.

Source code in src/taskq/obs/_otel.py

def update_actor_backlog_cache(data: dict[tuple[str, str], int]) -> None:
    """Replace the per-(actor, queue) pending-depth cache with fresh data."""
    global _actor_backlog_cache
    _actor_backlog_cache = dict(data)

update_actor_oldest_pending_age_cache

update_actor_oldest_pending_age_cache(
    data: dict[tuple[str, str], float],
) -> None

Replace the per-(actor, queue) oldest-pending-age cache with fresh data.

Source code in src/taskq/obs/_otel.py

def update_actor_oldest_pending_age_cache(data: dict[tuple[str, str], float]) -> None:
    """Replace the per-(actor, queue) oldest-pending-age cache with fresh data."""
    global _actor_oldest_pending_age_cache
    _actor_oldest_pending_age_cache = dict(data)

update_actor_oldest_running_age_cache

update_actor_oldest_running_age_cache(
    data: Mapping[str, float],
) -> None

Replace the per-actor oldest-running-age cache with fresh data.

Fed by the backlog sampler from the same grouped read as :func:update_jobs_running_cache, so count and age describe one moment. An attempt older than the actor's normal runtime while taskq.jobs.timeouts stays flat is an actor with no start_to_close: nothing will end the attempt, and no other gauge can show it (running_lease_expired reads 0 while the heartbeat keeps renewing). An actor with nothing running vanishes from the series.

Source code in src/taskq/obs/_otel.py

def update_actor_oldest_running_age_cache(data: Mapping[str, float]) -> None:
    """Replace the per-actor oldest-running-age cache with fresh data.

    Fed by the backlog sampler from the same grouped read as
    :func:`update_jobs_running_cache`, so count and age describe one
    moment. An attempt older than the actor's normal runtime while
    ``taskq.jobs.timeouts`` stays flat is an actor with no
    ``start_to_close``: nothing will end the attempt, and no other gauge
    can show it (``running_lease_expired`` reads 0 while the heartbeat
    keeps renewing). An actor with nothing running vanishes from the
    series.
    """
    global _actor_oldest_running_age_cache
    _actor_oldest_running_age_cache = dict(data)

update_cancel_pending_cache

update_cancel_pending_cache(count: int | None) -> None

Record the count of non-terminal jobs with a cancel in flight.

cancel_phase > 0 over the non-terminal population, sampled by the leader's queue-depth loop. This is the DURABLE cancel surface a scrape can see: a CLI cancel writes cancel_requested_at to the row and exits, the issuing process's own counter dies with it, and this gauge is what shows the protocol still has work in flight. A value pinned above zero across ticks is a cancel the grace ladder is not finishing (read it beside taskq.jobs.cancels_actored_total, which says the worker saw it).

None (the demotion clear) yields no data point: the series goes stale and the new leader's is the one answering, the same empty-not-zero discipline the leader-lease TTL gauge follows — a demoted process exporting 0 would be an active claim that no cancel is pending, silencing the gauge exactly at a failover.

Source code in src/taskq/obs/_otel.py

def update_cancel_pending_cache(count: int | None) -> None:
    """Record the count of non-terminal jobs with a cancel in flight.

    ``cancel_phase > 0`` over the non-terminal population, sampled by the
    leader's queue-depth loop. This is the DURABLE cancel surface a
    scrape can see: a CLI cancel writes cancel_requested_at to the row
    and exits, the issuing process's own counter dies with it, and this
    gauge is what shows the protocol still has work in flight. A value
    pinned above zero across ticks is a cancel the grace ladder is not
    finishing (read it beside taskq.jobs.cancels_actored_total, which
    says the worker saw it).

    ``None`` (the demotion clear) yields no data point: the series goes
    stale and the new leader's is the one answering, the same
    empty-not-zero discipline the leader-lease TTL gauge follows — a
    demoted process exporting 0 would be an active claim that no cancel
    is pending, silencing the gauge exactly at a failover.
    """
    global _cancel_pending_count
    _cancel_pending_count = count

update_cron_slots_behind

update_cron_slots_behind(data: Mapping[str, int]) -> None

Merge one cron tick's observed per-actor skip depths into the gauge.

Called from the tick's commit-gated emission with every actor that had a committed fire or suppression that tick -- clean fires publish 0, so an actor that caught up decays its series on its next observation. MERGE, not replace: the tick observes only DUE schedules (a sample, never the whole table), so unobserved actors keep their last known depth rather than flapping in and out of the series set. The actor key is admitted through the same bounded label as the skip counter (schedule rows accept any string at creation time).

Source code in src/taskq/obs/_otel.py

def update_cron_slots_behind(data: Mapping[str, int]) -> None:
    """Merge one cron tick's observed per-actor skip depths into the gauge.

    Called from the tick's commit-gated emission with every actor that had
    a committed fire or suppression that tick -- clean fires publish 0, so
    an actor that caught up decays its series on its next observation.
    MERGE, not replace: the tick observes only DUE schedules (a sample,
    never the whole table), so unobserved actors keep their last known
    depth rather than flapping in and out of the series set. The actor
    key is admitted through the same bounded label as the skip counter
    (schedule rows accept any string at creation time).
    """
    global _cron_slots_behind_cache
    merged = {**_cron_slots_behind_cache}
    for actor, depth in data.items():
        merged[_bounded_cron_actor(actor)] = depth
    _cron_slots_behind_cache = merged

update_disabled_schedules_count

update_disabled_schedules_count(count: int) -> None

Update the module-level disabled-schedules count.

Called by the leader's schedule management code when schedules are disabled or re-enabled.

Source code in src/taskq/obs/_otel.py

def update_disabled_schedules_count(count: int) -> None:
    """Update the module-level disabled-schedules count.

    Called by the leader's schedule management code when schedules are
    disabled or re-enabled.
    """
    global _disabled_schedules_count
    _disabled_schedules_count = count

update_heartbeat_consecutive_failures

update_heartbeat_consecutive_failures(
    worker_id: str, count: int
) -> None

Update the module-level heartbeat consecutive-failures value.

Called by the heartbeat loop after each tick so the synchronous gauge callback can read the latest value on scrape.

Source code in src/taskq/obs/_otel.py

def update_heartbeat_consecutive_failures(worker_id: str, count: int) -> None:
    """Update the module-level heartbeat consecutive-failures value.

    Called by the heartbeat loop after each tick so the synchronous
    gauge callback can read the latest value on scrape.
    """
    global _heartbeat_consecutive_failures_count
    del worker_id  # Why: not a dimension -- see the cardinality note above.
    _heartbeat_consecutive_failures_count = count

update_jobs_by_status_cache

update_jobs_by_status_cache(data: dict[str, int]) -> None

Replace the per-status job-count cache with fresh data.

Called by the backlog sampler. Replaces the merged pending+scheduled queue-depth view: a growing scheduled count next to a flat pending count is the promotion-stall signature, which a single summed number cannot express.

Source code in src/taskq/obs/_otel.py

def update_jobs_by_status_cache(data: dict[str, int]) -> None:
    """Replace the per-status job-count cache with fresh data.

    Called by the backlog sampler. Replaces the merged pending+scheduled
    queue-depth view: a growing `scheduled` count next to a flat `pending`
    count is the promotion-stall signature, which a single summed number
    cannot express.
    """
    global _jobs_by_status_cache
    _jobs_by_status_cache = dict(data)

update_jobs_retry_headroom_cache

update_jobs_retry_headroom_cache(
    data: Mapping[str, int],
) -> None

Replace the per-actor minimum-retry-headroom cache with fresh data.

Fed by the backlog sampler from the same grouped read as :func:update_jobs_retrying_cache (one statement, one moment).

Source code in src/taskq/obs/_otel.py

def update_jobs_retry_headroom_cache(data: Mapping[str, int]) -> None:
    """Replace the per-actor minimum-retry-headroom cache with fresh data.

    Fed by the backlog sampler from the same grouped read as
    :func:`update_jobs_retrying_cache` (one statement, one moment).
    """
    global _jobs_retry_headroom_cache
    _jobs_retry_headroom_cache = dict(data)

update_jobs_retrying_cache

update_jobs_retrying_cache(data: Mapping[str, int]) -> None

Replace the per-actor retrying-count cache with fresh data.

Fed by the backlog sampler from the same grouped read as :func:update_jobs_retry_headroom_cache, so count and headroom describe one moment. An actor with no non-terminal row past its first attempt vanishes from the series rather than freezing.

Source code in src/taskq/obs/_otel.py

def update_jobs_retrying_cache(data: Mapping[str, int]) -> None:
    """Replace the per-actor retrying-count cache with fresh data.

    Fed by the backlog sampler from the same grouped read as
    :func:`update_jobs_retry_headroom_cache`, so count and headroom
    describe one moment. An actor with no non-terminal row past its
    first attempt vanishes from the series rather than freezing.
    """
    global _jobs_retrying_cache
    _jobs_retrying_cache = dict(data)

update_jobs_running_cache

update_jobs_running_cache(data: Mapping[str, int]) -> None

Replace the per-actor running-count cache with fresh data.

Fed by the backlog sampler from one grouped read over the running population. Per actor, not per (actor, queue): the running row's queue label is not what dispatched it (re-pended rows route by the actor's assignment), and the capacity question is per actor , which actors hold the fleet's slots while pending work waits. An actor with no running jobs vanishes from the series rather than freezing at its last count.

Source code in src/taskq/obs/_otel.py

def update_jobs_running_cache(data: Mapping[str, int]) -> None:
    """Replace the per-actor running-count cache with fresh data.

    Fed by the backlog sampler from one grouped read over the running
    population. Per actor, not per (actor, queue): the running row's
    queue label is not what dispatched it (re-pended rows route by the
    actor's assignment), and the capacity question is per actor ,
    which actors hold the fleet's slots while pending work waits. An
    actor with no running jobs vanishes from the series rather than
    freezing at its last count.
    """
    global _jobs_running_cache
    _jobs_running_cache = dict(data)

update_keyed_reclaim_pending

update_keyed_reclaim_pending(depth: int) -> None

Replace the pending-reclaim depth (evicted bucket names waiting for the next drain tick).

A persistently non-zero value alongside a rising drain-failure counter is the visible signature of broken reclamation; the depth is a scalar because bucket names are caller-controlled and must not become label cardinality.

Source code in src/taskq/obs/_otel.py

def update_keyed_reclaim_pending(depth: int) -> None:
    """Replace the pending-reclaim depth (evicted bucket names waiting for
    the next drain tick).

    A persistently non-zero value alongside a rising drain-failure
    counter is the visible signature of broken reclamation; the depth is
    a scalar because bucket names are caller-controlled and must not
    become label cardinality.
    """
    global _keyed_reclaim_pending
    _keyed_reclaim_pending = depth

update_oldest_due_age_cache

update_oldest_due_age_cache(age_seconds: float) -> None

Record the age of the oldest due-but-still-scheduled job.

0.0 when nothing is due. Monotonic growth of this gauge is the single best promotion-stall detector: it moves under failure regardless of queue depth or throughput, where absolute depth thresholds are environment-specific guesses.

Source code in src/taskq/obs/_otel.py

def update_oldest_due_age_cache(age_seconds: float) -> None:
    """Record the age of the oldest due-but-still-scheduled job.

    0.0 when nothing is due. Monotonic growth of this gauge is the single
    best promotion-stall detector: it moves under failure regardless of
    queue depth or throughput, where absolute depth thresholds are
    environment-specific guesses.
    """
    global _oldest_due_age_seconds
    _oldest_due_age_seconds = age_seconds

update_queue_depth_by_status_cache

update_queue_depth_by_status_cache(
    data: Mapping[tuple[str, str], int],
) -> None

Replace the per-(queue, status) depth cache with fresh data.

Fed by the leader's queue-depth loop from the same tick's grouped read (the depth read's own GROUP BY extended with status), so the status split describes the same moment the depth gauge does. Keyed by (queue, status) — the JOINABLE status split: joining it against the other queue-keyed gauges works on queue, and the pending/scheduled split a promotion stall lives in is one label away, without a PromQL join modifier.

Source code in src/taskq/obs/_otel.py

def update_queue_depth_by_status_cache(data: Mapping[tuple[str, str], int]) -> None:
    """Replace the per-(queue, status) depth cache with fresh data.

    Fed by the leader's queue-depth loop from the same tick's grouped
    read (the depth read's own GROUP BY extended with ``status``), so the
    status split describes the same moment the depth gauge does. Keyed by
    ``(queue, status)`` — the JOINABLE status split: joining it against
    the other queue-keyed gauges works on ``queue``, and the
    pending/scheduled split a promotion stall lives in is one label
    away, without a PromQL join modifier.
    """
    global _queue_depth_by_status_cache
    _queue_depth_by_status_cache = dict(data)

update_queue_depth_cache

update_queue_depth_cache(data: dict[str, int]) -> None

Replace the queue-depth cache with fresh data from the leader's PG query.

Called by the background async task in the leader loop every 15s. The synchronous gauge callback reads from this cache.

Source code in src/taskq/obs/_otel.py

def update_queue_depth_cache(data: dict[str, int]) -> None:
    """Replace the queue-depth cache with fresh data from the leader's PG query.

    Called by the background async task in the leader loop every 15s.
    The synchronous gauge callback reads from this cache.
    """
    global _queue_depth_cache
    _queue_depth_cache = dict(data)

update_queue_live_workers_cache

update_queue_live_workers_cache(
    data: dict[str, int],
) -> None

Replace the per-queue live-worker cache with fresh data from the leader's query, sampled in the same tick as the queue depth, so the two can be joined on queue without describing different moments.

A worker is live when its last_seen_at is within the liveness window (admin_worker_liveness_seconds); a dead-but-unswept worker row does not count. A queue with pending work and no live worker is the condition TaskQQueueUnserved fires on, and it is invisible to every other gauge: depth alone cannot say whether anyone is consuming.

Source code in src/taskq/obs/_otel.py

def update_queue_live_workers_cache(data: dict[str, int]) -> None:
    """Replace the per-queue live-worker cache with fresh data from the
    leader's query, sampled in the same tick as the queue depth, so the
    two can be joined on ``queue`` without describing different moments.

    A worker is live when its ``last_seen_at`` is within the liveness
    window (``admin_worker_liveness_seconds``); a dead-but-unswept worker
    row does not count. A queue with pending work and no live worker is
    the condition ``TaskQQueueUnserved`` fires on, and it is invisible to
    every other gauge: depth alone cannot say whether anyone is consuming.
    """
    global _queue_live_workers_cache
    _queue_live_workers_cache = dict(data)

update_queue_utilization_cache

update_queue_utilization_cache(
    data: dict[str, float],
) -> None

Replace the per-queue utilization cache with fresh data from the leader's query, sampled in the same tick as the queue depth and the live workers, so the three gauges join on queue without describing different moments.

The ratio is :func:taskq.insights.fetch_queue_imbalance's utilization column — due depth ÷ effective capacity (routed actor capacity x live workers) — exported as a first-class gauge so a scaling operator reads it straight off the scrape instead of recomputing it client-side (the computation needs actor_config, a table the scraper does not have). A queue whose effective capacity is zero reports NO utilization series — the insights row's NULL, the starvation shape the depth + live_workers pair (and taskq.jobs.stranded) already expose; a frozen zero there would read as "idle", the opposite of the truth.

Source code in src/taskq/obs/_otel.py

def update_queue_utilization_cache(data: dict[str, float]) -> None:
    """Replace the per-queue utilization cache with fresh data from the
    leader's query, sampled in the same tick as the queue depth and the
    live workers, so the three gauges join on ``queue`` without
    describing different moments.

    The ratio is :func:`taskq.insights.fetch_queue_imbalance`'s
    ``utilization`` column — due depth ÷ effective capacity (routed actor
    capacity x live workers) — exported as a first-class gauge so a
    scaling operator reads it straight off the scrape instead of
    recomputing it client-side (the computation needs ``actor_config``,
    a table the scraper does not have). A queue whose effective capacity
    is zero reports NO utilization series — the insights row's NULL, the
    starvation shape the depth + live_workers pair (and
    ``taskq.jobs.stranded``) already expose; a frozen zero there would
    read as "idle", the opposite of the truth.
    """
    global _queue_utilization_cache
    _queue_utilization_cache = dict(data)

update_ratelimit_bucket_tokens_cache

update_ratelimit_bucket_tokens_cache(
    data: Mapping[tuple[str, str], float],
) -> None

Replace the rate-limit bucket-tokens cache with fresh peek data.

Fed by the leader's rate-limit sampler every 15s, keyed (bucket, kind); keyed-materialised buckets arrive PRE-AGGREGATED under bucket='_other_' with their own kind, so the cache's shape is the series set and nothing at scrape time can mint a per-key series.

Source code in src/taskq/obs/_otel.py

def update_ratelimit_bucket_tokens_cache(data: Mapping[tuple[str, str], float]) -> None:
    """Replace the rate-limit bucket-tokens cache with fresh peek data.

    Fed by the leader's rate-limit sampler every 15s, keyed
    ``(bucket, kind)``; keyed-materialised buckets arrive PRE-AGGREGATED
    under ``bucket='_other_'`` with their own kind, so the cache's shape
    is the series set and nothing at scrape time can mint a per-key
    series.
    """
    global _ratelimit_bucket_tokens_cache
    _ratelimit_bucket_tokens_cache = dict(data)

update_reservation_slots_cache

update_reservation_slots_cache(
    data: dict[str, int],
) -> None

Replace the reservation-slots cache with fresh data from the leader's PG query.

Called by the background async task in the leader loop every 15s. The synchronous gauge callback reads from this cache.

Source code in src/taskq/obs/_otel.py

def update_reservation_slots_cache(data: dict[str, int]) -> None:
    """Replace the reservation-slots cache with fresh data from the leader's PG query.

    Called by the background async task in the leader loop every 15s.
    The synchronous gauge callback reads from this cache.
    """
    global _reservation_slots_cache
    _reservation_slots_cache = dict(data)

update_running_lease_expired_cache

update_running_lease_expired_cache(count: int) -> None

Record the count of running jobs whose lock lease is past.

Fed by the backlog sampler (one statement beside jobs-by-status and oldest-due-age). The zombie-running shape, work claimed, lease expired, row still 'running', is invisible in jobs.by_status (it counts as a healthy running job) and in the miss counters (a dead worker emits nothing): this gauge is the direct count. A healthy fleet reads 0 (the reclaim sweep drains expired leases within a tick or two of expiry), so a SUSTAINED non-zero reading means reclaim is not draining. Rows with a cancel in flight (cancel_phase != 0) are carved out: the reclaim sweep deliberately waits out the cancel grace ladder for them, so their lease expiring mid-cancel is the protocol working, not a zombie: a cancel that never completes pages elsewhere (TaskQAbandonedJobs when its worker is alive to escalate through the phases, TaskQHeartbeatMisses when it died mid-cancel; reclaim honors the row to 'cancelled' either way). No dimensions: the fleet total is the alertable shape, and the per-job truth (locked_by_worker, lock_expires_at) lives on the row and the admin jobs page, not on a label.

Source code in src/taskq/obs/_otel.py

def update_running_lease_expired_cache(count: int) -> None:
    """Record the count of running jobs whose lock lease is past.

    Fed by the backlog sampler (one statement beside jobs-by-status and
    oldest-due-age). The zombie-running shape, work claimed, lease
    expired, row still 'running', is invisible in jobs.by_status (it
    counts as a healthy running job) and in the miss counters (a dead
    worker emits nothing): this gauge is the direct count. A healthy
    fleet reads 0 (the reclaim sweep drains expired leases within a tick
    or two of expiry), so a SUSTAINED non-zero reading means reclaim is
    not draining. Rows with a cancel in flight (cancel_phase != 0) are
    carved out: the reclaim sweep deliberately waits out the cancel grace
    ladder for them, so their lease expiring mid-cancel is the protocol
    working, not a zombie: a cancel that never completes pages
    elsewhere (TaskQAbandonedJobs when its worker is alive to escalate
    through the phases, TaskQHeartbeatMisses when it died mid-cancel;
    reclaim honors the row to 'cancelled' either way). No dimensions:
    the fleet total is the alertable shape, and the per-job truth
    (locked_by_worker, lock_expires_at) lives on the row and the admin
    jobs page, not on a label.
    """
    global _running_lease_expired_count
    _running_lease_expired_count = count

update_scheduled_count_cache

update_scheduled_count_cache(count: int) -> None

Record the current count of scheduled-status jobs.

Emitted label-less (unlike taskq.jobs.by_status, which carries a status label for every status) specifically so it joins on identical label sets with taskq.jobs.oldest_due_age_seconds, also label-less , in TaskQScheduledBacklogGrowing. That alert needs both "the oldest due job has waited a long time" AND "the scheduled count is actually rising", not the age of a single straggling job (which climbs monotonically toward its own promotion regardless of how healthily everything behind it drains). A vector and/comparison between two taskq_* series with mismatched label sets is a silent no-op join , valid PromQL that can never produce a result, so this gauge exists to keep the two operands directly comparable without a join modifier.

Source code in src/taskq/obs/_otel.py

def update_scheduled_count_cache(count: int) -> None:
    """Record the current count of `scheduled`-status jobs.

    Emitted label-less (unlike `taskq.jobs.by_status`, which carries a
    `status` label for every status) specifically so it joins on identical
    label sets with `taskq.jobs.oldest_due_age_seconds`, also label-less ,
    in `TaskQScheduledBacklogGrowing`. That alert needs both "the oldest
    due job has waited a long time" AND "the scheduled count is actually
    rising", not the age of a single straggling job (which climbs
    monotonically toward its own promotion regardless of how healthily
    everything behind it drains). A vector `and`/comparison between two
    `taskq_*` series with mismatched label sets is a silent no-op join ,
    valid PromQL that can never produce a result, so this gauge exists
    to keep the two operands directly comparable without a join modifier.
    """
    global _scheduled_count
    _scheduled_count = count

update_scheduled_horizon_cache

update_scheduled_horizon_cache(seconds: float) -> None

Record how far out the furthest scheduled job is armed.

MAX(scheduled_at) - now over the scheduled population, sampled by the backlog detector's tick beside its label-less siblings. 0.0 when nothing is scheduled; NEGATIVE when even the furthest-armed job is already overdue, the promotion-stall shape where every scheduled row has crossed its wake time. Label-less per the twin convention (taskq.jobs.scheduled_count): it joins the other label-less backlog operands under vector and without a join modifier.

Source code in src/taskq/obs/_otel.py

def update_scheduled_horizon_cache(seconds: float) -> None:
    """Record how far out the furthest scheduled job is armed.

    ``MAX(scheduled_at) - now`` over the scheduled population, sampled by
    the backlog detector's tick beside its label-less siblings. 0.0 when
    nothing is scheduled; NEGATIVE when even the furthest-armed job is
    already overdue, the promotion-stall shape where every scheduled row
    has crossed its wake time. Label-less per the twin convention
    (``taskq.jobs.scheduled_count``): it joins the other label-less
    backlog operands under vector ``and`` without a join modifier.
    """
    global _scheduled_horizon_seconds
    _scheduled_horizon_seconds = seconds

update_stranded_jobs_cache

update_stranded_jobs_cache(
    data: Mapping[tuple[str, StrandedReason], int],
) -> None

Replace the stranded-jobs cache with fresh data from the leader's query.

Stranded jobs are pending/scheduled rows that can never dispatch: the actor has no actor_config row (the dispatch CTE derives its candidates from per_actor_capacity, which is FROM actor_config), or the row sits on a queue no LIVE registered worker serves (dispatch probes only its own subscription's queues, and a worker whose last_seen_at has gone stale is not dispatching). Both shapes accumulate invisibly to dispatch and the deadline sweep. Keyed by (actor, reason) so the gauge says which condition held, the two have different remediations (register the actor vs. subscribe a worker to the queue), and a per-actor total made an operator who found the actor_config row present conclude the detector lied.

This gauge exists because a log line alone is invisible in metrics: the detector's WARN fires exactly once per actor per process lifetime -- so the condition is invisible in metrics and its only trace is a single WARN at onset, which is the moment nobody is looking. An empty mapping clears the gauge, so recovery is visible too.

Source code in src/taskq/obs/_otel.py

def update_stranded_jobs_cache(data: Mapping[tuple[str, StrandedReason], int]) -> None:
    """Replace the stranded-jobs cache with fresh data from the leader's query.

    Stranded jobs are pending/scheduled rows that can never dispatch: the
    actor has no `actor_config` row (the dispatch CTE derives its
    candidates from `per_actor_capacity`, which is `FROM actor_config`),
    or the row sits on a queue no LIVE registered worker serves (dispatch
    probes only its own subscription's queues, and a worker whose
    last_seen_at has gone stale is not dispatching). Both shapes
    accumulate invisibly to dispatch and the deadline sweep. Keyed by
    ``(actor, reason)`` so the gauge says which condition held, the two
    have different remediations (register the actor vs. subscribe a
    worker to the queue), and a per-actor total made an operator who
    found the actor_config row present conclude the detector lied.

    This gauge exists because a log line alone is invisible in metrics: the
    detector's WARN fires
    exactly once per actor per process lifetime -- so the
    condition is invisible in metrics and its only trace is a single WARN at
    onset, which is the moment nobody is looking. An empty mapping clears the
    gauge, so recovery is visible too.
    """
    global _stranded_jobs_cache
    _stranded_jobs_cache = dict(data)

update_sweep_batch_size_cache

update_sweep_batch_size_cache(
    sweep_name: str, batch_size: int
) -> None

Replace the recorded batch size for sweep_name (cache-push gauge).

Rebind, never write in place, same cross-thread reader discipline as :func:record_sweep_success.

Source code in src/taskq/obs/_otel.py

def update_sweep_batch_size_cache(sweep_name: str, batch_size: int) -> None:
    """Replace the recorded batch size for *sweep_name* (cache-push gauge).

    Rebind, never write in place, same cross-thread reader discipline as
    :func:`record_sweep_success`.
    """
    global _sweep_batch_size_cache
    _sweep_batch_size_cache = {**_sweep_batch_size_cache, sweep_name: batch_size}

record_exception_safe

record_exception_safe(
    span: Span, exc: BaseException
) -> None

Record exc on span without leaking row values or credentials.

The event-only form of :func:record_exception_text, for a call site that sets the span status itself.

Source code in src/taskq/obs/_redact_exc.py

def record_exception_safe(span: "Span", exc: BaseException) -> None:
    """Record *exc* on *span* without leaking row values or credentials.

    The event-only form of :func:`record_exception_text`, for a call site that
    sets the span status itself.
    """
    add_exception_event(span, render_exception(exc))

record_exception_text

record_exception_text(
    span: Span, text: ExceptionText
) -> None

Mark span failed by text: ERROR status described by the message, plus the event.

Source code in src/taskq/obs/_redact_exc.py

def record_exception_text(span: "Span", text: ExceptionText) -> None:
    """Mark *span* failed by *text*: ERROR status described by the message, plus the event."""
    span.set_status(StatusCode.ERROR, text.message)
    add_exception_event(span, text)

render_exception

render_exception(exc: BaseException) -> ExceptionText

Render and scrub exc once, for every sink that reports it.

A traceback's final line is the exception repr, so the same DETAIL text reappears there if it is not stripped. Chained causes are included, so each of their messages needs the same treatment -- hence scrubbing the rendered string rather than only the head exception.

Caveat: a traceback also quotes the SOURCE LINE of each frame. Those come from TaskQ's own source, not from data, so they carry no row values -- but a secret written as a literal in application code would appear. Do not put credentials in source.

Source code in src/taskq/obs/_redact_exc.py

def render_exception(exc: BaseException) -> ExceptionText:
    """Render and scrub *exc* once, for every sink that reports it.

    A traceback's final line is the exception repr, so the same DETAIL text
    reappears there if it is not stripped. Chained causes are included, so each
    of their messages needs the same treatment -- hence scrubbing the rendered
    string rather than only the head exception.

    Caveat: a traceback also quotes the SOURCE LINE of each
    frame. Those come from TaskQ's own source, not from data, so they carry no
    row values -- but a secret written as a literal in application code would
    appear. Do not put credentials in source.
    """
    raw_stacktrace = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
    return ExceptionText(
        type_name=type(exc).__qualname__,
        raw_stacktrace=raw_stacktrace,
        # Why safe_str: the exception is uncontrolled (an actor raises it);
        # a __str__ that raises must degrade to the constant marker, not
        # convert inside the handler that was classifying it.
        message=ScrubbedText(_bound_message(_scrub_text(safe_str(exc)))),
        stacktrace=ScrubbedText(_scrub_text(raw_stacktrace)),
    )

safe_exception_message

safe_exception_message(exc: BaseException) -> str

Exception text with the Postgres DETAIL dropped and URI creds masked.

The primary Postgres message is kept: it is a static template naming the constraint or relation, which is the part that is actually diagnostic. HINT and CONTEXT are kept for the same reason -- neither carries row values, and both are what an operator reads next.

The str() is guarded: the exception is uncontrolled input (an actor raises it), and an exception whose own __str__ raises would otherwise convert HERE, inside the handler that was classifying it, into an uncaught TypeError that escapes the classification entirely. The fallback mirrors CPython's own traceback rendering for the same condition (""), so sinks stay consistent.

Source code in src/taskq/obs/_redact_exc.py

def safe_exception_message(exc: BaseException) -> str:
    """Exception text with the Postgres DETAIL dropped and URI creds masked.

    The primary Postgres message is kept: it is a static template naming the
    constraint or relation, which is the part that is actually diagnostic.
    ``HINT`` and ``CONTEXT`` are kept for the same reason -- neither carries
    row values, and both are what an operator reads next.

    The ``str()`` is guarded: the exception is uncontrolled input (an actor
    raises it), and an exception whose own ``__str__`` raises would otherwise
    convert HERE, inside the handler that was classifying it, into an uncaught
    ``TypeError`` that escapes the classification entirely. The fallback
    mirrors CPython's own traceback rendering for the same condition
    ("<exception str() failed>"), so sinks stay consistent.
    """
    return _bound_message(_scrub_text(safe_str(exc)))

safe_repr

safe_repr(exc: BaseException) -> str

repr(exc) that cannot raise, for uncontrolled exceptions.

The except-handler logs that render repr(exc) (hook failures, classifier failures) would otherwise have their own swallow converted into an escape by a __repr__ that raises - the same defect shape as :func:safe_str, one level deeper. The fallback keeps the class name because it is the one diagnostic that survives (pinned), but the NAME ACCESS is guarded too: a metaclass can define __name__ as a property that raises, so even the interpolation is uncontrolled input and degrades to <unknown>.

Source code in src/taskq/obs/_redact_exc.py

def safe_repr(exc: BaseException) -> str:
    """``repr(exc)`` that cannot raise, for uncontrolled exceptions.

    The ``except``-handler logs that render ``repr(exc)`` (hook failures,
    classifier failures) would otherwise have their own swallow converted
    into an escape by a ``__repr__`` that raises - the same defect shape as
    :func:`safe_str`, one level deeper. The fallback keeps the class name
    because it is the one diagnostic that survives (pinned), but the NAME
    ACCESS is guarded too: a metaclass can define ``__name__`` as a
    property that raises, so even the interpolation is uncontrolled input
    and degrades to ``<unknown>``.
    """
    try:
        return repr(exc)
    # Why BaseException: same reasoning as :func:`safe_str` directly above -
    # a hostile __repr__ may raise any BaseException subclass, and this
    # helper is a pure string render, never an await point, so a swallowed
    # CancelledError cannot strand anything.
    except BaseException:
        try:
            name = type(exc).__name__
        except BaseException:
            name = "<unknown>"
        return f"<exception repr() failed: {name}>"

safe_str

safe_str(exc: BaseException) -> str

str(exc) that cannot raise, for uncontrolled exceptions.

The one conversion every sink of raw exception text shares: a hostile or broken __str__ (an actor's exception class is the attacker's code) must degrade to a constant marker instead of raising a fresh exception inside a caller's except block and escaping the classification that was running. Unlike :func:safe_exception_message this is the RAW text (no scrub, no bound) - call sites that derive stored/log fields from it apply their own guards (sanitize_nul_str), and the render-once invariant (render_exception is the single scrub) is not paid twice. The fallback marker mirrors CPython's own traceback rendering for the same condition (measured on 3.14: format_exception survives and prints Boom: <exception str() failed>), so a reader sees one idiom. The catch is a BARE except BaseException for the same reason CPython's traceback._safe_string uses one: a __str__ may raise any BaseException subclass, and except Exception would let it convert inside the guard. Swallowing a CancelledError here is safe: this is a string-rendering helper, never an await point, so there is no suspension it could strand.

Source code in src/taskq/obs/_redact_exc.py

def safe_str(exc: BaseException) -> str:
    """``str(exc)`` that cannot raise, for uncontrolled exceptions.

    The one conversion every sink of raw exception text shares: a hostile
    or broken ``__str__`` (an actor's exception class is the attacker's
    code) must degrade to a constant marker instead of raising a fresh
    exception inside a caller's ``except`` block and escaping the
    classification that was running. Unlike :func:`safe_exception_message`
    this is the RAW text (no scrub, no bound) - call sites that derive
    stored/log fields from it apply their own guards
    (``sanitize_nul_str``), and the render-once invariant
    (``render_exception`` is the single scrub) is not paid twice. The
    fallback marker mirrors CPython's own traceback rendering for the same
    condition (measured on 3.14: ``format_exception`` survives and prints
    ``Boom: <exception str() failed>``), so a reader sees one idiom. The
    catch is a BARE ``except BaseException`` for the same reason CPython's
    ``traceback._safe_string`` uses one: a ``__str__`` may raise any
    ``BaseException`` subclass, and ``except Exception`` would let it
    convert inside the guard. Swallowing a ``CancelledError`` here is safe:
    this is a string-rendering helper, never an await point, so there is no
    suspension it could strand.
    """
    try:
        return str(exc)
    # Why BaseException: CPython's traceback._safe_string idiom - a hostile
    # __str__ may raise any BaseException subclass, and this helper is a
    # pure string render, never an await point, so a swallowed
    # CancelledError cannot strand anything.
    except BaseException:
        return "<exception str() failed>"

set_exception_message_max_chars

set_exception_message_max_chars(limit: int) -> None

Set the module-level bound on scrubbed message text.

Mirrors :func:set_exception_redaction_enabled: a module global set once at worker startup, so the obs layer needs no import of settings.

Source code in src/taskq/obs/_redact_exc.py

def set_exception_message_max_chars(limit: int) -> None:
    """Set the module-level bound on scrubbed message text.

    Mirrors :func:`set_exception_redaction_enabled`: a module global set once
    at worker startup, so the obs layer needs no import of settings.
    """
    global _max_message_chars
    _max_message_chars = limit

set_exception_redaction_enabled

set_exception_redaction_enabled(enabled: bool) -> None

Set the module-level exception-redaction flag.

Mirrors :func:taskq.obs.set_otel_enabled: worker startup calls this once after loading WorkerSettings so every scrub site reads a module global instead of importing settings (which would be a circular import from the modules that depend on obs).

Passing False disables the DETAIL drop on BOTH the span and the log channel -- they share :func:_scrub_text, so the toggle cannot be applied to one and not the other. It does NOT disable the URI credential mask.

Source code in src/taskq/obs/_redact_exc.py

def set_exception_redaction_enabled(enabled: bool) -> None:
    """Set the module-level exception-redaction flag.

    Mirrors :func:`taskq.obs.set_otel_enabled`: worker startup calls this once
    after loading ``WorkerSettings`` so every scrub site reads a module global
    instead of importing ``settings`` (which would be a circular import from
    the modules that depend on ``obs``).

    Passing ``False`` disables the DETAIL drop on BOTH the span and the log
    channel -- they share :func:`_scrub_text`, so the toggle cannot be applied
    to one and not the other. It does NOT disable the URI credential mask.
    """
    global _redaction_enabled
    _redaction_enabled = enabled

bind_job_context

bind_job_context(
    log: BoundLogger,
    *,
    job_id: UUID,
    actor: str,
    queue: str,
    attempt: int,
    identity_key: str | None,
    trace_id: str,
    span_id: str | None = None,
    batch_id: str | None = None,
) -> structlog.stdlib.BoundLogger

Bind job-scope fields to a logger, returning a new immutable BoundLogger.

identity_key, span_id, and batch_id are omitted from the bound dict when None, not set to null or empty string . trace_id is always bound (defaults to "" when no active OTel span per spec). Returns a new BoundLogger; does not mutate the input.

Source code in src/taskq/obs/_structlog.py

def bind_job_context(
    log: structlog.stdlib.BoundLogger,
    *,
    job_id: UUID,
    actor: str,
    queue: str,
    attempt: int,
    identity_key: str | None,
    trace_id: str,
    span_id: str | None = None,
    batch_id: str | None = None,
) -> structlog.stdlib.BoundLogger:
    """Bind job-scope fields to a logger, returning a new immutable BoundLogger.

    ``identity_key``, ``span_id``, and ``batch_id`` are omitted from the bound
    dict when ``None``, not set to null or empty string .  ``trace_id``
    is always bound (defaults to ``""`` when no active OTel span per spec).
    Returns a new ``BoundLogger``; does not mutate the input.
    """
    fields: dict[str, str | int] = {
        "job_id": str(job_id),
        "actor": actor,
        "queue": queue,
        "attempt": attempt,
        "trace_id": trace_id,
    }
    if identity_key is not None:
        fields["identity_key"] = identity_key
    if span_id is not None:
        fields["span_id"] = span_id
    if batch_id is not None:
        fields["batch_id"] = batch_id
    return log.bind(**fields)

get_logger

get_logger(name: str) -> structlog.stdlib.BoundLogger

Return a structlog.stdlib.BoundLogger for the given dotted name.

Replaces direct structlog.get_logger() calls in library code so that pyright strict mode gets an explicit return type (structlog.get_logger returns Any).

Source code in src/taskq/obs/_structlog.py

def get_logger(name: str) -> structlog.stdlib.BoundLogger:
    """Return a ``structlog.stdlib.BoundLogger`` for the given dotted name.

    Replaces direct ``structlog.get_logger()`` calls in library code so that
    pyright strict mode gets an explicit return type (``structlog.get_logger``
    returns ``Any``).
    """
    return structlog.get_logger(name)

log_cancel_phase_change

log_cancel_phase_change(
    log: BoundLogger,
    *,
    from_phase: int,
    to_phase: int,
    **extra: object,
) -> None

Emit an INFO log line with kind="cancel_phase_change".

from_phase and to_phase are the cancel-phase integers before and after the escalation. cancel_observed_at is NOT included, it is per-handler context, not part of the canonical schema.

Source code in src/taskq/obs/_structlog.py

def log_cancel_phase_change(
    log: structlog.stdlib.BoundLogger,
    *,
    from_phase: int,
    to_phase: int,
    **extra: object,
) -> None:
    """Emit an INFO log line with ``kind="cancel_phase_change"``.

    ``from_phase`` and ``to_phase`` are the cancel-phase integers before
    and after the escalation.  ``cancel_observed_at`` is NOT included, it
    is per-handler context, not part of the canonical schema.
    """
    log.info(
        "cancel_phase_change",
        kind="cancel_phase_change",
        from_phase=from_phase,
        to_phase=to_phase,
        **extra,
    )

log_state_change

log_state_change(
    log: BoundLogger,
    *,
    from_state: str,
    to_state: str,
    **extra: object,
) -> None

Emit an INFO log line with kind="state_change".

from_state and to_state are the job-status values before and after the transition. All bound fields from the pre-bound log (which carries job context from :func:bind_job_context) are included automatically. The event name is "state-change" so the log is queryable by both event and kind.

Source code in src/taskq/obs/_structlog.py

def log_state_change(
    log: structlog.stdlib.BoundLogger,
    *,
    from_state: str,
    to_state: str,
    **extra: object,
) -> None:
    """Emit an INFO log line with ``kind="state_change"``.

    ``from_state`` and ``to_state`` are the job-status values before and
    after the transition.  All bound fields from the pre-bound ``log``
    (which carries job context from :func:`bind_job_context`) are included
    automatically.  The event name is ``"state-change"`` so the log is
    queryable by both event and kind.
    """
    log.info("state-change", kind="state_change", from_state=from_state, to_state=to_state, **extra)

redact_payload

redact_payload(payload: object) -> str

Return the first 16 characters of the SHA-256 hex digest of the JSON-serialized payload.

Raw payload content does not appear in the return value. Deterministic for the same input.

Source code in src/taskq/obs/_structlog.py

def redact_payload(payload: object) -> str:
    """Return the first 16 characters of the SHA-256 hex digest of the JSON-serialized payload.

    Raw payload content does not appear in the return value.  Deterministic
    for the same input.
    """
    # Why bytes directly, not dumps_str(...).encode(): the hash consumes
    # bytes, and dumps() already produces them, the str round-trip was a
    # decode+encode pair per redacted log line.
    serialized = dumps(payload)
    return hashlib.sha256(serialized).hexdigest()[:16]

setup_logging

setup_logging(
    *,
    level: str = "INFO",
    log_format: str = "json",
    events_level: str = "info",
) -> None

Configure structlog with the canonical processor chain.

Production (log_format="json"): JSONRenderer via ProcessorFormatter stdlib bridge. Development (log_format="console"): ConsoleRenderer via ProcessorFormatter. Idempotent, guarded by _logging_configured flag. Not called at import time (the pre-setup default chain :func:_install_default_chain is installed at import instead, so an embedding application that never calls this still gets level-filtered logging).

events_level: the TASKQ_LOG_EVENTS_LEVEL value (validated by :func:set_events_level, which raises ValueError on a typo). Applied by the _events_level_filter processor in the shared chain; the default info is today's stream byte-identical.

Source code in src/taskq/obs/_structlog.py

def setup_logging(
    *,
    level: str = "INFO",
    log_format: str = "json",
    events_level: str = "info",
) -> None:
    """Configure structlog with the canonical processor chain.

    Production (``log_format="json"``): ``JSONRenderer`` via
    ``ProcessorFormatter`` stdlib bridge. Development (``log_format="console"``):
    ``ConsoleRenderer`` via ``ProcessorFormatter``. Idempotent, guarded
    by ``_logging_configured`` flag. Not called at import time (the
    pre-setup default chain :func:`_install_default_chain` is installed
    at import instead, so an embedding application that never calls this
    still gets level-filtered logging).

    ``events_level``: the ``TASKQ_LOG_EVENTS_LEVEL`` value (validated by
    :func:`set_events_level`, which raises ``ValueError`` on a typo).
    Applied by the ``_events_level_filter`` processor in the shared
    chain; the default ``info`` is today's stream byte-identical.
    """
    global _logging_configured
    if _logging_configured:
        return

    set_events_level(events_level)

    shared_processors = _shared_processors()

    formatter_processors: list[structlog.types.Processor]
    if log_format == "console":
        renderer: structlog.types.Processor = structlog.dev.ConsoleRenderer()
        formatter_processors = [
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            renderer,
        ]
    else:
        from taskq._json import structlog_serializer

        renderer = structlog.processors.JSONRenderer(serializer=structlog_serializer)
        formatter_processors = [
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            # Still needed for FOREIGN records: ``ProcessorFormatter`` lifts
            # their ``record.exc_info`` onto the event dict here, after the
            # shared chain has run, and orjson drops the whole line on a raw
            # tuple. Idempotent for TaskQ's own records, ``exc_info`` is
            # already gone by then.
            _safe_processor_wrapper(_render_exc_info_safe),
            renderer,
        ]

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        wrapper_class=_ExcInfoSafeBoundLogger,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

    formatter = structlog.stdlib.ProcessorFormatter(
        processors=formatter_processors,
        foreign_pre_chain=[
            _safe_processor_wrapper(structlog.processors.TimeStamper(fmt="iso", utc=True)),
            _safe_processor_wrapper(structlog.stdlib.add_log_level),
            _safe_processor_wrapper(structlog.stdlib.ExtraAdder()),
            # After ExtraAdder so foreign records' extras are scrubbed too.
            _safe_processor_wrapper(_scrub_exception_fields),
        ],
    )

    handler = logging.StreamHandler()
    handler.setFormatter(formatter)

    if not any(
        isinstance(h, logging.StreamHandler)
        and isinstance(h.formatter, structlog.stdlib.ProcessorFormatter)
        for h in logging.root.handlers
    ):
        logging.root.addHandler(handler)

    logging.root.setLevel(level)

    _logging_configured = True

invoke_error_reporter async

invoke_error_reporter(
    reporter: ErrorReporter | None,
    job: JobRow,
    exception: BaseException,
    timeout: float = 3.0,
    *,
    log: BoundLogger | None = None,
) -> None

Invoke reporter.report(), swallowing and counting failures.

A None reporter is treated as a no-op (equivalent to :class:NullErrorReporter). Exceptions from report() are caught, logged at WARNING, and counted on the taskq.error_reporter.failures counter, they never propagate to the caller.

This mirrors the defensive pattern of :func:~taskq.retry.invoke_on_retry_exhausted: a user-supplied hook must never crash the worker's terminal-write path.

Source code in src/taskq/obs/error_reporter.py

async def invoke_error_reporter(
    reporter: ErrorReporter | None,
    job: JobRow,
    exception: BaseException,
    timeout: float = 3.0,  # noqa: ASYNC109  Why: parameter name matches the on_retry_exhausted/invoke_on_success convention; asyncio.wait_for requires a timeout value, not asyncio.timeout() context manager
    *,
    log: structlog.stdlib.BoundLogger | None = None,
) -> None:
    """Invoke *reporter*.report(), swallowing and counting failures.

    A ``None`` reporter is treated as a no-op (equivalent to
    :class:`NullErrorReporter`).  Exceptions from ``report()`` are caught,
    logged at WARNING, and counted on the ``taskq.error_reporter.failures``
    counter, they never propagate to the caller.

    This mirrors the defensive pattern of
    :func:`~taskq.retry.invoke_on_retry_exhausted`: a user-supplied hook
    must never crash the worker's terminal-write path.
    """
    if reporter is None:
        return

    logger: structlog.stdlib.BoundLogger = log if log is not None else _log
    reporter_type = type(reporter).__name__

    try:
        await asyncio.wait_for(reporter.report(job, exception), timeout=timeout)
    except TimeoutError:
        logger.warning(
            "error-reporter-timeout",
            job_id=str(job.id),
            actor=job.actor,
            reporter_type=reporter_type,
            timeout_seconds=timeout,
        )
    except Exception as exc:
        logger.warning(
            "error-reporter-failed",
            job_id=str(job.id),
            actor=job.actor,
            reporter_type=reporter_type,
            error=repr(exc),
        )
        record_error_reporter_failure(reporter_type)