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/metrics/logs and lets operators wire any OTLP-compatible backend by configuring environment variables (or by passing an already-configured TracerProvider / MeterProvider).

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",
    "NullErrorReporter",
    "bind_job_context",
    "get_logger",
    "get_meter",
    "get_tracer",
    "invoke_error_reporter",
    "log_cancel_phase_change",
    "log_state_change",
    "record_archived_jobs",
    "record_backpressure_error",
    "record_cancel_requested",
    "record_consumed_message",
    "record_cron_failure",
    "record_deadline_exceeded_swept",
    "record_dispatch_duration",
    "record_election_attempt",
    "record_error_reporter_failure",
    "record_expired_archive_jobs",
    "record_heartbeat_miss",
    "record_lock_expires_in_seconds",
    "record_process_duration",
    "record_progress_publish_failure",
    "record_pruned_jobs",
    "record_published_message",
    "record_ratelimit_refund_failure",
    "redact_payload",
    "safe_start_span",
    "set_otel_enabled",
    "setup_logging",
    "update_disabled_schedules_count",
    "update_heartbeat_consecutive_failures",
    "update_queue_depth_cache",
    "update_reservation_slots_cache",
]

ConsumedOutcome ¶

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

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.

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

get_meter ¶

get_meter() -> Meter

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

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

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

get_tracer ¶

get_tracer() -> Tracer

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

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

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

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_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.

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.
    """
    try:
        _backpressure_errors.add(1, {"actor": actor, "kind": kind})
    except Exception:
        _log.warning("otel-metric-record-failed", instrument_name="taskq.backpressure.errors")

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_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 constrained to the semconv-specified valid set {succeeded, failed, cancelled, abandoned}. The consumer-path AttemptOutcome includes "scheduled" for snooze/retry/reservation-denial; callers must map that to "abandoned" before calling (the consumer released the job back to the queue without completing it).

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 constrained to the semconv-specified valid set
    ``{succeeded, failed, cancelled, abandoned}``.
    The consumer-path ``AttemptOutcome`` includes ``"scheduled"`` for
    snooze/retry/reservation-denial; callers must map that to
    ``"abandoned"`` before calling (the consumer released the job back
    to the queue without completing it).
    """
    if not _otel_enabled:
        return
    _consumed_messages.add(1, {"actor": actor, "queue": queue, "outcome": outcome})

record_cron_failure ¶

record_cron_failure(schedule_id: 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. Respects _otel_enabled — no-op when False.

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

def record_cron_failure(schedule_id: 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.
    Respects ``_otel_enabled`` — no-op when False.
    """
    if not _otel_enabled:
        return
    _cron_consecutive_failures.add(delta, {"schedule_id": schedule_id})

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"
        )

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": queue})

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
    _leader_election_attempts.add(1, {"worker_id": worker_id})
    if not won:
        _leader_election_failures.add(1, {"worker_id": worker_id})

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
    _heartbeat_misses.add(1, {"worker_id": worker_id})

record_lock_expires_in_seconds ¶

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

Record remaining lock TTL on the histogram.

Called in heartbeat.py at each successful heartbeat renewal. 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 remaining lock TTL on the histogram.

    Called in heartbeat.py at each successful heartbeat renewal.
    Respects ``_otel_enabled`` — no-op when False.
    """
    if not _otel_enabled:
        return
    _lock_expires_in_seconds.record(remaining_ttl, {"worker_id": worker_id})

record_process_duration ¶

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

Record job execution duration on the histogram.

Called outside the CONSUMER span body for sampling independence. Respects _otel_enabled — no-op when False. Custom buckets are the operator's responsibility via SDK Views.

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

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

    Called outside the CONSUMER span body for sampling independence.
    Respects ``_otel_enabled`` — no-op when False.
    Custom buckets are the operator's responsibility via SDK Views.
    """
    if not _otel_enabled:
        return
    _process_duration.record(elapsed, {"actor": actor, "queue": queue})

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": queue})

record_ratelimit_refund_failure ¶

record_ratelimit_refund_failure(
    bucket: str, backend: str
) -> None

Bump the ratelimit.refund_failures counter.

Called at the rate-limit refund failure catch site. Respects _otel_enabled — no-op when False.

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

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

    Called at the rate-limit refund failure catch site.
    Respects ``_otel_enabled`` — no-op when False.
    """
    if not _otel_enabled:
        return
    _ratelimit_refund_failures.add(1, {"bucket": bucket, "backend": backend})

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.

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.
    """
    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,
        )
    except Exception:
        _log.warning("otel-span-creation-failed", span_name=name)
        yield trace.NonRecordingSpan(_NOOP_SPAN_CONTEXT)
        return

    with span_cm as span:
        yield span

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

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 cache.

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 cache.

    Called by the heartbeat loop after each tick so the synchronous
    gauge callback can read the latest value on scrape.
    """
    global _heartbeat_consecutive_failures_cache
    _heartbeat_consecutive_failures_cache[worker_id] = count

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_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)

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, UUID | str | int] = {
        "job_id": 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.
    """
    serialized = dumps_str(payload).encode()
    return hashlib.sha256(serialized).hexdigest()[:16]

setup_logging ¶

setup_logging(
    *, level: str = "INFO", log_format: str = "json"
) -> 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 .

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

def setup_logging(
    *,
    level: str = "INFO",
    log_format: str = "json",
) -> 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 .
    """
    global _logging_configured
    if _logging_configured:
        return

    shared_processors: list[structlog.types.Processor] = [
        _safe_processor_wrapper(structlog.contextvars.merge_contextvars),
        _safe_processor_wrapper(structlog.stdlib.add_log_level),
        _safe_processor_wrapper(structlog.stdlib.add_logger_name),
        _safe_processor_wrapper(structlog.processors.StackInfoRenderer()),
        _safe_processor_wrapper(structlog.processors.TimeStamper(fmt="iso", utc=True)),
        _safe_processor_wrapper(_otel_span_processor),
        _safe_processor_wrapper(structlog.processors.EventRenamer("event")),
    ]

    if log_format == "console":
        renderer: structlog.types.Processor = structlog.dev.ConsoleRenderer()
    else:
        from taskq._json import structlog_serializer

        renderer = structlog.processors.JSONRenderer(serializer=structlog_serializer)

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

    formatter = structlog.stdlib.ProcessorFormatter(
        processors=[
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            renderer,
        ],
        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()),
        ],
    )

    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=job.id,
            actor=job.actor,
            reporter_type=reporter_type,
            timeout_seconds=timeout,
        )
    except Exception as exc:
        logger.warning(
            "error-reporter-failed",
            job_id=job.id,
            actor=job.actor,
            reporter_type=reporter_type,
            error=repr(exc),
        )
        record_error_reporter_failure(reporter_type)

Observability¶

obs ¶

INSTRUMENTATION_NAME module-attribute ¶

__all__ module-attribute ¶

ConsumedOutcome ¶

ErrorReporterType ¶

ErrorReporter ¶

report async ¶

NullErrorReporter ¶

report async ¶

get_meter ¶

get_tracer ¶

record_archived_jobs ¶

record_backpressure_error ¶

record_cancel_requested ¶

record_consumed_message ¶

record_cron_failure ¶

record_deadline_exceeded_swept ¶

record_dispatch_duration ¶

record_election_attempt ¶

record_error_reporter_failure ¶

record_expired_archive_jobs ¶

record_heartbeat_miss ¶

record_lock_expires_in_seconds ¶

record_process_duration ¶

record_progress_publish_failure ¶

record_pruned_jobs ¶

record_published_message ¶

record_ratelimit_refund_failure ¶

safe_start_span ¶

set_otel_enabled ¶

update_disabled_schedules_count ¶

update_heartbeat_consecutive_failures ¶

update_queue_depth_cache ¶

update_reservation_slots_cache ¶

bind_job_context ¶

get_logger ¶

log_cancel_phase_change ¶

log_state_change ¶

redact_payload ¶

setup_logging ¶

invoke_error_reporter async ¶

INSTRUMENTATION_NAME `module-attribute` ¶

all `module-attribute` ¶

report `async` ¶

report `async` ¶

invoke_error_reporter `async` ¶