Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 36 additions & 5 deletions docs/reporting-source-adapters.md
Original file line number Diff line number Diff line change
Expand Up @@ -26,16 +26,37 @@ return InlineFetchResult(

This publishes one `partial` constituent with four independent metric cells.
It can complete a request whose `coverage.expected` is `partial`. A `full`
request receives retryable `PARTIAL_RESULT` and no publication until every
requested cell is available.
request receives retryable `PARTIAL_RESULT` while `viewability` is delayed.
When every applicable cell is available, the constituent is `present` and can
satisfy full coverage even though `completed_views` remains `unsupported`.

Offering metrics with `support: partial` are requestable alongside `exact`
metrics in both the service and source conformance. Every requested cell must
still have manifest evidence with the offering's semantic contract. The inline
adapter builds that evidence from the defaults and `cell_availability`
overrides; use an override for each inapplicable or unready cell. An offering's
`unavailable` metrics cannot pass source conformance.

Only a metric declared with `support: partial` and a stable offering reason may
emit `unsupported` cells. Declaring `exact` support promises applicability;
use `missing` or `delayed` when that measurement has not arrived. The inline
adapter rejects contradictory evidence before staging, and both producer
admission and source conformance enforce the same rule for custom executors.
For example, an exact-support spend metric awaiting billing must use
`MetricEvidence.delayed("billing_pending")` and retain partial coverage.

If a conditional metric is `unsupported` in every constituent, the batch can
still have full coverage when every constituent has other applicable, complete
measurements. Every cell of the inapplicable metric retains its `unsupported`
evidence, including its reason and lack of watermark.

| Constructor | Meaning and required evidence |
| --- | --- |
| `MetricEvidence.present(data_through)` | Measured values through a timezone-aware watermark. Every matched row for the constituent must carry a non-null value for this metric. |
| `MetricEvidence.explicit_zero(data_through=...)` | An observed zero. The watermark defaults to the fetch watermark. Any supplied values must be finite numeric zeros. |
| `MetricEvidence.missing(reason)` | No answer for this metric. A stable reason is required; a watermark is forbidden. |
| `MetricEvidence.delayed(reason, data_through=...)` | Not ready yet. A stable reason is required; retain a known watermark when available. |
| `MetricEvidence.unavailable(reason)` | The source cannot measure this metric for this constituent. Emits the existing wire status `unsupported`; a reason is required and a watermark is forbidden. |
| `MetricEvidence.unavailable(reason)` | This metric is inapplicable to this constituent. Emits `unsupported`; a reason is required and a watermark is forbidden. Use `missing` or `delayed` for an applicable measurement that has not arrived. |

Evidence is immutable. Direct `MetricEvidence(...)` construction enforces the
same invariants. Reasons use the existing manifest format: bounded ASCII,
Expand All @@ -56,8 +77,18 @@ behavior. Omitted cells inherit their constituent's status, reason, and
watermark. Explicit cells take precedence over `covered_constituent_ids` and
`unavailable_constituents`, including explicit measurements for an otherwise
missing constituent. Coverage is then reconciled from the resolved cells:
uniform statuses stay uniform, present plus explicit-zero is `present`, and
other mixtures are `partial`.
uniform statuses stay uniform. A mixture of `present`, `explicit_zero`, and
`unsupported` cells is `present` if at least one cell is available. Other
mixtures are `partial`; missing, delayed, and stale cells still prevent full
coverage.

An all-`unsupported` constituent has zero applicable metrics. It remains
`unsupported`, with a reason and no watermark, and cannot satisfy full
coverage. A manifest cannot label that constituent `partial` either. A result
containing only such constituents has coverage `none`, never an observed zero.
A partial-coverage request can retain that diagnostic result; a full-coverage
request returns `PARTIAL_RESULT` without publishing. Conformance rejects a
completed `none` result for a full-coverage request.

Explicit watermarks are bounded by period end, source read cutoff, and the
observation instant. A watermark before the period is rejected. An explicit
Expand Down
16 changes: 13 additions & 3 deletions src/adcp/reporting/conformance.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
ReportingSourceSliceRequestV1,
ReportingSourceStagedObjectReader,
SourceBatchManifestV1,
_validate_metric_applicability,
deterministic_source_publication_id_v1,
iso_duration_milliseconds_v1,
parse_verified_source_batch_manifest_v1,
Expand Down Expand Up @@ -221,11 +222,14 @@ def _validate_request_against_capabilities(
"the selected offering does not support immutable corrections",
)

exact_metrics = {metric.name for metric in offering.metrics if metric.support == "exact"}
requestable_metrics = {
metric.name for metric in offering.metrics if metric.support in {"exact", "partial"}
}
for metric in request.requested_metrics:
if metric not in exact_metrics:
if metric not in requestable_metrics:
raise _fail(
"CAPABILITY_MISMATCH", f"requested metric {metric!r} is not exact in the offering"
"CAPABILITY_MISMATCH",
f"requested metric {metric!r} is not supported in the offering",
)
exact_dimensions = {
dimension.name for dimension in offering.dimensions if dimension.support == "exact"
Expand Down Expand Up @@ -391,6 +395,12 @@ def _validate_manifest_against_request(
requested_metrics = set(request.requested_metrics)
requested_constituents = {item.constituent_id for item in request.coverage.constituents}
declared = {metric.name: metric for metric in offering.metrics}
try:
_validate_metric_applicability(offering.metrics, manifest.metric_availability)
except ValueError:
raise _fail(
"MANIFEST_MISMATCH", "unsupported cells require partial metric support with a reason"
) from None
published_cells = {
(cell.constituent_id, cell.metric): cell for cell in manifest.metric_availability
}
Expand Down
29 changes: 19 additions & 10 deletions src/adcp/reporting/inline_source.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,10 @@
For uneven metric support, return :class:`InlineFetchResult` with
``cell_availability={constituent_id: {metric_name: MetricEvidence(...)}}``.
Omitted cells retain the constituent defaults. Explicit evidence controls each
cell independently and mixed cells make the constituent partial. A cell that
withdraws a metric its own constituent's rows report withdraws that metric's
control total, so no total ever sums a disclaimed value; a cell that staged no
cell independently. Unsupported cells are inapplicable: alongside available
cells they do not make the constituent partial. A cell that withdraws a metric
its own constituent's rows report withdraws that metric's control total, so no
total ever sums a disclaimed value; a cell that staged no
rows reaches no sum and leaves its neighbours' subtotal intact. Money has no
such choice -- see ``monetary_metrics`` on :class:`InlineReportingSource`.
Metric semantics always come from the selected SDK offering.
Expand Down Expand Up @@ -100,6 +101,7 @@
SourceBatchManifestV1,
SourceBatchObjectV1,
SourceControlTotalV1,
_validate_metric_applicability,
deterministic_source_publication_id_v1,
encode_source_batch_manifest_v1,
publication_content_fingerprint_v1,
Expand Down Expand Up @@ -285,11 +287,13 @@ class InlineFetchResult:
Keys are the frozen request's constituent IDs, which need not equal media
buy IDs. Omitted cells retain the derived constituent status and watermark.
Explicit cells override even missing/unsupported constituent defaults;
mixed availability promotes the constituent to ``partial``. The SDK still
applies the source cutoff, observation ceiling, and authoritative freshness
gate. A cell watermark may advance the batch watermark without advancing
other cells. This is an adapter surface, not the manifest's wire-format
``metric_availability`` list.
unsupported cells do not demote otherwise available coverage. With no
applicable cells, the constituent stays ``unsupported`` and cannot satisfy
full coverage. Other mixed availability promotes it to ``partial``. The
SDK still applies the source cutoff, observation ceiling, and authoritative
freshness gate. A cell watermark may advance the batch watermark without
advancing other cells. This is an adapter surface, not the manifest's
wire-format ``metric_availability`` list.

Unknown keys, duplicate mapping entries, invalid evidence, contradictory
explicit zeros, and a withdrawn *monetary* cell whose own constituent's rows
Expand Down Expand Up @@ -1139,9 +1143,13 @@ def _resolve_availability(
reason = fallback_reason
if overrides.get(constituent_id):
cell_statuses = {cell.status for cell in constituent_cells}
if len(cell_statuses) == 1:
if cell_statuses == {"unsupported"}:
# Zero applicable metrics is no coverage evidence, not a
# vacuously complete measurement or an observed zero.
status = "unsupported"
elif len(cell_statuses) == 1:
status = constituent_cells[0].status
elif cell_statuses <= _AVAILABLE:
elif cell_statuses <= _AVAILABLE | {"unsupported"}:
status = "present"
else:
status = "partial"
Expand All @@ -1165,6 +1173,7 @@ def _resolve_availability(
constituent=item, status=status, data_through=watermark, reason=reason
)
)
_validate_metric_applicability(offering.metrics, cells)
return constituents, cells

def _seal_manifest(
Expand Down
15 changes: 15 additions & 0 deletions src/adcp/reporting/ledger/producer.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@
ReportingSourceSliceRequestV1,
ReportingSourceStagedObjectReader,
SourceBatchManifestV1,
_validate_metric_applicability,
coverage_denominator_fingerprint_v1,
iso_duration_milliseconds_v1,
parse_verified_source_batch_manifest_v1,
Expand Down Expand Up @@ -967,6 +968,11 @@ async def acquire_obligation(
return None

manifest = self._verified_manifest(result)
if request.coverage.expected == "full" and manifest.coverage.status != "full":
raise LedgerConflictError(
"MANIFEST_MISMATCH",
"a full-coverage request cannot complete with partial or missing coverage",
)
self._validate_manifest_currency(obligation, manifest)
rows = await self._read_rows(request, manifest)
# ``now`` freezes dispatch/lease/cutoff decisions, not publication.
Expand Down Expand Up @@ -1213,6 +1219,15 @@ async def commit_revision_from_manifest(
"""
obligation = await self._stored_obligation(obligation)
self._validate_manifest_currency(obligation, manifest)
if any(cell.status == "unsupported" for cell in manifest.metric_availability):
try:
offering = self._source.capabilities.offering(manifest.offering_id)
_validate_metric_applicability(offering.metrics, manifest.metric_availability)
except (KeyError, ValueError):
raise LedgerConflictError(
"MANIFEST_MISMATCH",
"unsupported cells require partial metric support with a reason",
) from None
now = now or self._clock()
turn = turn or WorkerTurn()
existing = await self._store.list_revisions(
Expand Down
25 changes: 21 additions & 4 deletions src/adcp/reporting/source.py
Original file line number Diff line number Diff line change
Expand Up @@ -235,10 +235,11 @@
_AVAILABLE_STATUSES = frozenset({"present", "explicit_zero"})

#: A constituent's roll-up status constrains what its metric cells may claim.
#: ``partial`` is the only genuinely mixed roll-up: its cells keep independent
#: statuses precisely so a consumer can select a compatible subset.
#: Unsupported cells are inapplicable and may accompany available cells in a
#: present constituent. Other unavailable cells still require a degraded
#: roll-up, retaining their independent evidence for consumers.
_CONSTITUENT_ALLOWS_METRIC: Mapping[str, frozenset[str]] = {
"present": frozenset({"present", "explicit_zero"}),
"present": frozenset({"present", "explicit_zero", "unsupported"}),
"explicit_zero": frozenset({"explicit_zero"}),
"unsupported": frozenset({"unsupported"}),
"delayed": frozenset({"unsupported", "delayed", "missing"}),
Expand Down Expand Up @@ -1056,6 +1057,17 @@ def _evidence_matches_status(self) -> ReportingMetricAvailabilityV1:
return self


def _validate_metric_applicability(
metrics: Sequence[MetricOfferingV1], cells: Sequence[ReportingMetricAvailabilityV1]
) -> None:
"""Only declared conditional support permits an inapplicable cell."""
conditional = {
metric.name for metric in metrics if metric.support == "partial" and metric.reason
}
if any(cell.status == "unsupported" and cell.metric not in conditional for cell in cells):
Comment thread
bokelley marked this conversation as resolved.
raise ValueError("unsupported cells require partial metric support with a reason")


class SourceBatchCoverageV1(_Frozen):
"""The slice's coverage roll-up plus its per-constituent evidence."""

Expand Down Expand Up @@ -1502,16 +1514,21 @@ def _validate_cells(self) -> None:
"metric availability must carry one record per constituent-metric cell"
)
by_constituent: dict[str, set[str]] = {}
applicable_constituents: set[str] = set()
for cell in self.metric_availability:
by_constituent.setdefault(cell.constituent_id, set()).add(cell.metric)
if cell.status != "unsupported":
applicable_constituents.add(cell.constituent_id)
published_metrics = {metric for _, metric in cells}
statuses = {item.constituent_id: item.status for item in self.coverage.constituents}
for constituent_id in statuses:
for constituent_id, status in statuses.items():
if by_constituent.get(constituent_id, set()) != published_metrics:
raise ValueError(
f"constituent {constituent_id!r} needs one availability record for every "
"published metric"
)
if status != "unsupported" and constituent_id not in applicable_constituents:
raise ValueError("a constituent without applicable metrics must be unsupported")
for cell in self.metric_availability:
constituent_status = statuses.get(cell.constituent_id)
if constituent_status is None:
Expand Down
Loading
Loading