diff --git a/AGENTS.md b/AGENTS.md index 93cb54e1f..be25dc751 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -174,3 +174,28 @@ add CODEOWNERS-based merge gates until multiple independent maintainers exist. - Maintain 100% production statement, branch, and public-docstring coverage with deterministic unit, concurrency, migration, rollback, integration, and documentation tests. + +## OpenTelemetry checkpoint signals + +- Keep `OpenTelemetryCheckpointStore` opt-in and dependency-injected. The package + must not configure a global tracer provider, meter provider, sampler, + processor, exporter, collector, or host resource. +- Emit only fixed operation names and finite low-cardinality `error.type` + classifications. Use `record_exception=False` and + `set_status_on_exception=False` because durable checkpoint exceptions may + retain protected structured details. +- Failed checkpoint spans must explicitly set the host OpenTelemetry API's + `StatusCode.ERROR` without a description when that optional API is available; + successful spans must leave status Unset. Status resolution or mutation is + best-effort observer work and must never alter application behavior. +- Never add tenant, consumer, batch, endpoint, file, digest, cursor, DSN, + provider payload, exception message, or dynamic exception-class values to + package-owned spans or metrics. +- Treat tracing, metric, status, optional status-code resolution, and clock + failures as best-effort observer failures. They must never alter checkpoint + return values, exception identity, compare-and-swap behavior, transaction + ownership, commit, or rollback. +- Keep counter and seconds-based duration histogram behavior deterministic and + maintain 100% production statement, branch, and public-docstring coverage for + success, conflict, validation, internal-error, Error-status, Unset-status, + caller-transaction, confidentiality, and observer-failure paths. diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f265ff06e..9053969e5 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -216,6 +216,48 @@ descriptive snake_case, RLS is enabled and forced, and the rollback refuses to drop a non-empty table. The stored digest remains prefix evidence only; durable storage does not add provider authentication or full-stream immutability. +## Checkpoint OpenTelemetry observability boundary + +`OpenTelemetryCheckpointStore` is an opt-in wrapper around a durable checkpoint +store. It receives a host-owned tracer and meter through dependency injection and +does not configure global providers, processors, samplers, exporters, +collectors, or resources. The wrapped store remains independently usable without +OpenTelemetry and may be the package-owned PostgreSQL store or a compatible +host-owned implementation. + +Package-owned spans and metrics use only fixed operation, transaction-owner, +outcome, and finite error-classification attributes. They never contain tenant, +consumer, batch, endpoint, file, digest, cursor, and DSN values, provider +payloads, exception messages, or dynamic exception class names. The package +operation span is deliberately storage-agnostic and does not emit +`db.system.name` or claim OpenTelemetry database-client semantics. Actual +database-client spans and database-system attributes belong to the host or +database instrumentation at the client boundary where the storage technology is +known. + +Automatic exception recording and status-on-exception are disabled because a +checkpoint exception may retain protected structured state even when its public +message is bounded. Instead, failed checkpoint spans explicitly set the host +OpenTelemetry API's `StatusCode.ERROR` without a description when the optional +API is available, while successful checkpoint spans leave status Unset. This +preserves standard failure discoverability without exposing exception text in +status descriptions. + +The operation counter records completed loads and saves. The duration histogram +uses seconds and a monotonic clock, clamping backward or unavailable clock +evidence to zero. Failures use only `checkpoint_conflict`, `validation_error`, or +`internal_error`; success omits `error.type`. Tracer, meter, span, export-surface, +optional status-code resolution, status mutation, and clock failures are +contained as observer failures. The exact checkpoint return value or application +exception remains authoritative, so best-effort telemetry cannot change +checkpoint operation semantics, compare-and-swap, transaction ownership, commit, +or rollback. + +Caller-owned transaction spans cover the package call only and do not claim that +the surrounding transaction later committed. The host owns telemetry retention, +access control, alerting, collector availability, database-client +instrumentation, and any correlation outside this confidential package boundary. + ## Modular interoperability CWL hosts such as `contextual-orchestrator` and `naruon` supply tenant context @@ -271,6 +313,11 @@ context-managed early close, and SHA-256 framing sensitivity. Durable-store test cover strict consumer identity, caller-owned transaction behavior, idempotent repeat, exact compare-and-swap, stale and regressive writers, equal and unequal first-writer races, disappearing conflict rows, forced-RLS migration text, -fail-closed rollback, documentation, and live PostgreSQL persistence. Final merge -evidence must be regenerated against the integrated base; successful stacked-base -runs are not reusable release evidence. +fail-closed rollback, documentation, and live PostgreSQL persistence. Checkpoint +telemetry tests additionally prove exact delegation, fixed low-cardinality signal +attributes, storage-agnostic operation spans, seconds-based nonnegative duration, +confidential failure classification, explicit Error status without descriptions, +Unset success status, disabled exception recording, and preservation of +application results and exception identity during ordinary tracer, meter, span, +status, and clock failures. Final merge evidence must be regenerated against the +integrated base; successful stacked-base runs are not reusable release evidence. diff --git a/CHANGELOG.md b/CHANGELOG.md index f61282a19..929c1277c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Optional OpenTelemetry-compatible checkpoint spans and metrics through + `OpenTelemetryCheckpointStore`, with dependency-injected tracer and meter, + fixed low-cardinality operation and transaction-owner labels, a seconds-based + monotonic duration histogram, and the finite failure vocabulary + `checkpoint_conflict, validation_error, and internal_error`. Package-owned + signals omit tenant, consumer, batch, endpoint, file, digest, cursor, DSN, + exception-message, and provider-payload data; automatic exception recording + and automatic status-on-exception are disabled. Failed checkpoint spans set + the host OpenTelemetry API's `StatusCode.ERROR` without a description when + available, while successful checkpoint spans leave status Unset. Optional + status resolution, explicit status mutation, ordinary telemetry, and clock + failures cannot alter checkpoint results, exception identity, + compare-and-swap, commit, rollback, or caller-owned transaction behavior. No + SDK/exporter dependency, migration, version bump, or release is included. - Optional durable result-checkpoint store through `PostgresBatchResultCheckpointStore` and `llm_result_stream_checkpoints`, with tenant-qualified consumer identity, diff --git a/CLAUDE.md b/CLAUDE.md index 6e66882af..025fab372 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -158,3 +158,25 @@ while any acknowledgement evidence remains. - Maintain deterministic concurrency, migration, rollback, live-PostgreSQL, documentation, and 100% production coverage tests. + +## OpenTelemetry checkpoint signals + +- Keep `OpenTelemetryCheckpointStore` optional, dependency-injected, and free of + package-owned global OpenTelemetry configuration. +- Emit fixed operation, transaction-owner, and outcome labels only. Use a finite + low-cardinality error vocabulary and disable automatic exception recording. +- Use `record_exception=False` and `set_status_on_exception=False` for every + package-owned checkpoint span. On failure, explicitly set the host + OpenTelemetry API's `StatusCode.ERROR` without a description when available; + on success, leave status Unset. +- Never add tenant, consumer, batch, endpoint, file, digest, cursor, or DSN values + to package-owned spans or metrics. Do not add exception messages, dynamic + exception class names, provider payloads, or database errors either. +- The original checkpoint result or exact application exception is authoritative; + tracer, metric, clock, optional status-code resolution, and status-mutation + failures must not mask or replace application results or exceptions. +- Preserve package-owned versus caller-owned transaction labels without changing + commit, rollback, compare-and-swap, or cursor ownership. +- Maintain 100% production statement, branch, and public-docstring coverage for + success, conflict, validation, internal-error, duration, Error-status, + Unset-status, confidentiality, delegation, and telemetry-failure paths. diff --git a/docs/adr/0008-checkpoint-opentelemetry-observability.md b/docs/adr/0008-checkpoint-opentelemetry-observability.md new file mode 100644 index 000000000..861f900ce --- /dev/null +++ b/docs/adr/0008-checkpoint-opentelemetry-observability.md @@ -0,0 +1,36 @@ +# ADR 0008: Checkpoint OpenTelemetry observability + +- Status: Accepted +- Date: 2026-08-07 + +## Context + +The durable result-checkpoint store provides tenant isolation, exact compare-and-swap advancement, and caller-owned transaction methods. Operators nevertheless need standard latency, outcome, and conflict signals. Logging checkpoint identifiers or exception bodies would create a confidentiality and cardinality risk, while requiring an OpenTelemetry SDK or exporter would compromise standalone operation and could let observer failure alter commit or rollback behavior. + +The wrapper may also decorate a compatible host-owned checkpoint store. A package-level checkpoint operation span therefore cannot truthfully assert a database technology or database-client semantic convention: the wrapped implementation may use PostgreSQL, another persistence technology, or no database at all. + +## Decision + +Add the opt-in `OpenTelemetryCheckpointStore` wrapper. It accepts a dependency-injected OpenTelemetry-compatible tracer and meter and delegates the four existing checkpoint operations without changing their arguments, return values, exception identity, transaction ownership, or storage semantics. + +Package-owned spans and metrics never contain resource identifiers. In particular, tenant, consumer, batch, endpoint, file, digest, cursor, and DSN values are excluded. Span names, metric names, operation labels, transaction-owner labels, outcomes, and `error.type` values are fixed finite vocabularies. Automatic exception recording and automatic exception status are disabled because checkpoint exceptions can retain sensitive structured details. + +The counter records completed operations. The histogram unit is seconds and records a nonnegative monotonic duration. Success omits `error.type`; failures use only `checkpoint_conflict`, `validation_error`, or `internal_error`. To follow the OpenTelemetry recording-errors contract without exposing exception text, failed checkpoint spans explicitly set OpenTelemetry status Error without a description, while successful checkpoint spans leave status Unset. The package resolves the host OpenTelemetry API's `StatusCode.ERROR` only when that optional API is available; missing or failing observer support never becomes an application dependency. + +Package operation spans are storage-agnostic: they do not emit `db.system.name` or claim to be OpenTelemetry database-client spans. When a host needs PostgreSQL client telemetry, the host or database instrumentation layer owns the corresponding database semantic-convention attributes and spans. + +Instrumentation is best effort. Tracer start, span entry, attribute mutation, explicit status mutation, span exit, metric creation, metric recording, optional status-code resolution, and clock failures are contained. The original checkpoint result or exact application exception remains authoritative. Exporter, processor, sampler, and provider ownership remains with the host. The package neither configures global OpenTelemetry state nor adds an SDK/exporter dependency. + +## Consequences + +Operators gain interoperable, low-cardinality checkpoint operation signals without exposing durable identities or provider-controlled data. Hosts may attach their own resource attributes outside the package boundary, but must not reinterpret package omission as authorization to add tenant or remote identifiers to high-cardinality telemetry. + +Failed spans are discoverable through standard OpenTelemetry status in addition to the bounded `error.type`, without placing exception messages or status descriptions into package telemetry. Successful operations preserve the default Unset status rather than forcing `Ok`. + +Storage-agnostic operation spans remain truthful when the wrapper decorates a compatible host-owned store. PostgreSQL-specific client instrumentation remains independently composable and can coexist with the package operation span without the package fabricating database semantics. + +Telemetry cannot prove database commit, replication, or downstream business-effect durability beyond the wrapped method boundary. Caller-owned transaction methods measure the package call only; the caller remains responsible for the surrounding transaction outcome. + +## Rollback + +Remove the wrapper from host composition and continue using `PostgresBatchResultCheckpointStore` or the compatible host-owned store directly. No database migration, stored state, public checkpoint schema, dependency, or release version needs rollback. Removing instrumentation must not change durable checkpoint data or transaction behavior. diff --git a/docs/checkpoint-observability.md b/docs/checkpoint-observability.md new file mode 100644 index 000000000..c77cfe7e3 --- /dev/null +++ b/docs/checkpoint-observability.md @@ -0,0 +1,114 @@ +# Durable checkpoint observability + +`OpenTelemetryCheckpointStore` is an optional wrapper for +`PostgresBatchResultCheckpointStore` or a compatible host-owned checkpoint +store. It emits one span, one completed-operation count, and one duration +measurement for each public load or save call. It does not install or configure +OpenTelemetry and does not assume what storage technology a compatible store +uses. + +## Host setup + +Install and configure the OpenTelemetry API, SDK, processors, and exporters in +the embedding service. Then pass the host-owned tracer and meter explicitly: + +```python +from opentelemetry import metrics, trace + +from pg_llm_batch import ( + OpenTelemetryCheckpointStore, + PostgresBatchResultCheckpointStore, +) + +checkpoint_store = PostgresBatchResultCheckpointStore( + "postgresql://application-role@database/operations", + tenant_scope="tenant-a", +) +observed_checkpoints = OpenTelemetryCheckpointStore( + checkpoint_store, + tracer=trace.get_tracer("buyer.application"), + meter=metrics.get_meter("buyer.application"), +) + +current = observed_checkpoints.load( + "result-worker", + "batch-123", + "default", +) +``` + +The wrapper delegates all arguments and returns unchanged. Use +`load_in_transaction()` and `save_in_transaction()` exactly as on the underlying +store when checkpoint advancement must share a caller-owned PostgreSQL +transaction with local business effects. The wrapper does not commit or roll +back that cursor. + +## Signal contract + +Spans: + +- `pg_llm_batch.checkpoint.load` +- `pg_llm_batch.checkpoint.save` + +Metrics: + +- `pg_llm_batch.checkpoint.operation.count`, unit `{operation}` +- `pg_llm_batch.checkpoint.operation.duration`, unit `s` + +Fixed attributes: + +- `pg_llm_batch.checkpoint.operation=load|save` +- `pg_llm_batch.checkpoint.transaction_owner=package|caller` +- `pg_llm_batch.checkpoint.outcome=success|conflict|validation_error|error` +- `error.type=checkpoint_conflict|validation_error|internal_error` on failures + +Success omits `error.type`. The duration is measured with a monotonic clock and +is never negative. By contract, failed checkpoint spans explicitly set +OpenTelemetry status Error without a description, while successful checkpoint +spans leave status Unset. This keeps failure discovery interoperable without +copying exception messages into span status descriptions. The wrapper resolves +the host OpenTelemetry API's `StatusCode.ERROR` only when that optional API is +available; if status resolution or mutation fails, the checkpoint result or +exact application exception remains authoritative. + +Package operation spans are deliberately storage-agnostic and do not emit +`db.system.name`. They describe the checkpoint abstraction, not a database client +call. A host that needs PostgreSQL client spans should enable its own database +instrumentation at the actual database-client boundary, where OpenTelemetry +database semantic conventions can be applied truthfully. This preserves the same +wrapper contract for non-PostgreSQL compatible stores. + +## Confidentiality and cardinality + +Package-owned telemetry never contains tenant scope, checkpoint consumer name, +remote batch identifier, endpoint alias, provider file identifier, checkpoint +digest, database cursor, DSN, provider payload, exception message, exception +object, or dynamic exception class name. Do not add those values as metric +attributes in host wrappers; they create confidentiality and unbounded-cardinality +risk. Use access-controlled logs or a separately reviewed audit store when an +operator must reconcile a specific durable identity. + +Automatic exception recording and automatic status-on-exception are disabled. +The wrapper instead applies the fixed Error status explicitly, without a status +description, after classifying a failed operation. It supplies `(None, None, +None)` when closing the span context so the application exception is not handed +to observer code through context-manager arguments. + +## Failure behavior + +Telemetry is best effort. Ordinary tracer, meter, span, exporter-surface, clock, +optional OpenTelemetry status-code resolution, status mutation, and +telemetry-originated cancellation failures are contained. They do not change the +checkpoint result, exact exception object, compare-and-swap decision, +transaction owner, commit, or rollback behavior. Non-cancellation process-control +exceptions remain outside this observer-failure guarantee. + +A caller-owned transaction span measures only the package method call. It does +not prove that the surrounding transaction later committed, replicated, or +completed downstream effects. Observe those boundaries in the embedding service. + +## Rollback + +Remove `OpenTelemetryCheckpointStore` from service composition and call the +underlying checkpoint store directly. No schema, migration, persisted row, +checkpoint digest, or package version needs rollback. \ No newline at end of file diff --git a/docs/doctoring/checkpoint-opentelemetry-observability.md b/docs/doctoring/checkpoint-opentelemetry-observability.md new file mode 100644 index 000000000..40d06c8ed --- /dev/null +++ b/docs/doctoring/checkpoint-opentelemetry-observability.md @@ -0,0 +1,35 @@ +# Checkpoint OpenTelemetry observability doctoring + +## Assurance statement + +`OpenTelemetryCheckpointStore` emits bounded OpenTelemetry-compatible spans, completed-operation counts, and operation-duration histograms around durable checkpoint loads and saves. The package injects no SDK, exporter, global provider, or remote collector. It does not record tenant, consumer, batch, endpoint, file, digest, cursor, DSN, exception message, exception object, or checkpoint payload data. + +The current OpenTelemetry semantic conventions 1.44.0 distinguish generic application-operation telemetry from semantic conventions for database client spans. Database client spans require `db.system.name` when the database system is known, but the checkpoint wrapper is not itself a database-client instrumentation layer and may wrap a compatible host-owned store with another persistence implementation. The package therefore does **not** emit `db.system.name` on its checkpoint operation spans. Database-client semantic attributes remain owned by host or database instrumentation at the actual client boundary. + +Recording errors requires a predictable low-cardinality `error.type` on failures and omission on success. The package narrows that vocabulary to three fixed values only: `checkpoint_conflict`, `validation_error`, and `internal_error`. The same OpenTelemetry recording-errors guidance says failed operations should use span status `Error` while successful operations leave status unset. Accordingly, failed checkpoint spans explicitly set OpenTelemetry status Error without a description, while successful checkpoint spans leave status Unset. + +Automatic span exception recording and automatic status-on-exception remain disabled. Checkpoint exceptions can retain structured consumer or batch details even when their public message is bounded, so package-owned observability records only the finite classification and an explicit Error status with no description. The package resolves `opentelemetry.trace.StatusCode.ERROR` only when the host's optional OpenTelemetry API is importable. Missing status support, status mutation failures, telemetry setup, mutation, and export failures are observer failures and cannot replace the application result or exact exception. + +## Operator boundary + +The host owns OpenTelemetry provider, sampler, processor, exporter, resource, retention, access-control, collector configuration, and database-client instrumentation. Operators should alert on bounded conflict/error ratios, Error-status spans, and duration distributions rather than adding remote resource identifiers as metric attributes. Caller-owned transaction methods describe the package call, not the eventual outer commit; database commit and downstream-effect observability remain host responsibilities. + +A PostgreSQL deployment may combine these storage-agnostic package spans with separately generated PostgreSQL client spans. The package span must not be reclassified as a database client span merely because the package-provided checkpoint store currently uses PostgreSQL; doing so would make the same public wrapper semantically false when a compatible non-PostgreSQL store is injected. + +## Verification + +Deterministic tests prove exact delegation, caller-versus-package transaction labels, fixed span and metric attributes, storage-agnostic operation spans, seconds-based nonnegative durations, finite error classification, explicit Error status on failed operations, Unset status on successful operations, absence of status descriptions and protected identifiers, disabled exception recording, optional-API fail-open observability behavior, and preservation of the original result or exact exception during tracer, meter, span, exporter-surface, status, and clock failures. Production statement, branch, and public-docstring coverage remain required at 100%. + +## References (APA 7) + +OpenTelemetry Authors. (n.d.). *OpenTelemetry semantic conventions 1.44.0*. OpenTelemetry. Retrieved August 7, 2026, from https://opentelemetry.io/docs/specs/semconv/ + +OpenTelemetry Authors. (n.d.). *Semantic conventions for database client spans*. OpenTelemetry. Retrieved August 7, 2026, from https://opentelemetry.io/docs/specs/semconv/db/database-spans/ + +OpenTelemetry Authors. (n.d.). *Recording errors*. OpenTelemetry. Retrieved August 7, 2026, from https://opentelemetry.io/docs/specs/semconv/general/recording-errors/ + +OpenTelemetry Authors. (n.d.). *General error attributes*. OpenTelemetry. Retrieved August 7, 2026, from https://opentelemetry.io/docs/specs/semconv/registry/attributes/error/ + +OpenTelemetry Authors. (n.d.). *Trace API*. OpenTelemetry. Retrieved August 7, 2026, from https://opentelemetry.io/docs/specs/otel/trace/api/ + +These references are recorded in APA 7 form. The normative boundary used here is that failed operations should set span status `Error` and `error.type`, while successful spans leave status unset and omit `error.type`; the implementation additionally constrains `error.type` to a package-owned three-value vocabulary and omits status descriptions to avoid sensitive exception text. diff --git a/pg_llm_batch/__init__.py b/pg_llm_batch/__init__.py index 74385e196..7f57ecaa7 100644 --- a/pg_llm_batch/__init__.py +++ b/pg_llm_batch/__init__.py @@ -9,6 +9,7 @@ StreamingBatchAPIClient -- bounded incremental result records BatchResultCheckpoint -- host-persistable resume evidence PostgresBatchResultCheckpointStore -- tenant-isolated durable checkpoints + OpenTelemetryCheckpointStore -- confidential checkpoint observability DurableBatchAPIClient -- standalone durable lifecycle state TenantDurableBatchAPIClient -- tenant-isolated lifecycle state PostgresConfigStore, SecretStore -- database configuration and secrets @@ -27,6 +28,7 @@ apply_result_checkpoint_schema, validate_checkpoint_consumer_name, ) +from .checkpoint_telemetry import OpenTelemetryCheckpointStore from .config import PostgresConfigStore, SecretStore, get_config_store from .db import ( DEFAULT_TENANT_SCOPE, @@ -65,6 +67,7 @@ "BatchResultCheckpoint", "CheckpointedBatchResultRecord", "PostgresBatchResultCheckpointStore", + "OpenTelemetryCheckpointStore", "CheckpointConflictError", "apply_result_checkpoint_schema", "validate_checkpoint_consumer_name", diff --git a/pg_llm_batch/checkpoint_telemetry.py b/pg_llm_batch/checkpoint_telemetry.py new file mode 100644 index 000000000..459848ea7 --- /dev/null +++ b/pg_llm_batch/checkpoint_telemetry.py @@ -0,0 +1,342 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) ContextualWisdomLab. +"""Best-effort OpenTelemetry-compatible signals for durable checkpoints.""" + +from __future__ import annotations + +import math +import time +from asyncio import CancelledError +from typing import Any, Callable, Optional, TypeVar + +from .checkpoint_store import CheckpointConflictError +from .exceptions import ValidationError +from .result_streaming import BatchResultCheckpoint + +_OPERATION_COUNT_NAME = "pg_llm_batch.checkpoint.operation.count" +_OPERATION_DURATION_NAME = "pg_llm_batch.checkpoint.operation.duration" +_OPERATION_ATTRIBUTE = "pg_llm_batch.checkpoint.operation" +_TRANSACTION_OWNER_ATTRIBUTE = "pg_llm_batch.checkpoint.transaction_owner" +_OUTCOME_ATTRIBUTE = "pg_llm_batch.checkpoint.outcome" +_ERROR_TYPE_ATTRIBUTE = "error.type" +_TELEMETRY_FAILURES = (Exception, CancelledError) + +_ResultT = TypeVar("_ResultT") + + +class _NoOpInstrument: + """Ignore one metric operation after instrumentation becomes unavailable.""" + + def add(self, _value: int, *, attributes: dict[str, str]) -> None: + """Ignore one counter measurement.""" + del attributes + + def record(self, _value: float, *, attributes: dict[str, str]) -> None: + """Ignore one histogram measurement.""" + del attributes + + +class _NoOpSpan: + """Ignore span mutation after tracing becomes unavailable.""" + + def set_attribute(self, _name: str, _value: str) -> None: + """Ignore one span attribute.""" + return None + + +_NO_OP_INSTRUMENT = _NoOpInstrument() +_NO_OP_SPAN = _NoOpSpan() + + +class _SafeSpanScope: + """Contain ordinary tracer failures without altering application behavior.""" + + def __init__( + self, + tracer: Any, + name: str, + attributes: dict[str, str], + ) -> None: + """Store one injected tracer and a bounded initial attribute set.""" + self._tracer = tracer + self._name = name + self._attributes = attributes + self._context: Any = None + self._span: Any = _NO_OP_SPAN + + def __enter__(self) -> Any: + """Start one span or degrade to a no-op span on telemetry failure.""" + try: + context = self._tracer.start_as_current_span( + self._name, + attributes=dict(self._attributes), + record_exception=False, + set_status_on_exception=False, + ) + span = context.__enter__() + except _TELEMETRY_FAILURES: + self._context = None + self._span = _NO_OP_SPAN + else: + self._context = context + self._span = span + return self._span + + def __exit__(self, *_exc: Any) -> bool: + """End one span without handing application exceptions to observers.""" + if self._context is not None: + try: + self._context.__exit__(None, None, None) + except _TELEMETRY_FAILURES: + pass + return False + + +def _create_counter(meter: Any) -> Any: + """Create the bounded operation counter or return a no-op instrument.""" + try: + return meter.create_counter( + _OPERATION_COUNT_NAME, + unit="{operation}", + description="Completed durable checkpoint operations by bounded outcome.", + ) + except _TELEMETRY_FAILURES: + return _NO_OP_INSTRUMENT + + +def _create_histogram(meter: Any) -> Any: + """Create the duration histogram or return a no-op instrument.""" + try: + return meter.create_histogram( + _OPERATION_DURATION_NAME, + unit="s", + description="Duration of durable checkpoint operations.", + ) + except _TELEMETRY_FAILURES: + return _NO_OP_INSTRUMENT + + +def _read_clock(clock: Callable[[], Any]) -> Any: + """Read a monotonic clock without making telemetry a business dependency.""" + try: + return clock() + except _TELEMETRY_FAILURES: + return None + + +def _duration_seconds(start_ns: Any, end_ns: Any) -> float: + """Return a finite nonnegative duration or zero for invalid clock evidence.""" + if start_ns is None or end_ns is None: + return 0.0 + try: + duration = (end_ns - start_ns) / 1_000_000_000 + if not math.isfinite(duration): + return 0.0 + return max(0.0, float(duration)) + except _TELEMETRY_FAILURES: + return 0.0 + + +def _resolve_error_status() -> Any: + """Resolve OpenTelemetry ``StatusCode.ERROR`` without adding a dependency.""" + try: + from opentelemetry.trace import StatusCode + except _TELEMETRY_FAILURES: + return None + return StatusCode.ERROR + + +def _classify_failure(error: BaseException) -> tuple[str, str]: + """Map one failure to a finite low-cardinality telemetry classification.""" + if isinstance(error, CheckpointConflictError): + return "conflict", "checkpoint_conflict" + if isinstance(error, ValidationError): + return "validation_error", "validation_error" + return "error", "internal_error" + + +def _safe_set_attribute(span: Any, name: str, value: str) -> None: + """Set one bounded span attribute without exposing exporter availability.""" + try: + span.set_attribute(name, value) + except _TELEMETRY_FAILURES: + pass + + +def _safe_set_status(span: Any, status: Any) -> None: + """Set one host API status without making tracing an application dependency.""" + if status is None: + return + try: + span.set_status(status) + except _TELEMETRY_FAILURES: + pass + + +def _safe_add(counter: Any, attributes: dict[str, str]) -> None: + """Record one completed operation without making metrics authoritative.""" + try: + counter.add(1, attributes=dict(attributes)) + except _TELEMETRY_FAILURES: + pass + + +def _safe_record( + histogram: Any, + duration_seconds: float, + attributes: dict[str, str], +) -> None: + """Record one duration without making metrics authoritative.""" + try: + histogram.record(duration_seconds, attributes=dict(attributes)) + except _TELEMETRY_FAILURES: + pass + + +class OpenTelemetryCheckpointStore: + """Wrap a durable checkpoint store with confidential best-effort telemetry. + + The injected ``tracer`` and ``meter`` use the stable OpenTelemetry API call + shapes, but this package does not require or configure an SDK or exporter. + Hosts retain ownership of providers, sampling, export, and resource metadata. + Checkpoint tenant, consumer, batch, endpoint, file, digest, cursor, and DSN + values are never added to package-owned telemetry. Package operation spans + are storage-agnostic and do not claim database-client semantic attributes; + database instrumentation remains the embedding host's responsibility. + """ + + def __init__( + self, + store: Any, + *, + tracer: Any, + meter: Any, + monotonic_ns: Callable[[], Any] = time.monotonic_ns, + ) -> None: + """Bind one store and host-owned OpenTelemetry-compatible instruments.""" + self._store = store + self._tracer = tracer + self._counter = _create_counter(meter) + self._histogram = _create_histogram(meter) + self._monotonic_ns = monotonic_ns + self._error_status = _resolve_error_status() + + def _execute( + self, + operation: str, + transaction_owner: str, + callback: Callable[[], _ResultT], + ) -> _ResultT: + """Run one checkpoint operation while containing telemetry failures.""" + span_attributes = { + _OPERATION_ATTRIBUTE: operation, + _TRANSACTION_OWNER_ATTRIBUTE: transaction_owner, + } + metric_attributes = { + _OPERATION_ATTRIBUTE: operation, + _TRANSACTION_OWNER_ATTRIBUTE: transaction_owner, + } + start_ns = _read_clock(self._monotonic_ns) + with _SafeSpanScope( + self._tracer, + f"pg_llm_batch.checkpoint.{operation}", + span_attributes, + ) as span: + try: + result = callback() + except BaseException as error: + outcome, error_type = _classify_failure(error) + metric_attributes[_OUTCOME_ATTRIBUTE] = outcome + metric_attributes[_ERROR_TYPE_ATTRIBUTE] = error_type + _safe_set_attribute(span, _OUTCOME_ATTRIBUTE, outcome) + _safe_set_attribute(span, _ERROR_TYPE_ATTRIBUTE, error_type) + _safe_set_status(span, self._error_status) + _safe_add(self._counter, metric_attributes) + _safe_record( + self._histogram, + _duration_seconds( + start_ns, + _read_clock(self._monotonic_ns), + ), + metric_attributes, + ) + raise + metric_attributes[_OUTCOME_ATTRIBUTE] = "success" + _safe_set_attribute(span, _OUTCOME_ATTRIBUTE, "success") + _safe_add(self._counter, metric_attributes) + _safe_record( + self._histogram, + _duration_seconds(start_ns, _read_clock(self._monotonic_ns)), + metric_attributes, + ) + return result + + def load( + self, + consumer_name: str, + batch_id: str, + endpoint_alias: str, + ) -> Optional[BatchResultCheckpoint]: + """Load one checkpoint and emit package-owned operation telemetry.""" + return self._execute( + "load", + "package", + lambda: self._store.load(consumer_name, batch_id, endpoint_alias), + ) + + def load_in_transaction( + self, + cursor: Any, + consumer_name: str, + batch_id: str, + endpoint_alias: str, + ) -> Optional[BatchResultCheckpoint]: + """Load through a caller transaction without changing its ownership.""" + return self._execute( + "load", + "caller", + lambda: self._store.load_in_transaction( + cursor, + consumer_name, + batch_id, + endpoint_alias, + ), + ) + + def save( + self, + consumer_name: str, + checkpoint: BatchResultCheckpoint, + *, + expected_previous: Optional[BatchResultCheckpoint] = None, + ) -> BatchResultCheckpoint: + """Save one checkpoint and emit package-owned operation telemetry.""" + return self._execute( + "save", + "package", + lambda: self._store.save( + consumer_name, + checkpoint, + expected_previous=expected_previous, + ), + ) + + def save_in_transaction( + self, + cursor: Any, + consumer_name: str, + checkpoint: BatchResultCheckpoint, + *, + expected_previous: Optional[BatchResultCheckpoint] = None, + ) -> BatchResultCheckpoint: + """Save through a caller transaction without changing its ownership.""" + return self._execute( + "save", + "caller", + lambda: self._store.save_in_transaction( + cursor, + consumer_name, + checkpoint, + expected_previous=expected_previous, + ), + ) diff --git a/tests/test_checkpoint_telemetry.py b/tests/test_checkpoint_telemetry.py new file mode 100644 index 000000000..8b6d6a933 --- /dev/null +++ b/tests/test_checkpoint_telemetry.py @@ -0,0 +1,528 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Tests for low-cardinality OpenTelemetry checkpoint instrumentation.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable + +import pytest + +from pg_llm_batch.checkpoint_store import CheckpointConflictError +from pg_llm_batch.checkpoint_telemetry import OpenTelemetryCheckpointStore +from pg_llm_batch.exceptions import ValidationError +from pg_llm_batch.result_streaming import BatchResultCheckpoint + + +@dataclass(frozen=True) +class RecordedMeasurement: + """Capture one deterministic metric call.""" + + value: float + attributes: dict[str, str] + + +class FakeInstrument: + """Collect metric calls or raise one configured telemetry failure.""" + + def __init__(self, *, failure: Exception | None = None) -> None: + self.failure = failure + self.measurements: list[RecordedMeasurement] = [] + + def add(self, value: int, attributes: dict[str, str]) -> None: + """Record one counter measurement.""" + if self.failure is not None: + raise self.failure + self.measurements.append(RecordedMeasurement(float(value), dict(attributes))) + + def record(self, value: float, attributes: dict[str, str]) -> None: + """Record one histogram measurement.""" + if self.failure is not None: + raise self.failure + self.measurements.append(RecordedMeasurement(float(value), dict(attributes))) + + +class FakeMeter: + """Create deterministic counter and histogram instruments.""" + + def __init__( + self, + *, + create_failure: Exception | None = None, + measurement_failure: Exception | None = None, + ) -> None: + self.create_failure = create_failure + self.counter = FakeInstrument(failure=measurement_failure) + self.histogram = FakeInstrument(failure=measurement_failure) + self.created: list[tuple[str, str, str]] = [] + + def create_counter(self, name: str, *, unit: str, description: str) -> FakeInstrument: + """Create the operation counter.""" + if self.create_failure is not None: + raise self.create_failure + self.created.append((name, unit, description)) + return self.counter + + def create_histogram( + self, + name: str, + *, + unit: str, + description: str, + ) -> FakeInstrument: + """Create the operation-duration histogram.""" + if self.create_failure is not None: + raise self.create_failure + self.created.append((name, unit, description)) + return self.histogram + + +class FakeSpan: + """Collect bounded span attributes without recording exceptions.""" + + def __init__(self, *, set_failure: Exception | None = None) -> None: + self.set_failure = set_failure + self.attributes: dict[str, str] = {} + self.recorded_exceptions: list[BaseException] = [] + + def set_attribute(self, name: str, value: str) -> None: + """Record one span attribute.""" + if self.set_failure is not None: + raise self.set_failure + self.attributes[name] = value + + def record_exception(self, error: BaseException) -> None: + """Expose accidental exception recording to confidentiality assertions.""" + self.recorded_exceptions.append(error) + + +class FakeSpanContext: + """Enter and exit one fake span with configurable telemetry failures.""" + + def __init__( + self, + span: FakeSpan, + *, + enter_failure: Exception | None = None, + exit_failure: Exception | None = None, + ) -> None: + self.span = span + self.enter_failure = enter_failure + self.exit_failure = exit_failure + self.exits: list[tuple[Any, Any, Any]] = [] + + def __enter__(self) -> FakeSpan: + """Enter the fake span context.""" + if self.enter_failure is not None: + raise self.enter_failure + return self.span + + def __exit__(self, *exc: Any) -> bool: + """Record context exit and never suppress application exceptions.""" + self.exits.append(exc) + if self.exit_failure is not None: + raise self.exit_failure + return False + + +class FakeTracer: + """Create deterministic spans using the OpenTelemetry tracer call shape.""" + + def __init__( + self, + *, + start_failure: Exception | None = None, + enter_failure: Exception | None = None, + exit_failure: Exception | None = None, + set_failure: Exception | None = None, + ) -> None: + self.start_failure = start_failure + self.enter_failure = enter_failure + self.exit_failure = exit_failure + self.set_failure = set_failure + self.starts: list[tuple[str, dict[str, str], bool, bool]] = [] + self.contexts: list[FakeSpanContext] = [] + + def start_as_current_span( + self, + name: str, + *, + attributes: dict[str, str], + record_exception: bool, + set_status_on_exception: bool, + ) -> FakeSpanContext: + """Create one current-span context with exception capture disabled.""" + if self.start_failure is not None: + raise self.start_failure + self.starts.append( + ( + name, + dict(attributes), + record_exception, + set_status_on_exception, + ) + ) + context = FakeSpanContext( + FakeSpan(set_failure=self.set_failure), + enter_failure=self.enter_failure, + exit_failure=self.exit_failure, + ) + self.contexts.append(context) + return context + + +class FakeCheckpointStore: + """Provide all four checkpoint operations with deterministic outcomes.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = [] + self.results: dict[str, Any] = {} + self.failures: dict[str, BaseException] = {} + + def _run(self, operation: str, *args: Any, **kwargs: Any) -> Any: + """Record and execute one configured checkpoint operation.""" + self.calls.append((operation, args, kwargs)) + failure = self.failures.get(operation) + if failure is not None: + raise failure + return self.results.get(operation) + + def load(self, *args: Any, **kwargs: Any) -> Any: + """Run one package-owned load.""" + return self._run("load", *args, **kwargs) + + def load_in_transaction(self, *args: Any, **kwargs: Any) -> Any: + """Run one caller-owned load.""" + return self._run("load_in_transaction", *args, **kwargs) + + def save(self, *args: Any, **kwargs: Any) -> Any: + """Run one package-owned save.""" + return self._run("save", *args, **kwargs) + + def save_in_transaction(self, *args: Any, **kwargs: Any) -> Any: + """Run one caller-owned save.""" + return self._run("save_in_transaction", *args, **kwargs) + + +def checkpoint() -> BatchResultCheckpoint: + """Build one valid checkpoint without embedding it in telemetry attributes.""" + return BatchResultCheckpoint( + schema_version=1, + batch_id="batch-secret", + endpoint_alias="private-endpoint", + file_kind="result", + file_id="file-secret", + file_line_number=2, + batch_line_count=2, + record_count=1, + prefix_sha256="a" * 64, + ) + + +def clock(*values: int | BaseException) -> Callable[[], int]: + """Return a deterministic monotonic clock with optional failures.""" + remaining = list(values) + + def read() -> int: + value = remaining.pop(0) + if isinstance(value, BaseException): + raise value + return value + + return read + + +def flattened_telemetry(tracer: FakeTracer, meter: FakeMeter) -> str: + """Render all captured telemetry for secret-leak assertions.""" + parts: list[str] = [repr(tracer.starts), repr(meter.created)] + for context in tracer.contexts: + parts.append(repr(context.span.attributes)) + parts.append(repr(context.span.recorded_exceptions)) + parts.append(repr(meter.counter.measurements)) + parts.append(repr(meter.histogram.measurements)) + return " ".join(parts) + + +def test_successful_load_emits_fixed_low_cardinality_signals() -> None: + """A successful load emits one safe span, count, and duration measurement.""" + store = FakeCheckpointStore() + expected = checkpoint() + store.results["load"] = expected + tracer = FakeTracer() + meter = FakeMeter() + observed = OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=meter, + monotonic_ns=clock(10, 2_000_000_010), + ) + + assert observed.load("worker-secret", "batch-secret", "private-endpoint") == expected + assert store.calls == [ + ( + "load", + ("worker-secret", "batch-secret", "private-endpoint"), + {}, + ) + ] + assert tracer.starts == [ + ( + "pg_llm_batch.checkpoint.load", + { + "pg_llm_batch.checkpoint.operation": "load", + "pg_llm_batch.checkpoint.transaction_owner": "package", + }, + False, + False, + ) + ] + assert tracer.contexts[0].span.attributes == { + "pg_llm_batch.checkpoint.outcome": "success" + } + assert meter.counter.measurements == [ + RecordedMeasurement( + 1.0, + { + "pg_llm_batch.checkpoint.operation": "load", + "pg_llm_batch.checkpoint.transaction_owner": "package", + "pg_llm_batch.checkpoint.outcome": "success", + }, + ) + ] + assert meter.histogram.measurements == [ + RecordedMeasurement( + 2.0, + { + "pg_llm_batch.checkpoint.operation": "load", + "pg_llm_batch.checkpoint.transaction_owner": "package", + "pg_llm_batch.checkpoint.outcome": "success", + }, + ) + ] + captured = flattened_telemetry(tracer, meter) + for secret in ( + "worker-secret", + "batch-secret", + "private-endpoint", + "file-secret", + "a" * 64, + ): + assert secret not in captured + + +def test_conflict_is_re_raised_without_recording_sensitive_exception_data() -> None: + """A checkpoint conflict uses a fixed error class and preserves the exception.""" + store = FakeCheckpointStore() + conflict = CheckpointConflictError( + "worker-secret", + "batch-secret", + "expected_previous_stale", + ) + store.failures["save"] = conflict + tracer = FakeTracer() + meter = FakeMeter() + observed = OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=meter, + monotonic_ns=clock(0, 500_000_000), + ) + + with pytest.raises(CheckpointConflictError) as raised: + observed.save("worker-secret", checkpoint()) + + assert raised.value is conflict + assert tracer.contexts[0].span.recorded_exceptions == [] + assert tracer.contexts[0].span.attributes == { + "pg_llm_batch.checkpoint.outcome": "conflict", + "error.type": "checkpoint_conflict", + } + expected_attributes = { + "pg_llm_batch.checkpoint.operation": "save", + "pg_llm_batch.checkpoint.transaction_owner": "package", + "pg_llm_batch.checkpoint.outcome": "conflict", + "error.type": "checkpoint_conflict", + } + assert meter.counter.measurements == [RecordedMeasurement(1.0, expected_attributes)] + assert meter.histogram.measurements == [ + RecordedMeasurement(0.5, expected_attributes) + ] + captured = flattened_telemetry(tracer, meter) + assert "worker-secret" not in captured + assert "batch-secret" not in captured + assert "expected_previous_stale" not in captured + + +@pytest.mark.parametrize( + ("failure", "outcome", "error_type"), + [ + ( + ValidationError(field="tenant_scope", value="secret", reason="invalid"), + "validation_error", + "validation_error", + ), + (RuntimeError("provider-secret"), "error", "internal_error"), + ], +) +def test_failure_classification_is_bounded( + failure: BaseException, + outcome: str, + error_type: str, +) -> None: + """Validation and internal failures use finite non-secret classifications.""" + store = FakeCheckpointStore() + store.failures["load"] = failure + tracer = FakeTracer() + meter = FakeMeter() + observed = OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=meter, + monotonic_ns=clock(100, 90), + ) + + with pytest.raises(type(failure)) as raised: + observed.load("worker-secret", "batch-secret", "private-endpoint") + + assert raised.value is failure + assert tracer.contexts[0].span.attributes == { + "pg_llm_batch.checkpoint.outcome": outcome, + "error.type": error_type, + } + assert meter.histogram.measurements[0].value == 0.0 + assert "secret" not in flattened_telemetry(tracer, meter) + + +def test_transaction_methods_preserve_cursor_and_expected_checkpoint_arguments() -> None: + """Caller-owned transaction methods delegate unchanged and use caller labels.""" + store = FakeCheckpointStore() + cursor = object() + previous = checkpoint() + candidate = BatchResultCheckpoint( + schema_version=1, + batch_id="batch-secret", + endpoint_alias="private-endpoint", + file_kind="result", + file_id="file-secret", + file_line_number=3, + batch_line_count=3, + record_count=2, + prefix_sha256="b" * 64, + ) + store.results["load_in_transaction"] = previous + store.results["save_in_transaction"] = candidate + tracer = FakeTracer() + meter = FakeMeter() + observed = OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=meter, + monotonic_ns=clock(0, 1, 2, 4), + ) + + assert ( + observed.load_in_transaction( + cursor, + "worker-secret", + "batch-secret", + "private-endpoint", + ) + == previous + ) + assert ( + observed.save_in_transaction( + cursor, + "worker-secret", + candidate, + expected_previous=previous, + ) + == candidate + ) + assert store.calls == [ + ( + "load_in_transaction", + (cursor, "worker-secret", "batch-secret", "private-endpoint"), + {}, + ), + ( + "save_in_transaction", + (cursor, "worker-secret", candidate), + {"expected_previous": previous}, + ), + ] + assert [start[0] for start in tracer.starts] == [ + "pg_llm_batch.checkpoint.load", + "pg_llm_batch.checkpoint.save", + ] + assert all( + measurement.attributes["pg_llm_batch.checkpoint.transaction_owner"] + == "caller" + for measurement in meter.counter.measurements + ) + + +@pytest.mark.parametrize( + "tracer", + [ + FakeTracer(start_failure=RuntimeError("telemetry-secret")), + FakeTracer(enter_failure=RuntimeError("telemetry-secret")), + FakeTracer(exit_failure=RuntimeError("telemetry-secret")), + FakeTracer(set_failure=RuntimeError("telemetry-secret")), + ], +) +def test_trace_failures_never_change_checkpoint_success(tracer: FakeTracer) -> None: + """Tracing failures are isolated from the durable checkpoint operation.""" + store = FakeCheckpointStore() + expected = checkpoint() + store.results["load"] = expected + meter = FakeMeter() + observed = OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=meter, + monotonic_ns=clock(0, 1), + ) + + assert observed.load("worker-secret", "batch-secret", "private-endpoint") == expected + assert store.calls[0][0] == "load" + + +@pytest.mark.parametrize( + "meter", + [ + FakeMeter(create_failure=RuntimeError("telemetry-secret")), + FakeMeter(measurement_failure=RuntimeError("telemetry-secret")), + ], +) +def test_metric_failures_never_change_checkpoint_success(meter: FakeMeter) -> None: + """Metric creation and export failures remain best-effort observability.""" + store = FakeCheckpointStore() + expected = checkpoint() + store.results["save"] = expected + observed = OpenTelemetryCheckpointStore( + store, + tracer=FakeTracer(), + meter=meter, + monotonic_ns=clock(0, 1), + ) + + assert observed.save("worker-secret", expected) == expected + assert store.calls[0][0] == "save" + + +def test_telemetry_and_clock_failures_never_mask_checkpoint_failure() -> None: + """The original application exception survives every observer-side failure.""" + store = FakeCheckpointStore() + failure = RuntimeError("application-secret") + store.failures["load"] = failure + observed = OpenTelemetryCheckpointStore( + store, + tracer=FakeTracer(exit_failure=RuntimeError("telemetry-secret")), + meter=FakeMeter(measurement_failure=RuntimeError("telemetry-secret")), + monotonic_ns=clock(RuntimeError("clock-secret"), RuntimeError("clock-secret")), + ) + + with pytest.raises(RuntimeError) as raised: + observed.load("worker-secret", "batch-secret", "private-endpoint") + + assert raised.value is failure diff --git a/tests/test_checkpoint_telemetry_clock.py b/tests/test_checkpoint_telemetry_clock.py new file mode 100644 index 000000000..03b3394ae --- /dev/null +++ b/tests/test_checkpoint_telemetry_clock.py @@ -0,0 +1,123 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Clock-failure coverage for durable checkpoint telemetry.""" + +from typing import Any + +import pytest + +from pg_llm_batch.checkpoint_telemetry import OpenTelemetryCheckpointStore + + +class Span: + """Accept bounded attributes.""" + + def set_attribute(self, _name: str, _value: str) -> None: + """Accept one attribute.""" + return None + + +class SpanContext: + """Provide one no-op span context.""" + + def __enter__(self) -> Span: + """Enter the context.""" + return Span() + + def __exit__(self, *_exc: Any) -> bool: + """Exit without suppressing application behavior.""" + return False + + +class Tracer: + """Create one no-op span context.""" + + def start_as_current_span(self, *_args: Any, **_kwargs: Any) -> SpanContext: + """Start a no-op span.""" + return SpanContext() + + +class Instrument: + """Capture histogram values while accepting counter calls.""" + + def __init__(self) -> None: + self.values: list[float] = [] + + def add(self, _value: int, *, attributes: dict[str, str]) -> None: + """Accept one counter value.""" + del attributes + + def record(self, value: float, *, attributes: dict[str, str]) -> None: + """Capture one duration value.""" + del attributes + self.values.append(value) + + +class Meter: + """Return deterministic instruments.""" + + def __init__(self) -> None: + self.counter = Instrument() + self.histogram = Instrument() + + def create_counter(self, *_args: Any, **_kwargs: Any) -> Instrument: + """Return the counter.""" + return self.counter + + def create_histogram(self, *_args: Any, **_kwargs: Any) -> Instrument: + """Return the histogram.""" + return self.histogram + + +class Store: + """Return one successful checkpoint sentinel.""" + + def load(self, *_args: Any, **_kwargs: Any) -> str: + """Return the sentinel.""" + return "checkpoint" + + +def observed_store(clock: Any, meter: Meter) -> OpenTelemetryCheckpointStore: + """Build one observed store around the supplied clock.""" + return OpenTelemetryCheckpointStore( + Store(), + tracer=Tracer(), + meter=meter, + monotonic_ns=clock, + ) + + +def test_end_clock_failure_records_zero_duration() -> None: + """A failed end-clock read remains a zero-duration observer signal.""" + readings: list[int | BaseException] = [1, RuntimeError("clock unavailable")] + + def clock() -> int: + value = readings.pop(0) + if isinstance(value, BaseException): + raise value + return value + + meter = Meter() + observed = observed_store(clock, meter) + + assert observed.load("consumer-a", "batch-a", "endpoint-a") == "checkpoint" + assert meter.histogram.values == [0.0] + + +@pytest.mark.parametrize( + "invalid_value", + [object(), "not-a-clock", None, float("nan"), float("inf")], +) +def test_nonnumeric_or_nonfinite_clock_values_cannot_change_application_success( + invalid_value: Any, +) -> None: + """Malformed host clock evidence degrades to a zero-duration measurement.""" + readings = [invalid_value, invalid_value] + + def clock() -> Any: + return readings.pop(0) + + meter = Meter() + observed = observed_store(clock, meter) + + assert observed.load("consumer-a", "batch-a", "endpoint-a") == "checkpoint" + assert meter.histogram.values == [0.0] diff --git a/tests/test_checkpoint_telemetry_concurrency.py b/tests/test_checkpoint_telemetry_concurrency.py new file mode 100644 index 000000000..ca276224e --- /dev/null +++ b/tests/test_checkpoint_telemetry_concurrency.py @@ -0,0 +1,173 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Concurrency tests for durable checkpoint telemetry.""" + +from concurrent.futures import ThreadPoolExecutor +from threading import Lock +from typing import Any + +from pg_llm_batch.checkpoint_telemetry import OpenTelemetryCheckpointStore + + +class ConcurrentSpan: + """Collect attributes local to one concurrent operation.""" + + def __init__(self, initial: dict[str, str]) -> None: + self.attributes = dict(initial) + + def set_attribute(self, name: str, value: str) -> None: + """Set one operation-local attribute.""" + self.attributes[name] = value + + +class ConcurrentSpanContext: + """Own one operation-local span.""" + + def __init__(self, span: ConcurrentSpan, completed: list[dict[str, str]], lock: Lock) -> None: + self.span = span + self.completed = completed + self.lock = lock + + def __enter__(self) -> ConcurrentSpan: + """Enter the operation-local span.""" + return self.span + + def __exit__(self, *_exc: Any) -> bool: + """Publish one completed immutable attribute snapshot.""" + with self.lock: + self.completed.append(dict(self.span.attributes)) + return False + + +class ConcurrentTracer: + """Create independent spans for concurrent calls.""" + + def __init__(self) -> None: + self.completed: list[dict[str, str]] = [] + self.lock = Lock() + + def start_as_current_span( + self, + _name: str, + *, + attributes: dict[str, str], + **_kwargs: Any, + ) -> ConcurrentSpanContext: + """Create one operation-local context from a copied initial mapping.""" + return ConcurrentSpanContext( + ConcurrentSpan(attributes), + self.completed, + self.lock, + ) + + +class ConcurrentInstrument: + """Collect copied metric attributes safely across threads.""" + + def __init__(self) -> None: + self.measurements: list[dict[str, str]] = [] + self.lock = Lock() + + def add(self, _value: int, *, attributes: dict[str, str]) -> None: + """Record one counter mapping.""" + with self.lock: + self.measurements.append(dict(attributes)) + + def record(self, _value: float, *, attributes: dict[str, str]) -> None: + """Record one histogram mapping.""" + with self.lock: + self.measurements.append(dict(attributes)) + + +class ConcurrentMeter: + """Provide independent counter and histogram collectors.""" + + def __init__(self) -> None: + self.counter = ConcurrentInstrument() + self.histogram = ConcurrentInstrument() + + def create_counter(self, *_args: Any, **_kwargs: Any) -> ConcurrentInstrument: + """Return the shared thread-safe counter.""" + return self.counter + + def create_histogram(self, *_args: Any, **_kwargs: Any) -> ConcurrentInstrument: + """Return the shared thread-safe histogram.""" + return self.histogram + + +class ConcurrentStore: + """Count delegated loads and saves safely across threads.""" + + def __init__(self) -> None: + self.calls: list[str] = [] + self.lock = Lock() + + def load(self, *_args: Any, **_kwargs: Any) -> str: + """Record and return one load result.""" + with self.lock: + self.calls.append("load") + return "loaded" + + def save(self, *_args: Any, **_kwargs: Any) -> str: + """Record and return one save result.""" + with self.lock: + self.calls.append("save") + return "saved" + + +def test_concurrent_operations_do_not_share_mutable_telemetry_attributes() -> None: + """Concurrent load and save signals retain independent finite attributes.""" + store = ConcurrentStore() + tracer = ConcurrentTracer() + meter = ConcurrentMeter() + clock_lock = Lock() + clock_value = 0 + + def clock() -> int: + nonlocal clock_value + with clock_lock: + clock_value += 1 + return clock_value + + observed = OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=meter, + monotonic_ns=clock, + ) + + def invoke(index: int) -> str: + if index % 2 == 0: + return observed.load("consumer-a", "batch-a", "endpoint-a") + return observed.save("consumer-a", object()) + + with ThreadPoolExecutor(max_workers=8) as executor: + results = list(executor.map(invoke, range(32))) + + assert results.count("loaded") == 16 + assert results.count("saved") == 16 + assert store.calls.count("load") == 16 + assert store.calls.count("save") == 16 + assert len(tracer.completed) == 32 + assert len(meter.counter.measurements) == 32 + assert len(meter.histogram.measurements) == 32 + + expected_attributes = { + "pg_llm_batch.checkpoint.operation", + "pg_llm_batch.checkpoint.transaction_owner", + "pg_llm_batch.checkpoint.outcome", + } + for measurement in meter.counter.measurements + meter.histogram.measurements: + assert measurement["pg_llm_batch.checkpoint.operation"] in {"load", "save"} + assert measurement["pg_llm_batch.checkpoint.transaction_owner"] == "package" + assert measurement["pg_llm_batch.checkpoint.outcome"] == "success" + assert "error.type" not in measurement + assert "db.system.name" not in measurement + assert set(measurement) == expected_attributes + + for attributes in tracer.completed: + assert attributes["pg_llm_batch.checkpoint.operation"] in {"load", "save"} + assert attributes["pg_llm_batch.checkpoint.transaction_owner"] == "package" + assert attributes["pg_llm_batch.checkpoint.outcome"] == "success" + assert "error.type" not in attributes + assert "db.system.name" not in attributes + assert set(attributes) == expected_attributes \ No newline at end of file diff --git a/tests/test_checkpoint_telemetry_documentation.py b/tests/test_checkpoint_telemetry_documentation.py new file mode 100644 index 000000000..0cc43a8e5 --- /dev/null +++ b/tests/test_checkpoint_telemetry_documentation.py @@ -0,0 +1,90 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Documentation contracts for durable checkpoint observability.""" + +from pathlib import Path + +ROOT = Path(__file__).parents[1] + + +def normalized(path: str) -> str: + """Read one project document with layout-insensitive whitespace.""" + return " ".join((ROOT / path).read_text(encoding="utf-8").split()) + + +def test_authoritative_documents_define_checkpoint_telemetry_boundary() -> None: + """Architecture and contributor contracts state the safe signal boundary.""" + required = { + "ARCHITECTURE.md": ( + "OpenTelemetryCheckpointStore", + "tenant, consumer, batch, endpoint, file, digest, cursor, and DSN values", + "best-effort telemetry cannot change checkpoint operation semantics", + "storage-agnostic", + "does not emit `db.system.name`", + "StatusCode.ERROR", + "successful checkpoint spans leave status Unset", + ), + "AGENTS.md": ( + "OpenTelemetry checkpoint signals", + "record_exception=False", + "set_status_on_exception=False", + "finite low-cardinality `error.type`", + "StatusCode.ERROR", + "successful spans must leave status Unset", + ), + "CLAUDE.md": ( + "OpenTelemetry checkpoint signals", + "Never add tenant, consumer, batch, endpoint, file, digest, cursor, or DSN values", + "StatusCode.ERROR", + "on success, leave status Unset", + ), + "CHANGELOG.md": ( + "OpenTelemetry-compatible checkpoint spans and metrics", + "checkpoint_conflict, validation_error, and internal_error", + ), + "docs/checkpoint-observability.md": ( + "The wrapper delegates all arguments and returns unchanged", + "Package-owned telemetry never contains tenant scope", + "supplies `(None, None, None)` when closing the span context", + "Non-cancellation process-control exceptions remain outside this observer-failure guarantee", + "Package operation spans are deliberately storage-agnostic", + "do not emit `db.system.name`", + "failed checkpoint spans explicitly set OpenTelemetry status Error without a description", + "successful checkpoint spans leave status Unset", + ), + } + for path, phrases in required.items(): + text = normalized(path) + for phrase in phrases: + assert phrase in text, f"{path} must contain {phrase!r}" + + +def test_adr_and_doctoring_record_operator_and_standards_contracts() -> None: + """The decision record and doctoring cite authoritative telemetry evidence.""" + adr = normalized("docs/adr/0008-checkpoint-opentelemetry-observability.md") + doctoring = normalized( + "docs/doctoring/checkpoint-opentelemetry-observability.md" + ) + for phrase in ( + "dependency-injected OpenTelemetry-compatible tracer and meter", + "Package-owned spans and metrics never contain resource identifiers", + "histogram unit is seconds", + "Exporter, processor, sampler, and provider ownership remains with the host", + "storage-agnostic", + "do not emit `db.system.name`", + "failed checkpoint spans explicitly set OpenTelemetry status Error without a description", + "successful checkpoint spans leave status Unset", + "Rollback", + ): + assert phrase in adr + for phrase in ( + "OpenTelemetry semantic conventions 1.44.0", + "Semantic conventions for database client spans", + "Recording errors", + "failed checkpoint spans explicitly set OpenTelemetry status Error without a description", + "successful checkpoint spans leave status Unset", + "does **not** emit `db.system.name`", + "PostgreSQL", + "storage-agnostic", + "APA 7", + ): + assert phrase in doctoring diff --git a/tests/test_checkpoint_telemetry_exception_boundary.py b/tests/test_checkpoint_telemetry_exception_boundary.py new file mode 100644 index 000000000..368049eb4 --- /dev/null +++ b/tests/test_checkpoint_telemetry_exception_boundary.py @@ -0,0 +1,130 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Regression tests for the checkpoint telemetry exception boundary.""" + +from asyncio import CancelledError +from typing import Any + +import pytest + +from pg_llm_batch.checkpoint_telemetry import OpenTelemetryCheckpointStore + + +class CapturingSpan: + """Accept bounded span attributes.""" + + def set_attribute(self, _name: str, _value: str) -> None: + """Accept one package-owned attribute.""" + return None + + +class CapturingContext: + """Capture arguments supplied when the span context closes.""" + + def __init__(self, *, exit_failure: BaseException | None = None) -> None: + self.exit_failure = exit_failure + self.exit_arguments: list[tuple[Any, Any, Any]] = [] + + def __enter__(self) -> CapturingSpan: + """Enter one deterministic span context.""" + return CapturingSpan() + + def __exit__(self, *exc: Any) -> bool: + """Capture close arguments and optionally raise observer cancellation.""" + self.exit_arguments.append(exc) + if self.exit_failure is not None: + raise self.exit_failure + return False + + +class CapturingTracer: + """Return one context without interpreting operation exceptions.""" + + def __init__(self, context: CapturingContext) -> None: + self.context = context + + def start_as_current_span(self, *_args: Any, **_kwargs: Any) -> CapturingContext: + """Start the configured span context.""" + return self.context + + +class CancellingMeter: + """Raise task cancellation from instrument creation.""" + + def create_counter(self, *_args: Any, **_kwargs: Any) -> Any: + """Cancel counter creation.""" + raise CancelledError() + + def create_histogram(self, *_args: Any, **_kwargs: Any) -> Any: + """Cancel histogram creation.""" + raise CancelledError() + + +class NoOpMeter: + """Provide instruments that accept metrics.""" + + def create_counter(self, *_args: Any, **_kwargs: Any) -> "NoOpMeter": + """Return this object as a counter.""" + return self + + def create_histogram(self, *_args: Any, **_kwargs: Any) -> "NoOpMeter": + """Return this object as a histogram.""" + return self + + def add(self, *_args: Any, **_kwargs: Any) -> None: + """Accept one counter measurement.""" + return None + + def record(self, *_args: Any, **_kwargs: Any) -> None: + """Accept one histogram measurement.""" + return None + + +class FailingStore: + """Raise one application exception from a checkpoint load.""" + + def __init__(self, failure: BaseException) -> None: + self.failure = failure + + def load(self, *_args: Any, **_kwargs: Any) -> Any: + """Raise the configured application failure.""" + raise self.failure + + +class SuccessfulStore: + """Return one deterministic application result.""" + + def load(self, *_args: Any, **_kwargs: Any) -> str: + """Return the durable checkpoint sentinel.""" + return "durable-result" + + +def test_application_exception_is_not_passed_to_span_context_exit() -> None: + """Observers never receive the application exception on context close.""" + context = CapturingContext() + failure = RuntimeError("application-state") + observed = OpenTelemetryCheckpointStore( + FailingStore(failure), + tracer=CapturingTracer(context), + meter=NoOpMeter(), + monotonic_ns=lambda: 0, + ) + + with pytest.raises(RuntimeError) as raised: + observed.load("consumer-a", "batch-a", "endpoint-a") + + assert raised.value is failure + assert context.exit_arguments == [(None, None, None)] + + +def test_observer_cancellation_does_not_cancel_successful_checkpoint_operation() -> None: + """Telemetry cancellation is isolated like other observer failure.""" + context = CapturingContext(exit_failure=CancelledError()) + observed = OpenTelemetryCheckpointStore( + SuccessfulStore(), + tracer=CapturingTracer(context), + meter=CancellingMeter(), + monotonic_ns=lambda: 0, + ) + + assert observed.load("consumer-a", "batch-a", "endpoint-a") == "durable-result" + assert context.exit_arguments == [(None, None, None)] diff --git a/tests/test_checkpoint_telemetry_public_api.py b/tests/test_checkpoint_telemetry_public_api.py new file mode 100644 index 000000000..8e3d0a397 --- /dev/null +++ b/tests/test_checkpoint_telemetry_public_api.py @@ -0,0 +1,12 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Public API contract for checkpoint observability.""" + +from pg_llm_batch import OpenTelemetryCheckpointStore +from pg_llm_batch.checkpoint_telemetry import ( + OpenTelemetryCheckpointStore as ModuleCheckpointStore, +) + + +def test_checkpoint_telemetry_wrapper_is_exported_from_package_root() -> None: + """Embedding hosts can import the wrapper from the documented package API.""" + assert OpenTelemetryCheckpointStore is ModuleCheckpointStore diff --git a/tests/test_checkpoint_telemetry_span_status.py b/tests/test_checkpoint_telemetry_span_status.py new file mode 100644 index 000000000..4b710e9e7 --- /dev/null +++ b/tests/test_checkpoint_telemetry_span_status.py @@ -0,0 +1,241 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Regression tests for OpenTelemetry checkpoint span error status.""" + +from __future__ import annotations + +import builtins +import sys +from pathlib import Path +from types import ModuleType +from typing import Any + +import pytest + +import pg_llm_batch.checkpoint_telemetry as telemetry + + +_ERROR_STATUS = object() + + +class _StatusCode: + """Provide the minimum OpenTelemetry StatusCode shape used by the wrapper.""" + + ERROR = _ERROR_STATUS + + +class _Instrument: + """Accept bounded counter and histogram measurements.""" + + def add(self, _value: int, *, attributes: dict[str, str]) -> None: + """Accept one counter measurement.""" + del attributes + + def record(self, _value: float, *, attributes: dict[str, str]) -> None: + """Accept one histogram measurement.""" + del attributes + + +class _Meter: + """Create no-op metric instruments with the expected API shape.""" + + def create_counter(self, _name: str, **_kwargs: Any) -> _Instrument: + """Create one counter.""" + return _Instrument() + + def create_histogram(self, _name: str, **_kwargs: Any) -> _Instrument: + """Create one histogram.""" + return _Instrument() + + +class _Span: + """Capture bounded attributes and explicit span status calls.""" + + def __init__(self, *, status_failure: BaseException | None = None) -> None: + self.attributes: dict[str, str] = {} + self.statuses: list[Any] = [] + self.status_failure = status_failure + + def set_attribute(self, name: str, value: str) -> None: + """Capture one span attribute.""" + self.attributes[name] = value + + def set_status(self, status: Any) -> None: + """Capture one status value or simulate observer failure.""" + self.statuses.append(status) + if self.status_failure is not None: + raise self.status_failure + + +class _SpanContext: + """Own one span without receiving the application exception.""" + + def __init__(self, span: _Span) -> None: + self.span = span + self.exits: list[tuple[Any, Any, Any]] = [] + + def __enter__(self) -> _Span: + """Enter the fake span.""" + return self.span + + def __exit__(self, *exc: Any) -> bool: + """Capture the sanitized context-manager exit tuple.""" + self.exits.append(exc) + return False + + +class _Tracer: + """Create one status-aware span per checkpoint operation.""" + + def __init__(self, *, status_failure: BaseException | None = None) -> None: + self.status_failure = status_failure + self.contexts: list[_SpanContext] = [] + + def start_as_current_span(self, _name: str, **_kwargs: Any) -> _SpanContext: + """Create one context with automatic exception handling disabled upstream.""" + context = _SpanContext(_Span(status_failure=self.status_failure)) + self.contexts.append(context) + return context + + +class _Store: + """Provide one deterministic load result or failure.""" + + def __init__(self, *, failure: BaseException | None = None) -> None: + self.failure = failure + self.result = object() + + def load(self, *_args: Any, **_kwargs: Any) -> Any: + """Return one result or raise the exact configured application failure.""" + if self.failure is not None: + raise self.failure + return self.result + + +def _install_fake_otel(monkeypatch: pytest.MonkeyPatch) -> None: + """Install only the OpenTelemetry API symbol needed for explicit Error status.""" + package = ModuleType("opentelemetry") + trace_module = ModuleType("opentelemetry.trace") + trace_module.StatusCode = _StatusCode + package.trace = trace_module + monkeypatch.setitem(sys.modules, "opentelemetry", package) + monkeypatch.setitem(sys.modules, "opentelemetry.trace", trace_module) + + +def _normalized(path: str) -> str: + """Normalize Markdown layout while preserving semantic documentation text.""" + return " ".join(Path(path).read_text(encoding="utf-8").split()) + + +def test_failed_operation_sets_error_status_without_exception_description( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Failed checkpoint spans use OpenTelemetry Error without secret description.""" + _install_fake_otel(monkeypatch) + failure = RuntimeError("provider-secret") + store = _Store(failure=failure) + tracer = _Tracer() + observed = telemetry.OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=_Meter(), + ) + + with pytest.raises(RuntimeError) as raised: + observed.load("consumer-secret", "batch-secret", "endpoint-secret") + + assert raised.value is failure + span = tracer.contexts[0].span + assert span.statuses == [_ERROR_STATUS] + assert span.attributes["error.type"] == "internal_error" + assert tracer.contexts[0].exits == [(None, None, None)] + assert "provider-secret" not in repr(span.statuses) + + +def test_successful_operation_leaves_span_status_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Successful checkpoint spans retain the OpenTelemetry default Unset status.""" + _install_fake_otel(monkeypatch) + store = _Store() + tracer = _Tracer() + observed = telemetry.OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=_Meter(), + ) + + assert observed.load("consumer", "batch", "endpoint") is store.result + assert tracer.contexts[0].span.statuses == [] + + +def test_span_status_failure_never_masks_application_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failing observer status mutation preserves the exact application error.""" + _install_fake_otel(monkeypatch) + failure = RuntimeError("application-secret") + store = _Store(failure=failure) + tracer = _Tracer(status_failure=RuntimeError("telemetry-secret")) + observed = telemetry.OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=_Meter(), + ) + + with pytest.raises(RuntimeError) as raised: + observed.load("consumer", "batch", "endpoint") + + assert raised.value is failure + assert tracer.contexts[0].span.statuses == [_ERROR_STATUS] + + +def test_missing_opentelemetry_api_keeps_failure_authoritative( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Missing optional OpenTelemetry API support degrades to status-less telemetry.""" + original_import = builtins.__import__ + + def reject_opentelemetry( + name: str, + globals: dict[str, Any] | None = None, + locals: dict[str, Any] | None = None, + fromlist: tuple[str, ...] = (), + level: int = 0, + ) -> Any: + if name == "opentelemetry.trace": + raise ImportError("optional API unavailable") + return original_import(name, globals, locals, fromlist, level) + + monkeypatch.delitem(sys.modules, "opentelemetry.trace", raising=False) + monkeypatch.delitem(sys.modules, "opentelemetry", raising=False) + monkeypatch.setattr(builtins, "__import__", reject_opentelemetry) + failure = RuntimeError("application-secret") + store = _Store(failure=failure) + tracer = _Tracer() + observed = telemetry.OpenTelemetryCheckpointStore( + store, + tracer=tracer, + meter=_Meter(), + ) + + with pytest.raises(RuntimeError) as raised: + observed.load("consumer", "batch", "endpoint") + + assert raised.value is failure + assert tracer.contexts[0].span.statuses == [] + + +def test_authoritative_docs_require_explicit_error_status_and_current_semconv() -> None: + """Authoritative guidance records the explicit Error-status confidentiality rule.""" + adr = _normalized("docs/adr/0008-checkpoint-opentelemetry-observability.md") + operator = _normalized("docs/checkpoint-observability.md") + doctoring = _normalized("docs/doctoring/checkpoint-opentelemetry-observability.md") + + required = ( + "failed checkpoint spans explicitly set OpenTelemetry status Error without a description", + "successful checkpoint spans leave status Unset", + ) + for document in (adr, operator, doctoring): + for phrase in required: + assert phrase in document + assert "OpenTelemetry semantic conventions 1.44.0" in doctoring diff --git a/tests/test_checkpoint_telemetry_storage_semantics.py b/tests/test_checkpoint_telemetry_storage_semantics.py new file mode 100644 index 000000000..4f259021f --- /dev/null +++ b/tests/test_checkpoint_telemetry_storage_semantics.py @@ -0,0 +1,121 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Regression tests for storage-agnostic checkpoint telemetry semantics.""" + +from __future__ import annotations + +from typing import Any + +from pg_llm_batch.checkpoint_telemetry import OpenTelemetryCheckpointStore + + +class _Instrument: + """Accept metric measurements without retaining application values.""" + + def add(self, _value: int, *, attributes: dict[str, str]) -> None: + """Accept one counter measurement.""" + assert "db.system.name" not in attributes + + def record(self, _value: float, *, attributes: dict[str, str]) -> None: + """Accept one histogram measurement.""" + assert "db.system.name" not in attributes + + +class _Meter: + """Provide no-op metric instruments using the OpenTelemetry API shape.""" + + def create_counter(self, _name: str, *, unit: str, description: str) -> _Instrument: + """Create one counter after validating bounded metadata.""" + assert unit == "{operation}" + assert description + return _Instrument() + + def create_histogram( + self, + _name: str, + *, + unit: str, + description: str, + ) -> _Instrument: + """Create one histogram after validating bounded metadata.""" + assert unit == "s" + assert description + return _Instrument() + + +class _Span: + """Accept bounded span mutation.""" + + def set_attribute(self, name: str, _value: str) -> None: + """Reject accidental database-client semantic claims.""" + assert name != "db.system.name" + + +class _SpanContext: + """Provide one deterministic context manager around a fake span.""" + + def __enter__(self) -> _Span: + """Enter the fake span.""" + return _Span() + + def __exit__(self, *_exc: Any) -> bool: + """Exit without suppressing application exceptions.""" + return False + + +class _Tracer: + """Capture the initial attributes of one package operation span.""" + + def __init__(self) -> None: + """Initialize an empty start record.""" + self.starts: list[tuple[str, dict[str, str], bool, bool]] = [] + + def start_as_current_span( + self, + name: str, + *, + attributes: dict[str, str], + record_exception: bool, + set_status_on_exception: bool, + ) -> _SpanContext: + """Record one span request using the dependency-injected API shape.""" + self.starts.append( + ( + name, + dict(attributes), + record_exception, + set_status_on_exception, + ) + ) + return _SpanContext() + + +class _CompatibleStore: + """Model a non-PostgreSQL host store that implements the public load seam.""" + + def load(self, _consumer_name: str, _batch_id: str, _endpoint_alias: str) -> None: + """Return an empty checkpoint without exposing a database technology.""" + return None + + +def test_operation_span_is_storage_agnostic_for_compatible_host_store() -> None: + """A compatible host store must not be mislabeled as a PostgreSQL client span.""" + tracer = _Tracer() + observed = OpenTelemetryCheckpointStore( + _CompatibleStore(), + tracer=tracer, + meter=_Meter(), + monotonic_ns=lambda: 0, + ) + + assert observed.load("consumer", "batch", "endpoint") is None + assert tracer.starts == [ + ( + "pg_llm_batch.checkpoint.load", + { + "pg_llm_batch.checkpoint.operation": "load", + "pg_llm_batch.checkpoint.transaction_owner": "package", + }, + False, + False, + ) + ]