diff --git a/docs/analytics-v2-format.md b/docs/analytics-v2-format.md index b370095b..5a388159 100644 --- a/docs/analytics-v2-format.md +++ b/docs/analytics-v2-format.md @@ -57,4 +57,12 @@ The measured-token average is `measured_tokens / measured_token_requests`, or un This unpublished format has exactly 20 columns. The earlier 18-column draft is not an accepted published format. Existing legacy public report formats remain separate inputs for consumers. +## Delivery and history + +The producer waits until every active collector has drained its queue past the reported UTC midnight. It commits the complete signed event bytes before any network send. Retries reuse those bytes across process restarts. A relay receipt requires both an affirmative `OK` for that event and readback of the complete identical signed event. A false `OK`, an acknowledgement for another ID, a forged readback with the right ID, or a missing readback does not mark delivery successful. + +The operator's configured public WSS relays are the delivery targets. The producer chooses a frame size that fits the advertised limits of the required relay quorum. Missing relay information uses the 96 KiB default. A newly discovered smaller limit can produce a corrected version with more models folded into `_other`; persisted bytes are never edited. Once a correction is stored, the report it corrects is retired from sending and its bytes are kept. Delivery requires receipts from two distinct configured relay URLs, or from the single URL when only one is configured. This proves storage redundancy across URLs, not independent relay ownership. No external manifest must exist before publication can work. Disabling sharing cancels pending sends and fences older delivery generations. Disabling cannot retract data already published publicly. + +The producer queries daily model/source/cohort groups in the database for the latest 365 completed days, expanding to the oldest week's Monday so it never truncates a weekly report. Grouping keeps eligible requests separate from missing-positive-cache requests so the measured numerator and denominator remain matched. It keeps the signed history already stored. This release does not automatically delete private records or signed reports; retention remains a separate operator/product policy decision. + The shared signed example is `tests/fixtures/analytics-v2/daily-models-signed.json`. It includes a paid model with reported usage and both cache components, a free model with estimated usage, an unknown model with missing usage, and actual zero-use days. Its three completions contain 190 recorded tokens and 3,250 msat of revenue. One completion qualifies for the measured average, with 170 tokens including 6 cache-read and 4 cache-creation tokens. diff --git a/migrations/versions/d9a6e2f4c7b1_add_analytics_v2_outbox.py b/migrations/versions/d9a6e2f4c7b1_add_analytics_v2_outbox.py new file mode 100644 index 00000000..bef60b14 --- /dev/null +++ b/migrations/versions/d9a6e2f4c7b1_add_analytics_v2_outbox.py @@ -0,0 +1,150 @@ +"""add analytics v2 outbox + +Revision ID: d9a6e2f4c7b1 +Revises: c8e4a1f2b3d5 +Create Date: 2026-08-31 00:00:00.000000 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "d9a6e2f4c7b1" +down_revision = "c8e4a1f2b3d5" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "analytics_v2_delivery_state", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("sharing_enabled", sa.Boolean(), nullable=False), + sa.Column("generation", sa.BigInteger(), nullable=False), + sa.Column("active_epoch_floor", sa.BigInteger(), nullable=True), + sa.Column("identity_pubkey", sa.String(), nullable=True), + sa.Column("provider_d", sa.String(), nullable=True), + sa.Column("updated_at_ms", sa.BigInteger(), nullable=False), + sa.CheckConstraint( + "id = 1", + name="ck_analytics_v2_delivery_state_singleton", + ), + sa.CheckConstraint( + "generation >= 0 AND updated_at_ms >= 0 " + "AND (active_epoch_floor IS NULL OR active_epoch_floor >= 0)", + name="ck_analytics_v2_delivery_state_nonnegative", + ), + sa.CheckConstraint( + "(sharing_enabled AND active_epoch_floor IS NOT NULL) OR " + "(NOT sharing_enabled AND active_epoch_floor IS NULL)", + name="ck_analytics_v2_delivery_state_epoch_floor", + ), + sa.CheckConstraint( + "(identity_pubkey IS NULL AND provider_d IS NULL) OR " + "(identity_pubkey IS NOT NULL AND provider_d IS NOT NULL)", + name="ck_analytics_v2_delivery_state_identity_pair", + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_table( + "analytics_v2_outbox", + sa.Column("event_id", sa.String(), nullable=False), + sa.Column("pubkey", sa.String(), nullable=False), + sa.Column("d_tag", sa.String(), nullable=False), + sa.Column("kind", sa.Integer(), nullable=False), + sa.Column("week", sa.Date(), nullable=False), + sa.Column("epoch", sa.BigInteger(), nullable=False), + sa.Column("through_day", sa.Date(), nullable=False), + sa.Column("semantic_slot", sa.String(), nullable=False), + sa.Column("delivery_generation", sa.BigInteger(), nullable=False), + sa.Column("frame", sa.LargeBinary(), nullable=False), + sa.Column("finalized", sa.Boolean(), nullable=False), + sa.Column("corrected", sa.Boolean(), nullable=False), + sa.Column("status", sa.String(), nullable=False), + sa.Column("created_at", sa.BigInteger(), nullable=False), + sa.Column("stored_at_ms", sa.BigInteger(), nullable=False), + sa.Column("next_attempt_at_ms", sa.BigInteger(), nullable=False), + sa.Column("attempt_count", sa.BigInteger(), nullable=False), + sa.Column("first_send_attempt_at_ms", sa.BigInteger(), nullable=True), + sa.Column("delivered_at_ms", sa.BigInteger(), nullable=True), + sa.CheckConstraint( + "kind = 38422", + name="ck_analytics_v2_outbox_kind", + ), + sa.CheckConstraint( + "status IN ('pending', 'delivered', 'superseded', 'cancelled')", + name="ck_analytics_v2_outbox_status", + ), + sa.CheckConstraint( + "epoch >= 0 AND delivery_generation >= 0 AND created_at >= 0 " + "AND stored_at_ms >= 0 AND next_attempt_at_ms >= 0 " + "AND attempt_count >= 0 " + "AND (first_send_attempt_at_ms IS NULL " + "OR first_send_attempt_at_ms >= 0) " + "AND (delivered_at_ms IS NULL OR delivered_at_ms >= 0)", + name="ck_analytics_v2_outbox_nonnegative", + ), + sa.CheckConstraint( + "length(frame) > 0", + name="ck_analytics_v2_outbox_frame_nonempty", + ), + sa.CheckConstraint( + "length(semantic_slot) > 0", + name="ck_analytics_v2_outbox_semantic_slot_nonempty", + ), + sa.CheckConstraint( + "(status = 'delivered' AND delivered_at_ms IS NOT NULL) OR " + "(status != 'delivered' AND delivered_at_ms IS NULL)", + name="ck_analytics_v2_outbox_delivery_state", + ), + sa.PrimaryKeyConstraint("event_id"), + sa.UniqueConstraint( + "pubkey", + "d_tag", + "semantic_slot", + name="uq_analytics_v2_outbox_semantic_slot", + ), + ) + op.create_index( + "ix_analytics_v2_outbox_pending", + "analytics_v2_outbox", + ["status", "delivery_generation", "next_attempt_at_ms"], + unique=False, + ) + op.create_index( + "ix_analytics_v2_outbox_coordinate", + "analytics_v2_outbox", + ["pubkey", "d_tag", "epoch", "created_at"], + unique=False, + ) + op.create_table( + "analytics_v2_relay_receipts", + sa.Column("event_id", sa.String(), nullable=False), + sa.Column("relay_url", sa.String(), nullable=False), + sa.Column("accepted_at_ms", sa.BigInteger(), nullable=False), + sa.Column("read_back_at_ms", sa.BigInteger(), nullable=False), + sa.CheckConstraint( + "accepted_at_ms >= 0 AND read_back_at_ms >= 0", + name="ck_analytics_v2_relay_receipts_nonnegative", + ), + sa.ForeignKeyConstraint( + ["event_id"], + ["analytics_v2_outbox.event_id"], + ), + sa.PrimaryKeyConstraint("event_id", "relay_url"), + ) + + +def downgrade() -> None: + op.drop_table("analytics_v2_relay_receipts") + op.drop_index( + "ix_analytics_v2_outbox_coordinate", + table_name="analytics_v2_outbox", + ) + op.drop_index( + "ix_analytics_v2_outbox_pending", + table_name="analytics_v2_outbox", + ) + op.drop_table("analytics_v2_outbox") + op.drop_table("analytics_v2_delivery_state") diff --git a/routstr/core/db.py b/routstr/core/db.py index f381e925..3a690e4f 100644 --- a/routstr/core/db.py +++ b/routstr/core/db.py @@ -18,6 +18,7 @@ from sqlalchemy import ( CheckConstraint, Date, Index, + LargeBinary, UniqueConstraint, case, delete, @@ -916,6 +917,133 @@ class TerminalOutcomeWriterRun(SQLModel, table=True): # type: ignore loss_day: date | None = Field(default=None, nullable=True, sa_type=Date) +class AnalyticsV2Outbox(SQLModel, table=True): # type: ignore + __tablename__ = "analytics_v2_outbox" + __table_args__ = ( + UniqueConstraint( + "pubkey", + "d_tag", + "semantic_slot", + name="uq_analytics_v2_outbox_semantic_slot", + ), + Index( + "ix_analytics_v2_outbox_pending", + "status", + "delivery_generation", + "next_attempt_at_ms", + ), + Index( + "ix_analytics_v2_outbox_coordinate", + "pubkey", + "d_tag", + "epoch", + "created_at", + ), + CheckConstraint( + "kind = 38422", + name="ck_analytics_v2_outbox_kind", + ), + CheckConstraint( + "status IN ('pending', 'delivered', 'superseded', 'cancelled')", + name="ck_analytics_v2_outbox_status", + ), + CheckConstraint( + "epoch >= 0 AND delivery_generation >= 0 AND created_at >= 0 " + "AND stored_at_ms >= 0 AND next_attempt_at_ms >= 0 " + "AND attempt_count >= 0 " + "AND (first_send_attempt_at_ms IS NULL " + "OR first_send_attempt_at_ms >= 0) " + "AND (delivered_at_ms IS NULL OR delivered_at_ms >= 0)", + name="ck_analytics_v2_outbox_nonnegative", + ), + CheckConstraint( + "length(frame) > 0", + name="ck_analytics_v2_outbox_frame_nonempty", + ), + CheckConstraint( + "length(semantic_slot) > 0", + name="ck_analytics_v2_outbox_semantic_slot_nonempty", + ), + CheckConstraint( + "(status = 'delivered' AND delivered_at_ms IS NOT NULL) OR " + "(status != 'delivered' AND delivered_at_ms IS NULL)", + name="ck_analytics_v2_outbox_delivery_state", + ), + ) + + event_id: str = Field(primary_key=True) + pubkey: str + d_tag: str + kind: int + week: date = Field(sa_type=Date) + epoch: int = Field(sa_type=BigInteger) + through_day: date = Field(sa_type=Date) + semantic_slot: str + delivery_generation: int = Field(sa_type=BigInteger) + frame: bytes = Field(sa_type=LargeBinary) + finalized: bool + corrected: bool + status: str = Field(default="pending") + created_at: int = Field(sa_type=BigInteger) + stored_at_ms: int = Field(sa_type=BigInteger) + next_attempt_at_ms: int = Field(sa_type=BigInteger) + attempt_count: int = Field(default=0, sa_type=BigInteger) + first_send_attempt_at_ms: int | None = Field( + default=None, nullable=True, sa_type=BigInteger + ) + delivered_at_ms: int | None = Field(default=None, nullable=True, sa_type=BigInteger) + + +class AnalyticsV2RelayReceipt(SQLModel, table=True): # type: ignore + __tablename__ = "analytics_v2_relay_receipts" + __table_args__ = ( + CheckConstraint( + "accepted_at_ms >= 0 AND read_back_at_ms >= 0", + name="ck_analytics_v2_relay_receipts_nonnegative", + ), + ) + + event_id: str = Field(primary_key=True, foreign_key="analytics_v2_outbox.event_id") + relay_url: str = Field(primary_key=True) + accepted_at_ms: int = Field(sa_type=BigInteger) + read_back_at_ms: int = Field(sa_type=BigInteger) + + +class AnalyticsV2DeliveryState(SQLModel, table=True): # type: ignore + __tablename__ = "analytics_v2_delivery_state" + __table_args__ = ( + CheckConstraint( + "id = 1", + name="ck_analytics_v2_delivery_state_singleton", + ), + CheckConstraint( + "generation >= 0 AND updated_at_ms >= 0 " + "AND (active_epoch_floor IS NULL OR active_epoch_floor >= 0)", + name="ck_analytics_v2_delivery_state_nonnegative", + ), + CheckConstraint( + "(sharing_enabled AND active_epoch_floor IS NOT NULL) OR " + "(NOT sharing_enabled AND active_epoch_floor IS NULL)", + name="ck_analytics_v2_delivery_state_epoch_floor", + ), + CheckConstraint( + "(identity_pubkey IS NULL AND provider_d IS NULL) OR " + "(identity_pubkey IS NOT NULL AND provider_d IS NOT NULL)", + name="ck_analytics_v2_delivery_state_identity_pair", + ), + ) + + id: int = Field(default=1, primary_key=True) + sharing_enabled: bool = Field(default=False) + generation: int = Field(default=0, sa_type=BigInteger) + active_epoch_floor: int | None = Field( + default=None, nullable=True, sa_type=BigInteger + ) + identity_pubkey: str | None = Field(default=None, nullable=True) + provider_d: str | None = Field(default=None, nullable=True) + updated_at_ms: int = Field(sa_type=BigInteger) + + class RoutstrFee(SQLModel, table=True): # type: ignore __tablename__ = "routstr_fees" id: int = Field(default=1, primary_key=True) diff --git a/routstr/nostr/analytics_v2_delivery.py b/routstr/nostr/analytics_v2_delivery.py new file mode 100644 index 00000000..16b85df2 --- /dev/null +++ b/routstr/nostr/analytics_v2_delivery.py @@ -0,0 +1,1696 @@ +from __future__ import annotations + +import asyncio +import ipaddress +import json +import re +import socket +import time +import unicodedata +import uuid +from collections.abc import Awaitable, Callable, Iterable, Sequence +from dataclasses import dataclass +from datetime import UTC, date, datetime, timedelta +from typing import Any, AsyncContextManager, Literal, Protocol +from urllib.parse import urlsplit, urlunsplit + +import aiohttp +import websockets +from aiohttp.abc import AbstractResolver +from sqlalchemy import and_, func +from sqlalchemy import select as sa_select +from sqlalchemy.exc import IntegrityError, OperationalError +from sqlmodel import col, or_, select, update +from sqlmodel.ext.asyncio.session import AsyncSession + +from ..core.db import ( + AnalyticsV2DeliveryState, + AnalyticsV2Outbox, + AnalyticsV2RelayReceipt, + TerminalOutcome, + TerminalOutcomeEpoch, + TerminalOutcomeWriterRun, +) +from ..core.logging import get_logger +from .analytics_v2 import ( + ANALYTICS_KIND, + DEFAULT_MAX_FRAME_BYTES, + EncodedAnalyticsEvent, + LedgerOutcome, + PriorVersion, + WeeklyAggregate, + aggregate_ledger_week, + build_analytics_address, + daily_models_changed, + encode_week_event, + prior_version_from_frame, +) + +logger = get_logger(__name__) + +ANALYTICS_RELAY_QUORUM = 2 +PUBLIC_HISTORY_DAYS = 365 +NIP11_MAX_DOCUMENT_BYTES = 65_536 + +SessionFactory = Callable[[], AsyncContextManager[AsyncSession]] +IdentityClaim = Literal["initialized", "matched", "mismatch"] + + +class AnalyticsV2DeliveryError(RuntimeError): + pass + + +class SharingDisabledError(AnalyticsV2DeliveryError): + pass + + +class IdentityMismatchError(AnalyticsV2DeliveryError): + pass + + +class OutboxConflictError(AnalyticsV2DeliveryError): + pass + + +def _require_lower_hex(value: object, length: int, name: str) -> None: + if ( + not isinstance(value, str) + or len(value) != length + or value != value.lower() + or any(character not in "0123456789abcdef" for character in value) + ): + raise AnalyticsV2DeliveryError( + f"{name} must be {length} lowercase hex characters" + ) + + +def _normalize_public_wss_url(value: str) -> str: + if not isinstance(value, str) or value != value.strip() or not value: + raise AnalyticsV2DeliveryError("Relay URL is empty or not normalized") + try: + parsed = urlsplit(value) + port = parsed.port + except ValueError as error: + raise AnalyticsV2DeliveryError("Relay URL has an invalid port") from error + if parsed.scheme.lower() != "wss" or not parsed.hostname: + raise AnalyticsV2DeliveryError("Relay URL must use wss") + if parsed.username is not None or parsed.password is not None: + raise AnalyticsV2DeliveryError("Relay URL must not contain credentials") + if parsed.query or parsed.fragment: + raise AnalyticsV2DeliveryError("Relay URL must not contain query credentials") + + host = parsed.hostname.rstrip(".").lower() + if not host: + raise AnalyticsV2DeliveryError("Relay URL hostname is empty") + try: + address = ipaddress.ip_address(host) + except ValueError: + if ( + "." not in host + or host.endswith( + ( + ".local", + ".localhost", + ".internal", + ".home", + ".lan", + ".test", + ".invalid", + ".example", + ) + ) + or re.fullmatch( + r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?" + r"(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+", + host, + ) + is None + ): + raise AnalyticsV2DeliveryError("Relay hostname is not public") + rendered_host = host + else: + if not address.is_global: + raise AnalyticsV2DeliveryError("Relay IP address is not public") + rendered_host = f"[{host}]" if address.version == 6 else host + + if port is not None and not 1 <= port <= 65535: + raise AnalyticsV2DeliveryError("Relay URL port is invalid") + netloc = rendered_host if port in {None, 443} else f"{rendered_host}:{port}" + path = "" if parsed.path in {"", "/"} else parsed.path + return urlunsplit(("wss", netloc, path, "", "")) + + +@dataclass(frozen=True) +class RelayTarget: + url: str + + +RelayLimitReader = Callable[[RelayTarget, float], Awaitable[int | None]] + + +@dataclass(frozen=True) +class ResolvedRelayEndpoint: + address: str + port: int + server_hostname: str + + +@dataclass(frozen=True) +class DeliveryStateSnapshot: + sharing_enabled: bool + generation: int + identity_pubkey: str | None + provider_d: str | None + updated_at_ms: int + active_epoch_floor: int | None = None + + +@dataclass(frozen=True) +class ActivationResult: + state: DeliveryStateSnapshot + transitioned: bool + + +@dataclass(frozen=True) +class EnqueueResult: + event_id: str + inserted: bool + created_at: int + + +@dataclass(frozen=True) +class RelaySendResult: + accepted: bool + read_back: bool + + +@dataclass(frozen=True) +class DeliveryPassResult: + attempted_events: int + delivered_events: int + + +@dataclass(frozen=True) +class _PendingFrame: + event_id: str + frame: bytes + delivery_generation: int + + +class RelaySender(Protocol): + async def __call__( + self, + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: ... + + +class _PinnedRelayResolver(AbstractResolver): + def __init__(self, endpoint: ResolvedRelayEndpoint) -> None: + self._endpoint = endpoint + + async def resolve( + self, + host: str, + port: int = 0, + family: socket.AddressFamily = socket.AF_INET, + ) -> list[Any]: + if host != self._endpoint.server_hostname: + raise OSError("Relay information lookup changed hostname") + address = ipaddress.ip_address(self._endpoint.address) + return [ + { + "hostname": host, + "host": address.compressed, + "port": self._endpoint.port, + "family": (socket.AF_INET6 if address.version == 6 else socket.AF_INET), + "proto": socket.IPPROTO_TCP, + "flags": socket.AI_NUMERICHOST, + } + ] + + async def close(self) -> None: + return None + + +async def fetch_relay_max_message_length( + target: RelayTarget, timeout_seconds: float +) -> int | None: + """Read an advertised NIP-11 frame limit through the pinned public endpoint.""" + try: + endpoint = await resolve_public_relay_endpoint(target.url) + parsed = urlsplit(target.url) + relay_information_url = urlunsplit( + ("https", parsed.netloc, parsed.path or "/", "", "") + ) + connector = aiohttp.TCPConnector( + resolver=_PinnedRelayResolver(endpoint), + use_dns_cache=False, + force_close=True, + ) + timeout = aiohttp.ClientTimeout(total=timeout_seconds) + async with aiohttp.ClientSession( + connector=connector, timeout=timeout + ) as session: + async with session.get( + relay_information_url, + headers={"Accept": "application/nostr+json"}, + allow_redirects=False, + ) as response: + if response.status != 200: + return None + if ( + response.content_length is not None + and response.content_length > NIP11_MAX_DOCUMENT_BYTES + ): + return None + body = await response.content.read(NIP11_MAX_DOCUMENT_BYTES + 1) + if len(body) > NIP11_MAX_DOCUMENT_BYTES: + return None + payload = json.loads(body) + if not isinstance(payload, dict): + return None + limitation = payload.get("limitation") + if not isinstance(limitation, dict): + return None + maximum = limitation.get("max_message_length") + if type(maximum) is int and maximum > 0: + return maximum + except asyncio.CancelledError: + raise + except Exception: + return None + return None + + +async def get_analytics_v2_delivery_state( + session_factory: SessionFactory, + *, + at_ms: int | None = None, +) -> DeliveryStateSnapshot: + timestamp = _clock_ms() if at_ms is None else at_ms + _require_nonnegative_int(timestamp, "at_ms") + async with session_factory() as session: + state = await _get_or_create_delivery_state(session, timestamp) + snapshot = _snapshot(state) + await session.commit() + return snapshot + + +async def claim_analytics_v2_identity( + session_factory: SessionFactory, + *, + pubkey: str, + provider_d: str, + at_ms: int | None = None, +) -> IdentityClaim: + _validate_identity(pubkey, provider_d) + timestamp = _clock_ms() if at_ms is None else at_ms + _require_nonnegative_int(timestamp, "at_ms") + while True: + async with session_factory() as session: + try: + state = await _get_or_create_delivery_state(session, timestamp) + claimed = await session.exec( # type: ignore[call-overload] + update(AnalyticsV2DeliveryState) + .where(col(AnalyticsV2DeliveryState.id) == 1) + .where(col(AnalyticsV2DeliveryState.identity_pubkey).is_(None)) + .where(col(AnalyticsV2DeliveryState.provider_d).is_(None)) + .values( + identity_pubkey=pubkey, + provider_d=provider_d, + updated_at_ms=timestamp, + ) + ) + if claimed.rowcount == 1: + await session.commit() + return "initialized" + await session.refresh(state) + result: IdentityClaim = ( + "matched" + if state.identity_pubkey == pubkey + and state.provider_d == provider_d + else "mismatch" + ) + await session.commit() + return result + except (IntegrityError, OperationalError): + await session.rollback() + await asyncio.sleep(0) + + +async def transition_analytics_v2_sharing( + session_factory: SessionFactory, + *, + enabled: bool, + at_ms: int | None = None, +) -> DeliveryStateSnapshot: + if not isinstance(enabled, bool): + raise AnalyticsV2DeliveryError("enabled must be a boolean") + timestamp = _clock_ms() if at_ms is None else at_ms + _require_nonnegative_int(timestamp, "at_ms") + while True: + async with session_factory() as session: + try: + state = await _get_or_create_delivery_state(session, timestamp) + if enabled and not state.sharing_enabled: + raise AnalyticsV2DeliveryError( + "Use activate_analytics_v2_sharing to enable publication" + ) + if state.sharing_enabled != enabled: + # Private collection keeps its coverage; activation sets the + # next public floor. + state.sharing_enabled = enabled + state.generation += 1 + state.active_epoch_floor = None + state.updated_at_ms = timestamp + if not enabled: + await session.exec( # type: ignore[call-overload] + update(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.status) == "pending") + .values(status="cancelled") + ) + snapshot = _snapshot(state) + await session.commit() + return snapshot + except OperationalError: + await session.rollback() + await asyncio.sleep(0) + + +async def fence_analytics_v2_opt_out(session: AsyncSession) -> None: + """Disable sharing inside the caller's transaction. + + Committed together with a saved opt-out, the new generation refuses any + activation that read its flags before the save. + """ + await session.exec( # type: ignore[call-overload] + update(AnalyticsV2DeliveryState) + .where(col(AnalyticsV2DeliveryState.id) == 1) + .values( + sharing_enabled=False, + generation=AnalyticsV2DeliveryState.generation + 1, + active_epoch_floor=None, + updated_at_ms=_clock_ms(), + ) + ) + await session.exec( # type: ignore[call-overload] + update(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.status) == "pending") + .values(status="cancelled") + ) + + +async def activate_analytics_v2_sharing( + session_factory: SessionFactory, + *, + coverage_day: date, + at_ms: int | None = None, + expected_generation: int | None = None, +) -> ActivationResult: + """Rotate continuity and expose a false-to-true transition atomically. + + ``expected_generation`` refuses a caller whose flags predate an opt-out. + """ + if type(coverage_day) is not date: + raise AnalyticsV2DeliveryError("coverage_day must be a date") + timestamp = _clock_ms() if at_ms is None else at_ms + _require_nonnegative_int(timestamp, "at_ms") + + while True: + async with session_factory() as session: + state_result = await session.exec( + select(AnalyticsV2DeliveryState) + .where(col(AnalyticsV2DeliveryState.id) == 1) + .with_for_update() + ) + state = state_result.first() + if state is None: + state = AnalyticsV2DeliveryState(id=1, updated_at_ms=timestamp) + session.add(state) + try: + await session.flush() + except (IntegrityError, OperationalError): + await session.rollback() + await asyncio.sleep(0) + continue + if state.sharing_enabled: + snapshot = _snapshot(state) + await session.commit() + return ActivationResult(snapshot, False) + if expected_generation not in (None, state.generation): + raise SharingDisabledError("Analytics v2 sharing changed meanwhile") + if state.identity_pubkey is None or state.provider_d is None: + raise IdentityMismatchError( + "Analytics v2 identity must be claimed before activation" + ) + + epoch_result = await session.exec( + select(TerminalOutcomeEpoch) + .where(col(TerminalOutcomeEpoch.current_slot) == 1) + .with_for_update() + ) + current = epoch_result.first() + if current is None: + latest_result = await session.exec( + select(TerminalOutcomeEpoch).order_by( + col(TerminalOutcomeEpoch.epoch).desc() + ) + ) + latest = latest_result.first() + next_epoch = 0 if latest is None else latest.epoch + 1 + current = TerminalOutcomeEpoch( + epoch=next_epoch, + coverage_start_day=coverage_day + timedelta(days=1), + current_slot=1, + ) + session.add(current) + try: + await session.flush() + except (IntegrityError, OperationalError): + await session.rollback() + await asyncio.sleep(0) + continue + elif current.coverage_start_day != coverage_day + timedelta(days=1): + try: + await _close_current_epoch( + session, + coverage_end_day=coverage_day - timedelta(days=1), + ) + except OperationalError: + await session.rollback() + await asyncio.sleep(0) + continue + latest_result = await session.exec( + select(TerminalOutcomeEpoch).order_by( + col(TerminalOutcomeEpoch.epoch).desc() + ) + ) + latest = latest_result.first() + next_epoch = 0 if latest is None else latest.epoch + 1 + current = TerminalOutcomeEpoch( + epoch=next_epoch, + coverage_start_day=coverage_day + timedelta(days=1), + current_slot=1, + ) + session.add(current) + try: + await session.flush() + except (IntegrityError, OperationalError): + await session.rollback() + await asyncio.sleep(0) + continue + await session.exec( # type: ignore[call-overload] + update(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.status) == "pending") + .values(status="cancelled") + ) + # Compare-and-set: SQLite ignores FOR UPDATE, so the row read above + # may already be stale. + claimed = await session.exec( # type: ignore[call-overload] + update(AnalyticsV2DeliveryState) + .where(col(AnalyticsV2DeliveryState.id) == 1) + .where(col(AnalyticsV2DeliveryState.sharing_enabled).is_(False)) + .where(col(AnalyticsV2DeliveryState.generation) == state.generation) + .values( + sharing_enabled=True, + generation=state.generation + 1, + active_epoch_floor=current.epoch, + updated_at_ms=timestamp, + ) + ) + if claimed.rowcount != 1: + await session.rollback() + await asyncio.sleep(0) + continue + await session.refresh(state) + snapshot = _snapshot(state) + try: + await session.commit() + except (IntegrityError, OperationalError): + await session.rollback() + await asyncio.sleep(0) + continue + return ActivationResult(snapshot, True) + + +async def rotate_analytics_v2_identity( + session_factory: SessionFactory, + *, + pubkey: str, + provider_d: str, + at_ms: int | None = None, +) -> DeliveryStateSnapshot: + """Bind a changed identity behind a disabled delivery generation.""" + _validate_identity(pubkey, provider_d) + timestamp = _clock_ms() if at_ms is None else at_ms + _require_nonnegative_int(timestamp, "at_ms") + async with session_factory() as session: + state = await _get_or_create_delivery_state(session, timestamp) + if state.identity_pubkey != pubkey or state.provider_d != provider_d: + state.sharing_enabled = False + state.generation += 1 + state.active_epoch_floor = None + state.identity_pubkey = pubkey + state.provider_d = provider_d + state.updated_at_ms = timestamp + await session.exec( # type: ignore[call-overload] + update(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.status) == "pending") + .values(status="cancelled") + ) + snapshot = _snapshot(state) + await session.commit() + return snapshot + + +async def enqueue_signed_event( + session_factory: SessionFactory, + event: EncodedAnalyticsEvent, + *, + stored_at_ms: int | None = None, +) -> EnqueueResult: + """Commit exact signed bytes and coalesce older pending open versions.""" + timestamp = _clock_ms() if stored_at_ms is None else stored_at_ms + _require_nonnegative_int(timestamp, "stored_at_ms") + parsed = prior_version_from_frame(event.frame) + _validate_encoded_metadata(event, parsed) + semantic_slot = _semantic_slot(event) + + async with session_factory() as session: + state = await _get_or_create_delivery_state(session, timestamp) + if not state.sharing_enabled: + raise SharingDisabledError("Analytics v2 sharing is disabled") + if state.active_epoch_floor is None or event.epoch < state.active_epoch_floor: + raise SharingDisabledError( + "Analytics v2 event is outside the active epoch lane" + ) + if ( + state.identity_pubkey != event.pubkey + or state.provider_d is None + or build_analytics_address( + event.pubkey, state.provider_d, event.week, event.epoch + ).d_tag + != event.d_tag + ): + raise IdentityMismatchError( + "Analytics v2 identity does not match the event" + ) + generation = state.generation + + existing = await session.get(AnalyticsV2Outbox, event.event_id) + if existing is not None: + _require_exact_outbox(existing, event, generation, semantic_slot) + await session.commit() + return EnqueueResult(event.event_id, False, existing.created_at) + + winner_result = await session.exec( + select(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.pubkey) == event.pubkey) + .where(col(AnalyticsV2Outbox.d_tag) == event.d_tag) + .where(col(AnalyticsV2Outbox.semantic_slot) == semantic_slot) + ) + winner = winner_result.first() + if winner is not None: + try: + _require_semantic_winner(winner, event, generation, semantic_slot) + except OutboxConflictError: + if ( + winner.status != "pending" + or winner.first_send_attempt_at_ms is not None + ): + raise + winner.status = "superseded" + winner.semantic_slot = f"superseded:{winner.event_id}" + await session.flush() + else: + await session.commit() + return EnqueueResult(winner.event_id, False, winner.created_at) + + await session.exec( # type: ignore[call-overload] + update(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.pubkey) == event.pubkey) + .where(col(AnalyticsV2Outbox.d_tag) == event.d_tag) + .where(col(AnalyticsV2Outbox.epoch) == event.epoch) + .where(col(AnalyticsV2Outbox.delivery_generation) == state.generation) + .where(col(AnalyticsV2Outbox.status) == "pending") + .where(col(AnalyticsV2Outbox.semantic_slot) != semantic_slot) + .where( + or_( + col(AnalyticsV2Outbox.finalized).is_(False), + col(AnalyticsV2Outbox.first_send_attempt_at_ms).is_(None), + # A correction replaces the exact report it names. + col(AnalyticsV2Outbox.event_id) + == semantic_slot.removeprefix("correction:"), + ) + ) + .values(status="superseded") + ) + row = AnalyticsV2Outbox( + event_id=event.event_id, + pubkey=event.pubkey, + d_tag=event.d_tag, + kind=ANALYTICS_KIND, + week=event.week, + epoch=event.epoch, + through_day=event.through, + semantic_slot=semantic_slot, + delivery_generation=generation, + frame=event.frame, + finalized=event.complete, + corrected=event.corrected, + status="pending", + created_at=event.created_at, + stored_at_ms=timestamp, + next_attempt_at_ms=timestamp, + attempt_count=0, + ) + session.add(row) + try: + await session.commit() + except IntegrityError: + await session.rollback() + existing = await session.get(AnalyticsV2Outbox, event.event_id) + if existing is not None: + _require_exact_outbox(existing, event, generation, semantic_slot) + return EnqueueResult(event.event_id, False, existing.created_at) + winner_result = await session.exec( + select(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.pubkey) == event.pubkey) + .where(col(AnalyticsV2Outbox.d_tag) == event.d_tag) + .where(col(AnalyticsV2Outbox.semantic_slot) == semantic_slot) + ) + winner = winner_result.first() + if winner is None: + raise + _require_semantic_winner(winner, event, generation, semantic_slot) + return EnqueueResult(winner.event_id, False, winner.created_at) + return EnqueueResult(event.event_id, True, event.created_at) + + +class AnalyticsV2Producer: + """Turn the current continuity epoch into durable weekly signed events.""" + + def __init__( + self, + session_factory: SessionFactory, + *, + private_key_hex: str, + public_key_hex: str, + provider_d: str, + max_frame_bytes: int = DEFAULT_MAX_FRAME_BYTES, + ) -> None: + _validate_identity(public_key_hex, provider_d) + self._session_factory = session_factory + self._private_key_hex = private_key_hex + self._public_key_hex = public_key_hex + self._provider_d = provider_d + self._max_frame_bytes = max_frame_bytes + self._lock = asyncio.Lock() + + @property + def session_factory(self) -> SessionFactory: + return self._session_factory + + async def produce_once( + self, *, now: datetime | None = None, max_frame_bytes: int | None = None + ) -> int: + frame_limit = min( + self._max_frame_bytes, max_frame_bytes or self._max_frame_bytes + ) + instant = datetime.now(UTC) if now is None else now + if instant.tzinfo is None: + raise AnalyticsV2DeliveryError("Producer time must be timezone-aware") + instant = instant.astimezone(UTC) + timestamp_ms = int(instant.timestamp() * 1000) + async with self._lock: + state = await get_analytics_v2_delivery_state( + self._session_factory, at_ms=timestamp_ms + ) + if not state.sharing_enabled: + return 0 + if ( + state.identity_pubkey != self._public_key_hex + or state.provider_d != self._provider_d + ): + raise IdentityMismatchError("Analytics v2 producer identity changed") + + if state.active_epoch_floor is None: + return 0 + completed_day_limit = await self._completed_day_limit(instant.date()) + epochs, outcomes = await self._load_eligible_epochs( + state.active_epoch_floor, completed_day_limit + ) + if not epochs: + return 0 + latest, created_at_floors, rows_by_id = await self._load_version_state( + state.active_epoch_floor, state.generation + ) + produced = 0 + yesterday = completed_day_limit - timedelta(days=1) + # Each week aggregates only its own rows instead of rescanning all. + by_week: dict[date, list[LedgerOutcome]] = {} + for outcome in outcomes: + day = outcome.terminal_day + by_week.setdefault(day - timedelta(days=day.weekday()), []).append( + outcome + ) + for epoch in epochs: + coverage_end = min(epoch.coverage_end_day or yesterday, yesterday) + cutoff = instant.date() - timedelta(days=PUBLIC_HISTORY_DAYS) + first_week = cutoff - timedelta(days=cutoff.weekday()) + for week in _covered_weeks( + max(epoch.coverage_start_day, first_week), coverage_end + ): + address = build_analytics_address( + self._public_key_hex, self._provider_d, week, epoch.epoch + ) + prior_row = latest.get((epoch.epoch, address.d_tag)) + aggregate = self._next_aggregate( + by_week.get(week, ()), + epoch=epoch, + week=week, + today_utc=completed_day_limit, + prior_row=prior_row, + rows_by_id=rows_by_id, + max_frame_bytes=frame_limit, + ) + if aggregate is None: + continue + created_at = max( + int(instant.timestamp()), + created_at_floors.get(address.d_tag, -1) + 1, + ) + encoded = encode_week_event( + aggregate, + private_key_hex=self._private_key_hex, + provider_d=self._provider_d, + created_at=created_at, + max_frame_bytes=frame_limit, + ) + if encoded.pubkey != self._public_key_hex: + raise IdentityMismatchError( + "Analytics v2 private key does not match the claimed pubkey" + ) + enqueue_result = await enqueue_signed_event( + self._session_factory, encoded, stored_at_ms=timestamp_ms + ) + created_at_floors[address.d_tag] = max( + created_at_floors.get(address.d_tag, -1), + enqueue_result.created_at, + ) + produced += int(enqueue_result.inserted) + return produced + + async def _completed_day_limit(self, today_utc: date) -> date: + async with self._session_factory() as session: + result = await session.exec( + select(TerminalOutcomeWriterRun).where( + col(TerminalOutcomeWriterRun.status).in_( + ("active", "degraded", "lost") + ) + ) + ) + for run in result.all(): + checkpoint = run.flushed_through_ms or run.started_at_ms + day = _utc_day_from_ms(checkpoint) + if run.loss_day is not None: + day = min(day, run.loss_day) + today_utc = min(today_utc, day) + return today_utc + + async def _load_eligible_epochs( + self, active_epoch_floor: int, today_utc: date + ) -> tuple[tuple[TerminalOutcomeEpoch, ...], tuple[LedgerOutcome, ...]]: + yesterday = today_utc - timedelta(days=1) + async with self._session_factory() as session: + result = await session.exec( + select(TerminalOutcomeEpoch) + .where(col(TerminalOutcomeEpoch.epoch) >= active_epoch_floor) + .order_by(col(TerminalOutcomeEpoch.epoch)) + ) + epochs = tuple( + epoch + for epoch in result.all() + if epoch.coverage_start_day + <= min(epoch.coverage_end_day or yesterday, yesterday) + ) + if not epochs: + return (), () + cutoff = today_utc - timedelta(days=PUBLIC_HISTORY_DAYS) + first_week = cutoff - timedelta(days=cutoff.weekday()) + start = max(min(epoch.coverage_start_day for epoch in epochs), first_week) + sources = ( + col(TerminalOutcome.input_source), + col(TerminalOutcome.output_source), + col(TerminalOutcome.cache_read_source), + col(TerminalOutcome.cache_creation_source), + ) + measured = and_( + sources[0] == "reported", + sources[1] == "reported", + or_( + sources[2] == "reported", + and_( + sources[2] == "missing", + col(TerminalOutcome.cache_read_input_tokens) == 0, + ), + ), + or_( + sources[3] == "reported", + and_( + sources[3] == "missing", + col(TerminalOutcome.cache_creation_input_tokens) == 0, + ), + ), + ) + totals = await session.execute( + sa_select( + col(TerminalOutcome.terminal_day), + col(TerminalOutcome.model_identifier), + *sources, + func.count(col(TerminalOutcome.outcome_id)), + func.sum(TerminalOutcome.input_tokens), + func.sum(TerminalOutcome.output_tokens), + func.sum(TerminalOutcome.cache_read_input_tokens), + func.sum(TerminalOutcome.cache_creation_input_tokens), + func.sum(TerminalOutcome.revenue_msats), + ) + .where(col(TerminalOutcome.terminal_day) >= start) + .where(col(TerminalOutcome.terminal_day) <= yesterday) + .group_by( + col(TerminalOutcome.terminal_day), + col(TerminalOutcome.model_identifier), + *sources, + # Missing positive cache counts cannot contaminate eligible rows. + measured, + ) + ) + rows = tuple( + LedgerOutcome( + terminal_day=day, + model_identifier=model, + # PostgreSQL returns SUM(bigint) as Decimal. + input_tokens=int(input_tokens), + output_tokens=int(output_tokens), + cache_read_input_tokens=int(cache_read_tokens), + cache_creation_input_tokens=int(cache_creation_tokens), + revenue_msats=int(revenue), + input_source=input_source, + output_source=output_source, + cache_read_source=cache_read_source, + cache_creation_source=cache_creation_source, + completed_requests=count, + ) + for day, model, input_source, output_source, cache_read_source, cache_creation_source, count, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, revenue in totals.all() + ) + return epochs, rows + + async def _load_version_state( + self, active_epoch_floor: int, generation: int + ) -> tuple[ + dict[tuple[int, str], AnalyticsV2Outbox], + dict[str, int], + dict[str, AnalyticsV2Outbox], + ]: + async with self._session_factory() as session: + result = await session.exec( + select(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.pubkey) == self._public_key_hex) + .order_by( + col(AnalyticsV2Outbox.created_at).desc(), + col(AnalyticsV2Outbox.event_id), + ) + ) + rows = result.all() + latest: dict[tuple[int, str], AnalyticsV2Outbox] = {} + created_at_floors: dict[str, int] = {} + rows_by_id: dict[str, AnalyticsV2Outbox] = {} + for row in rows: + created_at_floors.setdefault(row.d_tag, row.created_at) + rows_by_id[row.event_id] = row + if ( + row.epoch >= active_epoch_floor + and row.delivery_generation == generation + ): + latest.setdefault((row.epoch, row.d_tag), row) + return latest, created_at_floors, rows_by_id + + def _next_aggregate( + self, + outcomes: Sequence[LedgerOutcome], + *, + epoch: TerminalOutcomeEpoch, + week: date, + today_utc: date, + prior_row: AnalyticsV2Outbox | None, + rows_by_id: dict[str, AnalyticsV2Outbox], + max_frame_bytes: int, + ) -> WeeklyAggregate | None: + fresh = aggregate_ledger_week( + outcomes, + epoch=epoch.epoch, + epoch_coverage_start=epoch.coverage_start_day, + epoch_coverage_end=epoch.coverage_end_day, + week=week, + today_utc=today_utc, + ) + if fresh is None or prior_row is None: + return fresh + + latest = prior_version_from_frame(bytes(prior_row.frame)) + if fresh.through < latest.through: + return None + if ( + not _published_days_changed(fresh, latest) + and fresh.through == latest.through + and fresh.complete == latest.complete + and len(prior_row.frame) <= max_frame_bytes + ): + return None + published_row = ( + prior_row + if prior_row.first_send_attempt_at_ms is not None + else _latest_attempted_predecessor(prior_row, rows_by_id.values()) + ) + if published_row is None: + return fresh + prior = prior_version_from_frame(bytes(published_row.frame)) + return aggregate_ledger_week( + outcomes, + epoch=epoch.epoch, + epoch_coverage_start=epoch.coverage_start_day, + epoch_coverage_end=epoch.coverage_end_day, + week=week, + today_utc=today_utc, + prior_version=prior, + correction=( + _published_days_changed(fresh, prior) + or fresh.through == prior.through + or len(published_row.frame) > max_frame_bytes + ), + ) + + +class AnalyticsV2Delivery: + """Deliver persisted frames with relay acceptance and exact event readback.""" + + def __init__( + self, + session_factory: SessionFactory, + *, + operator_relays: Sequence[str], + retry_seconds: float = 60.0, + timeout_seconds: float = 10.0, + sender: RelaySender | None = None, + relay_limit_reader: RelayLimitReader = fetch_relay_max_message_length, + ) -> None: + if retry_seconds < 0 or timeout_seconds <= 0: + raise AnalyticsV2DeliveryError("Delivery timing is invalid") + self._session_factory = session_factory + relays = sorted({_normalize_public_wss_url(url) for url in operator_relays}) + if not relays: + raise AnalyticsV2DeliveryError("At least one analytics relay is required") + self._targets = tuple(RelayTarget(url) for url in relays) + self._quorum = min(ANALYTICS_RELAY_QUORUM, len(self._targets)) + self._retry_ms = int(retry_seconds * 1000) + self._timeout_seconds = timeout_seconds + self._sender = sender or self._send_to_relay + self._relay_limit_reader = relay_limit_reader + self._accepting = True + self._active: set[asyncio.Task[RelaySendResult]] = set() + self._relay_limits: dict[str, int | None] = {} + + async def frame_limit(self) -> int: + self._relay_limits.clear() + await self._load_relay_limits(self._targets) + limits = sorted( + ( + limit if limit is not None else DEFAULT_MAX_FRAME_BYTES + for limit in self._relay_limits.values() + ), + reverse=True, + ) + return min(DEFAULT_MAX_FRAME_BYTES, limits[self._quorum - 1]) + + async def _load_relay_limits(self, targets: Sequence[RelayTarget]) -> None: + async def read_limit(target: RelayTarget) -> int | None: + try: + value = await self._relay_limit_reader(target, self._timeout_seconds) + return value if type(value) is int and value > 0 else None + except asyncio.CancelledError: + raise + except Exception: + return None + + missing = [target for target in targets if target.url not in self._relay_limits] + limits = await asyncio.gather(*(read_limit(target) for target in missing)) + self._relay_limits.update( + (target.url, limit) for target, limit in zip(missing, limits) + ) + + async def deliver_pending_once( + self, *, at_ms: int | None = None + ) -> DeliveryPassResult: + timestamp = _clock_ms() if at_ms is None else at_ms + _require_nonnegative_int(timestamp, "at_ms") + if not self._accepting: + return DeliveryPassResult(0, 0) + state = await get_analytics_v2_delivery_state( + self._session_factory, at_ms=timestamp + ) + if not state.sharing_enabled: + return DeliveryPassResult(0, 0) + + async with self._session_factory() as session: + result = await session.exec( + select(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.status) == "pending") + .where(col(AnalyticsV2Outbox.delivery_generation) == state.generation) + .where(col(AnalyticsV2Outbox.next_attempt_at_ms) <= timestamp) + .order_by( + col(AnalyticsV2Outbox.created_at), + col(AnalyticsV2Outbox.event_id), + ) + ) + event_ids = [row.event_id for row in result.all()] + + attempted = 0 + delivered = 0 + for event_id in event_ids: + if not self._accepting: + break + outcome = await self._deliver_event(event_id, timestamp) + if outcome is None: + continue + attempted += 1 + delivered += int(outcome) + self._relay_limits.clear() + return DeliveryPassResult(attempted, delivered) + + async def disable(self, *, at_ms: int | None = None) -> DeliveryStateSnapshot: + self._accepting = False + await self._cancel_active() + return await transition_analytics_v2_sharing( + self._session_factory, enabled=False, at_ms=at_ms + ) + + async def resume(self) -> bool: + state = await get_analytics_v2_delivery_state(self._session_factory) + self._accepting = state.sharing_enabled + return self._accepting + + async def stop(self) -> None: + self._accepting = False + await self._cancel_active() + + async def _cancel_active(self) -> None: + tasks = tuple(self._active) + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + async def _deliver_event(self, event_id: str, timestamp: int) -> bool | None: + pending = await self._load_pending_frame(event_id) + if pending is None: + return None + targets = self._targets + async with self._session_factory() as session: + receipt_result = await session.exec( + select(AnalyticsV2RelayReceipt.relay_url).where( + col(AnalyticsV2RelayReceipt.event_id) == event_id + ) + ) + completed_urls = set(receipt_result.all()) + pending_targets = [ + target for target in targets if target.url not in completed_urls + ] + eligible_targets: Sequence[RelayTarget] = () + if pending_targets: + eligible_targets = await self._eligible_targets( + pending_targets, pending.frame + ) + if not eligible_targets: + reconciled = await self._finish_attempt(event_id, _clock_ms()) + return True if reconciled else None + row = await self._begin_attempt(event_id, timestamp) + if row is None or row != pending: + return None + + async def is_active() -> bool: + return self._accepting and await self._attempt_is_active(row) + + async def send(target: RelayTarget) -> RelaySendResult: + if not await is_active(): + return RelaySendResult(False, False) + return await self._sender(target, event_id, bytes(row.frame), is_active) + + tasks: list[tuple[RelayTarget, asyncio.Task[RelaySendResult]]] = [] + for target in eligible_targets: + task: asyncio.Task[RelaySendResult] = asyncio.create_task(send(target)) + self._active.add(task) + task.add_done_callback(self._active.discard) + tasks.append((target, task)) + try: + results = await asyncio.gather( + *(task for _, task in tasks), return_exceptions=True + ) + except asyncio.CancelledError: + for _, task in tasks: + task.cancel() + raise + if not self._accepting or not await self._attempt_is_active(row): + return None + for (target, _), send_result in zip(tasks, results): + if ( + isinstance(send_result, RelaySendResult) + and send_result.accepted + and send_result.read_back + ): + await self._record_receipt(event_id, target, _clock_ms()) + return await self._finish_attempt(event_id, _clock_ms()) + + async def _load_pending_frame(self, event_id: str) -> _PendingFrame | None: + async with self._session_factory() as session: + state = await session.get(AnalyticsV2DeliveryState, 1) + row = await session.get(AnalyticsV2Outbox, event_id) + if ( + not self._accepting + or state is None + or not state.sharing_enabled + or row is None + or row.status != "pending" + or row.delivery_generation != state.generation + ): + return None + return _PendingFrame(row.event_id, bytes(row.frame), row.delivery_generation) + + async def _eligible_targets( + self, targets: Sequence[RelayTarget], frame: bytes + ) -> tuple[RelayTarget, ...]: + await self._load_relay_limits(targets) + return tuple( + target + for target in targets + if len(frame) <= (self._relay_limits[target.url] or DEFAULT_MAX_FRAME_BYTES) + ) + + async def _begin_attempt( + self, event_id: str, timestamp: int + ) -> _PendingFrame | None: + async with self._session_factory() as session: + state = await _get_or_create_delivery_state(session, timestamp) + row = await session.get(AnalyticsV2Outbox, event_id) + if ( + not self._accepting + or not state.sharing_enabled + or row is None + or row.status != "pending" + or row.delivery_generation != state.generation + ): + await session.commit() + return None + row.attempt_count += 1 + row.next_attempt_at_ms = timestamp + self._retry_ms + if row.first_send_attempt_at_ms is None: + row.first_send_attempt_at_ms = timestamp + pending = _PendingFrame( + row.event_id, bytes(row.frame), row.delivery_generation + ) + await session.commit() + return pending + + async def _attempt_is_active(self, pending: _PendingFrame) -> bool: + async with self._session_factory() as session: + state = await session.get(AnalyticsV2DeliveryState, 1) + row = await session.get(AnalyticsV2Outbox, pending.event_id) + return bool( + state is not None + and state.sharing_enabled + and state.generation == pending.delivery_generation + and row is not None + and row.status == "pending" + and row.delivery_generation == pending.delivery_generation + ) + + async def _record_receipt( + self, event_id: str, target: RelayTarget, timestamp: int + ) -> None: + key = (event_id, target.url) + async with self._session_factory() as session: + receipt = await session.get(AnalyticsV2RelayReceipt, key) + if receipt is None: + session.add( + AnalyticsV2RelayReceipt( + event_id=event_id, + relay_url=target.url, + accepted_at_ms=timestamp, + read_back_at_ms=timestamp, + ) + ) + try: + await session.commit() + except IntegrityError: + await session.rollback() + + async def _finish_attempt(self, event_id: str, timestamp: int) -> bool: + async with self._session_factory() as session: + state = await session.get(AnalyticsV2DeliveryState, 1) + row = await session.get(AnalyticsV2Outbox, event_id) + result = await session.exec( + select(AnalyticsV2RelayReceipt.relay_url).where( + col(AnalyticsV2RelayReceipt.event_id) == event_id + ) + ) + accepted_urls = set(result.all()) & {target.url for target in self._targets} + healthy = bool( + state is not None + and state.sharing_enabled + and row is not None + and row.status == "pending" + and row.delivery_generation == state.generation + and len(accepted_urls) >= self._quorum + ) + if row is not None and row.status == "pending" and healthy: + row.status = "delivered" + row.delivered_at_ms = timestamp + await session.commit() + return healthy + + async def _send_to_relay( + self, + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + return await publish_frame_to_relay( + target, + event_id, + frame, + is_active=is_active, + timeout_seconds=self._timeout_seconds, + ) + + +async def publish_frame_to_relay( + target: RelayTarget, + event_id: str, + frame: bytes, + *, + is_active: Callable[[], Awaitable[bool]] | None = None, + timeout_seconds: float = 10.0, +) -> RelaySendResult: + """Send an exact persisted frame, then require OK and exact-id readback.""" + prior = prior_version_from_frame(frame) + if prior.event_id != event_id: + raise OutboxConflictError("Relay event ID does not match its signed frame") + frame_text = frame.decode("utf-8") + expected_event = json.loads(frame_text)[1] + subscription_id = uuid.uuid4().hex + try: + endpoint = await resolve_public_relay_endpoint(target.url) + connection = websockets.connect( + target.url, + open_timeout=timeout_seconds, + close_timeout=timeout_seconds, + host=endpoint.address, + port=endpoint.port, + server_hostname=endpoint.server_hostname, + ) + connection.MAX_REDIRECTS_ALLOWED = 1 + async with asyncio.timeout(timeout_seconds): + async with connection as websocket: + if is_active is not None and not await is_active(): + return RelaySendResult(False, False) + await websocket.send(frame_text) + accepted = False + while True: + parsed = parse_relay_ok(await websocket.recv(), event_id) + if parsed is None: + continue + if not parsed: + return RelaySendResult(False, False) + accepted = True + break + + request = json.dumps( + ["REQ", subscription_id, {"ids": [event_id]}], + separators=(",", ":"), + ) + await websocket.send(request) + read_back = False + while True: + message = _decode_relay_message(await websocket.recv()) + if ( + isinstance(message, list) + and len(message) == 3 + and message[0] == "EVENT" + and message[1] == subscription_id + and isinstance(message[2], dict) + and message[2] == expected_event + ): + read_back = True + break + elif ( + isinstance(message, list) + and len(message) >= 2 + and message[0] == "EOSE" + and message[1] == subscription_id + ): + break + await websocket.send( + json.dumps(["CLOSE", subscription_id], separators=(",", ":")) + ) + return RelaySendResult(accepted, read_back) + except asyncio.CancelledError: + raise + except Exception: + return RelaySendResult(False, False) + + +def parse_relay_ok(message: str | bytes, event_id: str) -> bool | None: + """Return the exact relay decision, or None for an unrelated frame.""" + try: + parsed = _decode_relay_message(message) + except (json.JSONDecodeError, UnicodeDecodeError): + return None + if ( + not isinstance(parsed, list) + or len(parsed) < 3 + or parsed[0] != "OK" + or parsed[1] != event_id + ): + return None + if parsed[2] is True: + return True + if parsed[2] is False: + return False + return None + + +async def resolve_public_relay_endpoint(relay_url: str) -> ResolvedRelayEndpoint: + """Resolve once, reject any private answer, and pin the socket destination.""" + normalized = _normalize_public_wss_url(relay_url) + parsed = urlsplit(normalized) + hostname = parsed.hostname + if hostname is None: + raise AnalyticsV2DeliveryError("Relay URL hostname is empty") + port = parsed.port or 443 + loop = asyncio.get_running_loop() + try: + answers = await loop.getaddrinfo( + hostname, + port, + family=socket.AF_UNSPEC, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + ) + except OSError as error: + raise AnalyticsV2DeliveryError("Relay hostname did not resolve") from error + addresses = {ipaddress.ip_address(answer[4][0]) for answer in answers if answer[4]} + if not addresses or any(not address.is_global for address in addresses): + raise AnalyticsV2DeliveryError("Relay hostname resolved to a non-public IP") + selected = sorted( + addresses, key=lambda address: (address.version, address.compressed) + )[0] + return ResolvedRelayEndpoint(selected.compressed, port, hostname) + + +async def run_analytics_v2_publisher( + producer: AnalyticsV2Producer, + delivery: AnalyticsV2Delivery, + *, + interval_seconds: float = 300.0, +) -> None: + """Run cancellation-safe production and background delivery.""" + if interval_seconds <= 0: + raise AnalyticsV2DeliveryError("Publisher interval must be positive") + try: + while True: + state = await get_analytics_v2_delivery_state(producer.session_factory) + if state.sharing_enabled: + try: + await producer.produce_once( + max_frame_bytes=await delivery.frame_limit() + ) + await delivery.deliver_pending_once() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("Analytics v2 publisher pass failed") + await asyncio.sleep(interval_seconds) + finally: + await delivery.stop() + + +async def _get_or_create_delivery_state( + session: AsyncSession, timestamp: int +) -> AnalyticsV2DeliveryState: + state = await session.get(AnalyticsV2DeliveryState, 1) + if state is None: + state = AnalyticsV2DeliveryState(id=1, updated_at_ms=timestamp) + session.add(state) + try: + await session.flush() + except IntegrityError: + await session.rollback() + state = await session.get(AnalyticsV2DeliveryState, 1) + if state is None: + raise + return state + + +async def _close_current_epoch( + session: AsyncSession, *, coverage_end_day: date +) -> None: + result = await session.exec( + select(TerminalOutcomeEpoch) + .where(col(TerminalOutcomeEpoch.current_slot) == 1) + .with_for_update() + ) + current = result.first() + if current is None: + return + if coverage_end_day < current.coverage_start_day: + outbox_result = await session.exec( + select(AnalyticsV2Outbox.event_id) + .where(col(AnalyticsV2Outbox.epoch) == current.epoch) + .limit(1) + ) + if outbox_result.first() is not None: + raise AnalyticsV2DeliveryError( + "Cannot discard an empty epoch with a durable outbox event" + ) + await session.delete(current) + return + current.coverage_end_day = coverage_end_day + current.current_slot = None + + +def _snapshot(state: AnalyticsV2DeliveryState) -> DeliveryStateSnapshot: + return DeliveryStateSnapshot( + sharing_enabled=state.sharing_enabled, + generation=state.generation, + active_epoch_floor=state.active_epoch_floor, + identity_pubkey=state.identity_pubkey, + provider_d=state.provider_d, + updated_at_ms=state.updated_at_ms, + ) + + +def _validate_encoded_metadata( + event: EncodedAnalyticsEvent, parsed: PriorVersion +) -> None: + expected = ( + event.event_id, + event.pubkey, + event.d_tag, + event.created_at, + event.week, + event.epoch, + event.coverage_start, + event.through, + event.complete, + event.corrected, + event.days, + event.daily_models, + ) + actual = ( + parsed.event_id, + parsed.pubkey, + parsed.d_tag, + parsed.created_at, + parsed.week, + parsed.epoch, + parsed.coverage_start, + parsed.through, + parsed.complete, + parsed.corrected, + parsed.days, + parsed.daily_models, + ) + if expected != actual: + raise OutboxConflictError("Encoded metadata does not match the signed frame") + + +def _require_exact_outbox( + row: AnalyticsV2Outbox, + event: EncodedAnalyticsEvent, + generation: int, + semantic_slot: str, +) -> None: + if row.semantic_slot not in { + semantic_slot, + f"superseded:{row.event_id}", + }: + raise OutboxConflictError("Event id has conflicting outbox data") + stored_prior = prior_version_from_frame(bytes(row.frame)) + if stored_prior.event_id != event.event_id: + raise OutboxConflictError("Event id has conflicting outbox data") + existing = ( + row.event_id, + row.pubkey, + row.d_tag, + row.kind, + row.week, + row.epoch, + row.through_day, + row.delivery_generation, + json.loads(bytes(row.frame))[1]["content"], + row.finalized, + row.corrected, + row.created_at, + ) + candidate = ( + event.event_id, + event.pubkey, + event.d_tag, + ANALYTICS_KIND, + event.week, + event.epoch, + event.through, + generation, + event.content.decode("utf-8"), + event.complete, + event.corrected, + event.created_at, + ) + if existing != candidate: + raise OutboxConflictError("Event id has conflicting outbox data") + + +def _require_semantic_winner( + row: AnalyticsV2Outbox, + event: EncodedAnalyticsEvent, + generation: int, + semantic_slot: str, +) -> None: + parsed = prior_version_from_frame(bytes(row.frame)) + frame = json.loads(bytes(row.frame)) + stored_content = frame[1]["content"].encode("utf-8") + existing = ( + row.pubkey, + row.d_tag, + row.week, + row.epoch, + row.through_day, + row.semantic_slot, + row.delivery_generation, + row.finalized, + row.corrected, + parsed.days, + stored_content, + ) + candidate = ( + event.pubkey, + event.d_tag, + event.week, + event.epoch, + event.through, + semantic_slot, + generation, + event.complete, + event.corrected, + event.days, + event.content, + ) + if existing != candidate: + raise OutboxConflictError("Semantic slot has conflicting signed content") + + +def _semantic_slot(event: EncodedAnalyticsEvent) -> str: + payload = json.loads(event.content) + corrects = payload.get("corrects") + if isinstance(corrects, str): + return f"correction:{corrects}" + return f"ordinary:{event.epoch}:{event.through.isoformat()}" + + +def _published_days_changed(current: WeeklyAggregate, prior: PriorVersion) -> bool: + current_days = {row.day: row.values for row in current.days} + return ( + current.coverage_start != prior.coverage_start + or current.epoch != prior.epoch + or any(current_days.get(row.day) != row.values for row in prior.days) + or daily_models_changed(current, prior) + ) + + +def _latest_attempted_predecessor( + row: AnalyticsV2Outbox, versions: Iterable[AnalyticsV2Outbox] +) -> AnalyticsV2Outbox | None: + candidates = [ + candidate + for candidate in versions + if candidate.epoch == row.epoch + and candidate.d_tag == row.d_tag + and candidate.delivery_generation == row.delivery_generation + and candidate.first_send_attempt_at_ms is not None + ] + if not candidates: + return None + return min( + candidates, key=lambda candidate: (-candidate.created_at, candidate.event_id) + ) + + +def _covered_weeks(coverage_start: date, coverage_end: date) -> tuple[date, ...]: + if coverage_start > coverage_end: + return () + first = coverage_start - timedelta(days=coverage_start.weekday()) + last = coverage_end - timedelta(days=coverage_end.weekday()) + return tuple( + first + timedelta(days=7 * offset) + for offset in range(((last - first).days // 7) + 1) + ) + + +def _validate_identity(pubkey: str, provider_d: str) -> None: + _require_lower_hex(pubkey, 64, "analytics pubkey") + if not isinstance(provider_d, str) or not 1 <= len(provider_d) <= 64: + raise IdentityMismatchError("provider_d must contain 1 to 64 characters") + if any(unicodedata.category(character) == "Cc" for character in provider_d): + raise IdentityMismatchError("provider_d contains control characters") + + +def _require_nonnegative_int(value: object, name: str) -> None: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise AnalyticsV2DeliveryError(f"{name} must be a non-negative integer") + + +def _decode_relay_message(message: str | bytes) -> Any: + if isinstance(message, bytes): + message = message.decode("utf-8") + return json.loads(message) + + +def _clock_ms() -> int: + return time.time_ns() // 1_000_000 + + +def _utc_day_from_ms(timestamp: int) -> date: + return datetime.fromtimestamp(timestamp / 1000, UTC).date() diff --git a/tests/unit/test_analytics_v2_delivery.py b/tests/unit/test_analytics_v2_delivery.py new file mode 100644 index 00000000..66d740d0 --- /dev/null +++ b/tests/unit/test_analytics_v2_delivery.py @@ -0,0 +1,1110 @@ +from __future__ import annotations + +import asyncio +import json +import socket +from collections.abc import AsyncIterator, Awaitable, Callable +from datetime import UTC, date, datetime, timedelta +from pathlib import Path +from typing import Any + +import pytest +import pytest_asyncio +from nostr_sdk import Keys +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlmodel import SQLModel, col, select +from sqlmodel.ext.asyncio.session import AsyncSession + +from routstr.core.db import ( + AnalyticsV2Outbox, + AnalyticsV2RelayReceipt, + TerminalOutcomeEpoch, +) +from routstr.nostr import analytics_v2_delivery as delivery_module +from routstr.nostr.analytics_v2 import ( + EncodedAnalyticsEvent, + LedgerOutcome, + aggregate_ledger_week, + encode_week_event, +) +from routstr.nostr.analytics_v2_delivery import ( + ActivationResult, + AnalyticsV2Delivery, + AnalyticsV2DeliveryError, + OutboxConflictError, + RelaySendResult, + RelayTarget, + ResolvedRelayEndpoint, + SharingDisabledError, + activate_analytics_v2_sharing, + claim_analytics_v2_identity, + enqueue_signed_event, + fetch_relay_max_message_length, + get_analytics_v2_delivery_state, + parse_relay_ok, + publish_frame_to_relay, + resolve_public_relay_endpoint, + rotate_analytics_v2_identity, + transition_analytics_v2_sharing, +) + +PRIVATE_KEY = "11" * 32 +OTHER_PRIVATE_KEY = "22" * 32 +PUBLIC_KEY = Keys.parse(PRIVATE_KEY).public_key().to_hex() +OTHER_PUBLIC_KEY = Keys.parse(OTHER_PRIVATE_KEY).public_key().to_hex() +WEEK = date(2026, 8, 31) +RELAYS = ( + "wss://relay-a.valid.net", + "wss://relay-b.valid.net", + "wss://relay-c.valid.net", +) + + +def _at_ms(day: date) -> int: + return int(datetime(day.year, day.month, day.day, tzinfo=UTC).timestamp() * 1000) + + +async def _no_advertised_limit(_target: RelayTarget, _timeout_seconds: float) -> None: + return None + + +@pytest_asyncio.fixture +async def session_factory( + tmp_path: Path, +) -> AsyncIterator[async_sessionmaker[AsyncSession]]: + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'delivery.db'}") + async with engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + yield factory + await engine.dispose() + + +async def _activate( + factory: async_sessionmaker[AsyncSession], +) -> ActivationResult: + assert ( + await claim_analytics_v2_identity( + factory, pubkey=PUBLIC_KEY, provider_d="provider", at_ms=1 + ) + == "initialized" + ) + return await activate_analytics_v2_sharing( + factory, coverage_day=WEEK - timedelta(days=1), at_ms=2 + ) + + +def _open_versions() -> tuple[EncodedAnalyticsEvent, EncodedAnalyticsEvent]: + first_aggregate = aggregate_ledger_week( + [], + epoch=0, + epoch_coverage_start=WEEK, + epoch_coverage_end=None, + week=WEEK, + today_utc=WEEK + timedelta(days=1), + ) + assert first_aggregate is not None + first = encode_week_event( + first_aggregate, + private_key_hex=PRIVATE_KEY, + provider_d="provider", + created_at=100, + ) + second_aggregate = aggregate_ledger_week( + [], + epoch=0, + epoch_coverage_start=WEEK, + epoch_coverage_end=None, + week=WEEK, + today_utc=WEEK + timedelta(days=2), + prior_version=first.as_prior_version(), + ) + assert second_aggregate is not None + second = encode_week_event( + second_aggregate, + private_key_hex=PRIVATE_KEY, + provider_d="provider", + created_at=101, + ) + return first, second + + +@pytest.mark.asyncio +async def test_activation_rotates_continuity_in_same_idempotent_transition( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + activation = await _activate(session_factory) + assert activation.transitioned is True + assert activation.state.sharing_enabled is True + assert activation.state.generation == 1 + + repeated = await activate_analytics_v2_sharing( + session_factory, coverage_day=WEEK, at_ms=3 + ) + assert repeated.transitioned is False + assert repeated.state.generation == 1 + async with session_factory() as session: + result = await session.exec( + select(TerminalOutcomeEpoch).order_by(col(TerminalOutcomeEpoch.epoch)) + ) + epochs = result.all() + assert [ + (row.epoch, row.coverage_start_day, row.coverage_end_day) for row in epochs + ] == [ + (0, WEEK, None), + ] + + with pytest.raises(AnalyticsV2DeliveryError, match="activate_analytics_v2"): + disabled = await transition_analytics_v2_sharing( + session_factory, + enabled=False, + at_ms=_at_ms(WEEK + timedelta(days=2)), + ) + assert disabled.sharing_enabled is False + await transition_analytics_v2_sharing(session_factory, enabled=True, at_ms=5) + reenabled = await activate_analytics_v2_sharing( + session_factory, coverage_day=WEEK + timedelta(days=4), at_ms=6 + ) + assert reenabled.transitioned is True + assert reenabled.state.generation == 3 + async with session_factory() as session: + result = await session.exec( + select(TerminalOutcomeEpoch).order_by(col(TerminalOutcomeEpoch.epoch)) + ) + epochs = result.all() + assert [ + (row.epoch, row.coverage_start_day, row.coverage_end_day) for row in epochs + ] == [ + # Opting out kept private coverage open until the next activation. + (0, WEEK, WEEK + timedelta(days=3)), + (1, WEEK + timedelta(days=5), None), + ] + + +@pytest.mark.asyncio +async def test_activation_on_flags_older_than_an_opt_out_is_refused( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + seen = (await _activate(session_factory)).state.generation + await transition_analytics_v2_sharing(session_factory, enabled=False, at_ms=3) + + with pytest.raises(SharingDisabledError): + await activate_analytics_v2_sharing( + session_factory, coverage_day=WEEK, at_ms=4, expected_generation=seen + ) + state = await get_analytics_v2_delivery_state(session_factory) + assert not state.sharing_enabled + + +@pytest.mark.asyncio +async def test_concurrent_activation_rotates_exactly_once( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await claim_analytics_v2_identity( + session_factory, + pubkey=PUBLIC_KEY, + provider_d="provider", + at_ms=1, + ) + results = await asyncio.gather( + activate_analytics_v2_sharing( + session_factory, coverage_day=WEEK - timedelta(days=1), at_ms=2 + ), + activate_analytics_v2_sharing( + session_factory, coverage_day=WEEK - timedelta(days=1), at_ms=2 + ), + ) + assert sorted(result.transitioned for result in results) == [False, True] + async with session_factory() as session: + epochs = (await session.exec(select(TerminalOutcomeEpoch))).all() + assert len(epochs) == 1 + assert sum(epoch.current_slot == 1 for epoch in epochs) == 1 + + +@pytest.mark.asyncio +async def test_first_activation_adopts_writer_created_epoch_zero( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + async with session_factory() as session: + session.add( + TerminalOutcomeEpoch( + epoch=0, + coverage_start_day=WEEK, + current_slot=1, + ) + ) + await session.commit() + await claim_analytics_v2_identity( + session_factory, + pubkey=PUBLIC_KEY, + provider_d="provider", + at_ms=1, + ) + + activation = await activate_analytics_v2_sharing( + session_factory, + coverage_day=WEEK - timedelta(days=1), + at_ms=2, + ) + + assert activation.transitioned is True + assert activation.state.active_epoch_floor == 0 + async with session_factory() as session: + epochs = (await session.exec(select(TerminalOutcomeEpoch))).all() + assert [ + ( + epoch.epoch, + epoch.coverage_start_day, + epoch.coverage_end_day, + epoch.current_slot, + ) + for epoch in epochs + ] == [(0, WEEK, None, 1)] + + +@pytest.mark.asyncio +async def test_concurrent_identity_claim_initializes_once( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + results = await asyncio.gather( + claim_analytics_v2_identity( + session_factory, + pubkey=PUBLIC_KEY, + provider_d="provider", + at_ms=1, + ), + claim_analytics_v2_identity( + session_factory, + pubkey=OTHER_PUBLIC_KEY, + provider_d="other", + at_ms=2, + ), + ) + assert sorted(results) == ["initialized", "mismatch"] + state = await get_analytics_v2_delivery_state(session_factory) + assert (state.identity_pubkey, state.provider_d) in { + (PUBLIC_KEY, "provider"), + (OTHER_PUBLIC_KEY, "other"), + } + + +@pytest.mark.asyncio +async def test_dns_resolution_rejects_private_answers_and_pins_public_ip( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def private_answer(*args: Any, **kwargs: Any) -> list[tuple[Any, ...]]: + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 443))] + + monkeypatch.setattr(delivery_module.socket, "getaddrinfo", private_answer) + with pytest.raises(AnalyticsV2DeliveryError, match="non-public"): + await resolve_public_relay_endpoint("wss://relay.valid.net") + + def public_answer(*args: Any, **kwargs: Any) -> list[tuple[Any, ...]]: + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 443))] + + monkeypatch.setattr(delivery_module.socket, "getaddrinfo", public_answer) + assert await resolve_public_relay_endpoint("wss://relay.valid.net") == ( + ResolvedRelayEndpoint("8.8.8.8", 443, "relay.valid.net") + ) + + +@pytest.mark.parametrize( + ("payload", "expected"), + [ + ('["OK","event",true,""]', True), + ('["OK","event",false,"blocked"]', False), + ('["OK","other",true,""]', None), + ('["OK","event",1,""]', None), + ('["NOTICE","event",true]', None), + ], +) +def test_parse_relay_ok_requires_exact_id_and_boolean( + payload: str, expected: bool | None +) -> None: + assert parse_relay_ok(payload, "event") is expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize("readback_kind", ["exact", "wrong_id", "forged_content"]) +async def test_websocket_send_preserves_frame_and_requires_exact_readback( + monkeypatch: pytest.MonkeyPatch, + readback_kind: str, +) -> None: + sent: list[str] = [] + encoded, _ = _open_versions() + event_id = encoded.event_id + returned_event = dict(encoded.event) + if readback_kind == "wrong_id": + returned_event["id"] = "00" * 32 + elif readback_kind == "forged_content": + returned_event["content"] = "{}" + + class FakeWebSocket: + receive_count = 0 + + async def __aenter__(self) -> FakeWebSocket: + return self + + async def __aexit__(self, *args: object) -> None: + return None + + async def send(self, message: str) -> None: + sent.append(message) + + async def recv(self) -> str: + self.receive_count += 1 + if self.receive_count == 1: + return json.dumps(["OK", event_id, True, ""]) + subscription_id = json.loads(sent[1])[1] + if self.receive_count == 2: + return json.dumps( + ["EVENT", subscription_id, returned_event], + separators=(",", ":"), + ) + return json.dumps(["EOSE", subscription_id], separators=(",", ":")) + + async def resolved(url: str) -> ResolvedRelayEndpoint: + return ResolvedRelayEndpoint("8.8.8.8", 443, "relay.valid.net") + + monkeypatch.setattr(delivery_module, "resolve_public_relay_endpoint", resolved) + monkeypatch.setattr( + delivery_module.websockets, + "connect", + lambda *args, **kwargs: FakeWebSocket(), + ) + frame = encoded.frame + result = await publish_frame_to_relay( + RelayTarget("wss://relay.valid.net"), + event_id, + frame, + ) + assert sent[0].encode() == frame + assert result == RelaySendResult(True, readback_kind == "exact") + + +@pytest.mark.asyncio +async def test_nip11_lookup_is_pinned_bounded_and_disables_redirects( + monkeypatch: pytest.MonkeyPatch, +) -> None: + maximum = delivery_module.NIP11_MAX_DOCUMENT_BYTES + valid_body = json.dumps({"limitation": {"max_message_length": 123}}).encode() + responses = [ + (200, valid_body, len(valid_body)), + (302, valid_body, len(valid_body)), + (200, valid_body, maximum + 1), + (200, b"x" * (maximum + 1), None), + ] + requests: list[tuple[str, dict[str, Any]]] = [] + connector_arguments: list[dict[str, Any]] = [] + read_sizes: list[int] = [] + + class FakeContent: + def __init__(self, body: bytes) -> None: + self._body = body + + async def read(self, size: int) -> bytes: + read_sizes.append(size) + return self._body + + class FakeResponse: + def __init__( + self, status: int, body: bytes, content_length: int | None + ) -> None: + self.status = status + self.content_length = content_length + self.content = FakeContent(body) + + async def __aenter__(self) -> FakeResponse: + return self + + async def __aexit__(self, *args: object) -> None: + return None + + class FakeSession: + def __init__(self, **kwargs: Any) -> None: + self._arguments = kwargs + + async def __aenter__(self) -> FakeSession: + return self + + async def __aexit__(self, *args: object) -> None: + return None + + def get(self, url: str, **kwargs: Any) -> FakeResponse: + requests.append((url, kwargs)) + return FakeResponse(*responses.pop(0)) + + async def resolved(_url: str) -> ResolvedRelayEndpoint: + return ResolvedRelayEndpoint("8.8.8.8", 443, "relay.valid.net") + + def connector(**kwargs: Any) -> object: + connector_arguments.append(kwargs) + return object() + + monkeypatch.setattr(delivery_module, "resolve_public_relay_endpoint", resolved) + monkeypatch.setattr(delivery_module.aiohttp, "TCPConnector", connector) + monkeypatch.setattr(delivery_module.aiohttp, "ClientSession", FakeSession) + target = RelayTarget("wss://relay.valid.net/path") + + assert await fetch_relay_max_message_length(target, 1) == 123 + assert await fetch_relay_max_message_length(target, 1) is None + assert await fetch_relay_max_message_length(target, 1) is None + assert await fetch_relay_max_message_length(target, 1) is None + assert all(url == "https://relay.valid.net/path" for url, _ in requests) + assert all( + request["allow_redirects"] is False + and request["headers"] == {"Accept": "application/nostr+json"} + for _, request in requests + ) + assert read_sizes == [maximum + 1, maximum + 1] + + resolver = connector_arguments[0]["resolver"] + resolved_addresses = await resolver.resolve("relay.valid.net", 443) + assert resolved_addresses[0]["host"] == "8.8.8.8" + with pytest.raises(OSError, match="changed hostname"): + await resolver.resolve("127.0.0.1", 443) + + +@pytest.mark.asyncio +async def test_delivery_applies_nip11_limit_to_exact_persisted_frame_bytes( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + limits = { + "wss://relay-a.valid.net": len(first.frame) - 1, + "wss://relay-b.valid.net": len(first.frame), + "wss://relay-c.valid.net": None, + } + sends: list[str] = [] + + async def relay_limit_reader( + target: RelayTarget, _timeout_seconds: float + ) -> int | None: + return limits[target.url] + + async def sender( + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + assert event_id == first.event_id + assert frame == first.frame + sends.append(target.url) + return RelaySendResult(True, True) + + delivery = AnalyticsV2Delivery( + session_factory, + operator_relays=RELAYS, + sender=sender, + relay_limit_reader=relay_limit_reader, + ) + assert await delivery.deliver_pending_once(at_ms=4) == ( + delivery_module.DeliveryPassResult(1, 1) + ) + assert set(sends) == { + "wss://relay-b.valid.net", + "wss://relay-c.valid.net", + } + async with session_factory() as session: + receipts = ( + await session.exec( + select(AnalyticsV2RelayReceipt).where( + col(AnalyticsV2RelayReceipt.event_id) == first.event_id + ) + ) + ).all() + assert {receipt.relay_url for receipt in receipts} == set(sends) + + +@pytest.mark.asyncio +async def test_all_nip11_ineligible_relays_do_not_record_a_send_attempt( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + sends = 0 + + async def relay_limit_reader(_target: RelayTarget, _timeout_seconds: float) -> int: + return len(first.frame) - 1 + + async def sender( + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + nonlocal sends + sends += 1 + return RelaySendResult(True, True) + + delivery = AnalyticsV2Delivery( + session_factory, + operator_relays=RELAYS, + sender=sender, + relay_limit_reader=relay_limit_reader, + ) + assert await delivery.deliver_pending_once(at_ms=4) == ( + delivery_module.DeliveryPassResult(0, 0) + ) + async with session_factory() as session: + row = await session.get(AnalyticsV2Outbox, first.event_id) + receipts = ( + await session.exec( + select(AnalyticsV2RelayReceipt).where( + col(AnalyticsV2RelayReceipt.event_id) == first.event_id + ) + ) + ).all() + assert sends == 0 + assert row is not None and row.attempt_count == 0 + assert row.first_send_attempt_at_ms is None + assert receipts == [] + + +@pytest.mark.asyncio +async def test_nip11_ineligible_remainder_reconciles_durable_quorum( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + async with session_factory() as session: + row = await session.get(AnalyticsV2Outbox, first.event_id) + assert row is not None + row.attempt_count = 1 + row.first_send_attempt_at_ms = 3 + session.add_all( + [ + AnalyticsV2RelayReceipt( + event_id=first.event_id, + relay_url="wss://relay-a.valid.net", + accepted_at_ms=3, + read_back_at_ms=3, + ), + AnalyticsV2RelayReceipt( + event_id=first.event_id, + relay_url="wss://relay-c.valid.net", + accepted_at_ms=3, + read_back_at_ms=3, + ), + ] + ) + await session.commit() + sends = 0 + + async def relay_limit_reader(target: RelayTarget, _timeout_seconds: float) -> int: + assert target.url == "wss://relay-b.valid.net" + return len(first.frame) - 1 + + async def sender( + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + nonlocal sends + sends += 1 + return RelaySendResult(True, True) + + delivery = AnalyticsV2Delivery( + session_factory, + operator_relays=RELAYS, + sender=sender, + relay_limit_reader=relay_limit_reader, + ) + assert await delivery.deliver_pending_once(at_ms=4) == ( + delivery_module.DeliveryPassResult(1, 1) + ) + async with session_factory() as session: + row = await session.get(AnalyticsV2Outbox, first.event_id) + assert sends == 0 + assert row is not None and row.status == "delivered" + assert row.attempt_count == 1 + + +@pytest.mark.asyncio +async def test_delivery_commits_before_network_and_retries_exact_bytes_after_restart( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + seen_frames: dict[str, list[bytes]] = {} + group_b_attempts = 0 + + async def sender( + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + nonlocal group_b_attempts + async with session_factory() as session: + row = await session.get(AnalyticsV2Outbox, event_id) + assert row is not None and bytes(row.frame) == frame + assert row.first_send_attempt_at_ms == 4 + assert row.attempt_count >= 1 + seen_frames.setdefault(target.url, []).append(frame) + if target.url == RELAYS[1]: + return RelaySendResult(False, False) + if target.url == RELAYS[2]: + group_b_attempts += 1 + if group_b_attempts == 1: + return RelaySendResult(False, False) + return RelaySendResult(True, True) + + delivery = AnalyticsV2Delivery( + session_factory, + operator_relays=RELAYS, + retry_seconds=0, + sender=sender, + relay_limit_reader=_no_advertised_limit, + ) + first_pass = await delivery.deliver_pending_once(at_ms=4) + assert first_pass == delivery_module.DeliveryPassResult(1, 0) + delivery = AnalyticsV2Delivery( + session_factory, + operator_relays=RELAYS, + retry_seconds=0, + sender=sender, + relay_limit_reader=_no_advertised_limit, + ) + second_pass = await delivery.deliver_pending_once(at_ms=5) + assert second_pass == delivery_module.DeliveryPassResult(1, 1) + + assert seen_frames["wss://relay-c.valid.net"] == [first.frame, first.frame] + async with session_factory() as session: + row = await session.get(AnalyticsV2Outbox, first.event_id) + receipts_result = await session.exec( + select(AnalyticsV2RelayReceipt).where( + col(AnalyticsV2RelayReceipt.event_id) == first.event_id + ) + ) + assert row is not None and row.status == "delivered" + assert {receipt.relay_url for receipt in receipts_result.all()} == { + RELAYS[0], + RELAYS[2], + } + + +@pytest.mark.asyncio +async def test_delivery_retains_and_attempts_same_week_epochs_in_order( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + closed_aggregate = aggregate_ledger_week( + [], + epoch=0, + epoch_coverage_start=WEEK, + epoch_coverage_end=WEEK + timedelta(days=2), + week=WEEK, + today_utc=WEEK + timedelta(days=4), + ) + current_aggregate = aggregate_ledger_week( + [], + epoch=1, + epoch_coverage_start=WEEK + timedelta(days=4), + epoch_coverage_end=None, + week=WEEK, + today_utc=WEEK + timedelta(days=6), + ) + assert closed_aggregate is not None and current_aggregate is not None + closed = encode_week_event( + closed_aggregate, + private_key_hex=PRIVATE_KEY, + provider_d="provider", + created_at=100, + ) + current = encode_week_event( + current_aggregate, + private_key_hex=PRIVATE_KEY, + provider_d="provider", + created_at=101, + ) + await enqueue_signed_event(session_factory, closed, stored_at_ms=3) + await enqueue_signed_event(session_factory, current, stored_at_ms=4) + attempts: list[str] = [] + + async def sender( + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + attempts.append(event_id) + return RelaySendResult(True, True) + + delivery = AnalyticsV2Delivery( + session_factory, + operator_relays=RELAYS, + sender=sender, + relay_limit_reader=_no_advertised_limit, + ) + assert await delivery.deliver_pending_once(at_ms=5) == ( + delivery_module.DeliveryPassResult(2, 2) + ) + assert attempts == [closed.event_id] * 3 + [current.event_id] * 3 + async with session_factory() as session: + rows = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + ).all() + assert [(row.epoch, row.status) for row in rows] == [ + (0, "delivered"), + (1, "delivered"), + ] + + +@pytest.mark.asyncio +async def test_open_versions_coalesce_but_finalized_versions_are_retained( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, second = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + await enqueue_signed_event(session_factory, second, stored_at_ms=4) + + final_aggregate = aggregate_ledger_week( + [], + epoch=0, + epoch_coverage_start=WEEK, + epoch_coverage_end=None, + week=WEEK, + today_utc=WEEK + timedelta(days=7), + prior_version=second.as_prior_version(), + ) + assert final_aggregate is not None + final = encode_week_event( + final_aggregate, + private_key_hex=PRIVATE_KEY, + provider_d="provider", + created_at=102, + ) + await enqueue_signed_event(session_factory, final, stored_at_ms=5) + async with session_factory() as session: + stored_final = await session.get(AnalyticsV2Outbox, final.event_id) + assert stored_final is not None + assert stored_final.status == "pending" + stored_final.first_send_attempt_at_ms = 5 + await session.commit() + + late = LedgerOutcome( + terminal_day=WEEK, + model_identifier="model/a", + input_source="reported", + output_source="reported", + cache_read_source="missing", + cache_creation_source="missing", + input_tokens=1, + output_tokens=1, + cache_read_input_tokens=0, + cache_creation_input_tokens=0, + revenue_msats=1, + ) + correction_aggregate = aggregate_ledger_week( + [late], + epoch=0, + epoch_coverage_start=WEEK, + epoch_coverage_end=None, + week=WEEK, + today_utc=WEEK + timedelta(days=7), + prior_version=final.as_prior_version(), + correction=True, + ) + assert correction_aggregate is not None + correction = encode_week_event( + correction_aggregate, + private_key_hex=PRIVATE_KEY, + provider_d="provider", + created_at=103, + ) + inserted = await enqueue_signed_event(session_factory, correction, stored_at_ms=6) + repeated = await enqueue_signed_event(session_factory, correction, stored_at_ms=7) + assert inserted.inserted is True + assert repeated.inserted is False + + async with session_factory() as session: + result = await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + rows = result.all() + # Only its own correction retires a finalized report that may be on relays. + assert [(row.status, row.finalized) for row in rows] == [ + ("superseded", False), + ("superseded", False), + ("superseded", True), + ("pending", True), + ] + assert bytes(rows[2].frame) == final.frame + + +@pytest.mark.asyncio +async def test_duplicate_event_id_requires_exact_outbox_readback( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + async with session_factory() as session: + stored = await session.get(AnalyticsV2Outbox, first.event_id) + assert stored is not None + stored.d_tag = "conflicting-coordinate" + await session.commit() + + with pytest.raises(OutboxConflictError, match="conflicting outbox"): + await enqueue_signed_event(session_factory, first, stored_at_ms=4) + + +@pytest.mark.asyncio +async def test_disable_cancels_inflight_and_old_generation_never_retries( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + started = asyncio.Event() + cancelled = asyncio.Event() + + async def blocking_sender( + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + started.set() + try: + await asyncio.Future() + raise AssertionError("blocking sender unexpectedly resumed") + except asyncio.CancelledError: + cancelled.set() + raise + + delivery = AnalyticsV2Delivery( + session_factory, + operator_relays=RELAYS, + sender=blocking_sender, + relay_limit_reader=_no_advertised_limit, + ) + delivery_pass = asyncio.create_task(delivery.deliver_pending_once(at_ms=4)) + await started.wait() + disabled = await delivery.disable(at_ms=_at_ms(WEEK + timedelta(days=2))) + await delivery_pass + + assert cancelled.is_set() + assert disabled.sharing_enabled is False + assert disabled.generation == 2 + assert await delivery.resume() is False + assert await delivery.deliver_pending_once(at_ms=6) == ( + delivery_module.DeliveryPassResult(0, 0) + ) + async with session_factory() as session: + row = await session.get(AnalyticsV2Outbox, first.event_id) + assert row is not None and row.status == "cancelled" + + +@pytest.mark.asyncio +async def test_disable_during_receipt_lookup_prevents_sender_start( + session_factory: async_sessionmaker[AsyncSession], + monkeypatch: pytest.MonkeyPatch, +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + lookup_started = asyncio.Event() + resume_lookup = asyncio.Event() + sends = 0 + original_exec = AsyncSession.exec + + async def paused_exec( + session: AsyncSession, statement: Any, *args: Any, **kwargs: Any + ) -> Any: + if "analytics_v2_relay_receipts" in str(statement): + lookup_started.set() + await resume_lookup.wait() + return await original_exec(session, statement, *args, **kwargs) + + async def sender( + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + nonlocal sends + sends += 1 + return RelaySendResult(True, True) + + monkeypatch.setattr(AsyncSession, "exec", paused_exec) + delivery = AnalyticsV2Delivery( + session_factory, + operator_relays=RELAYS, + sender=sender, + relay_limit_reader=_no_advertised_limit, + ) + delivery_pass = asyncio.create_task(delivery.deliver_pending_once(at_ms=4)) + await lookup_started.wait() + await delivery.disable(at_ms=_at_ms(WEEK + timedelta(days=2))) + resume_lookup.set() + + assert await delivery_pass == delivery_module.DeliveryPassResult(0, 0) + assert sends == 0 + + +@pytest.mark.asyncio +async def test_cross_instance_disable_fences_event_write_after_handshake( + session_factory: async_sessionmaker[AsyncSession], + monkeypatch: pytest.MonkeyPatch, +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + handshake_started = asyncio.Event() + resume_handshake = asyncio.Event() + event_writes = 0 + + class PausedWebSocket: + MAX_REDIRECTS_ALLOWED = 10 + + async def __aenter__(self) -> PausedWebSocket: + handshake_started.set() + await resume_handshake.wait() + return self + + async def __aexit__(self, *args: object) -> None: + return None + + async def send(self, message: str) -> None: + nonlocal event_writes + if json.loads(message)[0] == "EVENT": + event_writes += 1 + + async def resolved(url: str) -> ResolvedRelayEndpoint: + return ResolvedRelayEndpoint("8.8.8.8", 443, "relay.valid.net") + + monkeypatch.setattr(delivery_module, "resolve_public_relay_endpoint", resolved) + monkeypatch.setattr( + delivery_module.websockets, + "connect", + lambda *args, **kwargs: PausedWebSocket(), + ) + disabling_instance = AnalyticsV2Delivery(session_factory, operator_relays=RELAYS) + sending_instance = AnalyticsV2Delivery( + session_factory, + operator_relays=RELAYS, + relay_limit_reader=_no_advertised_limit, + ) + delivery_pass = asyncio.create_task(sending_instance.deliver_pending_once(at_ms=4)) + await handshake_started.wait() + disabled = await disabling_instance.disable(at_ms=_at_ms(WEEK + timedelta(days=2))) + resume_handshake.set() + + assert await delivery_pass == delivery_module.DeliveryPassResult(0, 0) + assert disabled.sharing_enabled is False + assert event_writes == 0 + + +@pytest.mark.asyncio +async def test_identity_rotation_is_atomic_and_cancels_pending_reports( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + rotated = await rotate_analytics_v2_identity( + session_factory, + pubkey=OTHER_PUBLIC_KEY, + provider_d="other-provider", + at_ms=4, + ) + assert rotated.sharing_enabled is False + assert rotated.generation == 2 + assert rotated.identity_pubkey == OTHER_PUBLIC_KEY + assert rotated.provider_d == "other-provider" + async with session_factory() as session: + row = await session.get(AnalyticsV2Outbox, first.event_id) + assert row is not None and row.status == "cancelled" + assert await get_analytics_v2_delivery_state(session_factory) == rotated + + +@pytest.mark.asyncio +async def test_sdk_signature_randomness_reuses_the_first_durable_event( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + second, _ = _open_versions() + assert first.event_id == second.event_id + initial = await enqueue_signed_event(session_factory, first, stored_at_ms=3) + duplicate = await enqueue_signed_event(session_factory, second, stored_at_ms=4) + assert initial.inserted is True + assert duplicate.inserted is False + async with session_factory() as session: + row = await session.get(AnalyticsV2Outbox, first.event_id) + assert row is not None + assert bytes(row.frame) == first.frame + + +@pytest.mark.asyncio +async def test_single_configured_relay_can_deliver_without_a_manifest( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + first, _ = _open_versions() + await enqueue_signed_event(session_factory, first, stored_at_ms=3) + + async def sender( + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + assert await is_active() + return RelaySendResult(True, True) + + delivery = AnalyticsV2Delivery( + session_factory, + operator_relays=(RELAYS[0], RELAYS[0]), + sender=sender, + relay_limit_reader=_no_advertised_limit, + ) + assert await delivery.deliver_pending_once( + at_ms=4 + ) == delivery_module.DeliveryPassResult(1, 1) + + +@pytest.mark.asyncio +async def test_frame_limit_uses_the_required_relay_quorum_and_caches_each_pass( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + reads: list[str] = [] + advertised = {RELAYS[0]: 12_000, RELAYS[1]: 8_000, RELAYS[2]: 500} + + async def reader(target: RelayTarget, timeout: float) -> int: + reads.append(target.url) + return advertised[target.url] + + delivery = AnalyticsV2Delivery( + session_factory, operator_relays=RELAYS, relay_limit_reader=reader + ) + assert await delivery.frame_limit() == 8_000 + assert set( + await delivery._eligible_targets(delivery._targets, b"x" * 8_000) + ) == set(delivery._targets[:2]) + assert sorted(reads) == sorted(RELAYS) + + +@pytest.mark.asyncio +async def test_unavailable_relay_information_keeps_a_bounded_default_frame_limit( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + async def reader(target: RelayTarget, timeout: float) -> int: + raise OSError("information endpoint unavailable") + + delivery = AnalyticsV2Delivery( + session_factory, operator_relays=RELAYS, relay_limit_reader=reader + ) + assert await delivery.frame_limit() == 96 * 1024 diff --git a/tests/unit/test_analytics_v2_outbox_migration.py b/tests/unit/test_analytics_v2_outbox_migration.py new file mode 100644 index 00000000..460113cd --- /dev/null +++ b/tests/unit/test_analytics_v2_outbox_migration.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import os +import sqlite3 +import subprocess +import sys +from pathlib import Path + +import pytest + +REVISION = "d9a6e2f4c7b1" +PREVIOUS_REVISION = "c8e4a1f2b3d5" + + +def _run_alembic(root: Path, database_url: str, command: str, revision: str) -> None: + env = os.environ.copy() + env["DATABASE_URL"] = database_url + subprocess.run( + [sys.executable, "-m", "alembic", command, revision], + cwd=root, + env=env, + check=True, + capture_output=True, + text=True, + ) + + +def _tables(connection: sqlite3.Connection) -> set[str]: + return { + row[0] + for row in connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ) + } + + +def _columns(connection: sqlite3.Connection, table: str) -> dict[str, tuple[str, int]]: + return { + row[1]: (row[2], row[3]) + for row in connection.execute(f"PRAGMA table_info({table})") + } + + +def test_analytics_v2_outbox_migration_round_trips(tmp_path: Path) -> None: + root = Path(__file__).resolve().parents[2] + database_path = tmp_path / "analytics-v2-outbox.db" + database_url = f"sqlite+aiosqlite:///{database_path}" + tables = { + "analytics_v2_delivery_state", + "analytics_v2_outbox", + "analytics_v2_relay_receipts", + } + + _run_alembic(root, database_url, "upgrade", PREVIOUS_REVISION) + with sqlite3.connect(database_path) as connection: + assert not tables & _tables(connection) + + _run_alembic(root, database_url, "upgrade", REVISION) + with sqlite3.connect(database_path) as connection: + assert tables <= _tables(connection) + assert all( + connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone() == (0,) + for table in tables + ) + + outbox_columns = _columns(connection, "analytics_v2_outbox") + assert outbox_columns["frame"] == ("BLOB", 1) + for column in ( + "epoch", + "delivery_generation", + "created_at", + "stored_at_ms", + "next_attempt_at_ms", + "attempt_count", + ): + assert outbox_columns[column] == ("BIGINT", 1) + assert outbox_columns["through_day"] == ("DATE", 1) + assert outbox_columns["semantic_slot"] == ("VARCHAR", 1) + assert outbox_columns["first_send_attempt_at_ms"] == ("BIGINT", 0) + assert "acknowledged_at_ms" not in outbox_columns + assert outbox_columns["delivered_at_ms"] == ("BIGINT", 0) + state_columns = _columns(connection, "analytics_v2_delivery_state") + assert state_columns["active_epoch_floor"] == ("BIGINT", 0) + assert not {"ever_send_attempted", "ever_acknowledged"} & state_columns.keys() + + pending_index = connection.execute( + "PRAGMA index_info(ix_analytics_v2_outbox_pending)" + ).fetchall() + assert [row[2] for row in pending_index] == [ + "status", + "delivery_generation", + "next_attempt_at_ms", + ] + coordinate_index = connection.execute( + "PRAGMA index_info(ix_analytics_v2_outbox_coordinate)" + ).fetchall() + assert [row[2] for row in coordinate_index] == [ + "pubkey", + "d_tag", + "epoch", + "created_at", + ] + + connection.execute( + "INSERT INTO analytics_v2_delivery_state VALUES " + "(1, 1, 1, 0, ?, 'provider', 1)", + ("11" * 32,), + ) + connection.execute( + "INSERT INTO analytics_v2_outbox VALUES " + "(?, ?, 'routstr.analytics.v2:test:week:2026-08-31', 38422, " + "'2026-08-31', 1, '2026-08-31', " + "'ordinary:1:2026-08-31', 1, X'5B5D', 0, 0, 'pending', " + "1, 1, 1, 0, NULL, NULL)", + ("33" * 32, "11" * 32), + ) + connection.execute( + "INSERT INTO analytics_v2_relay_receipts VALUES " + "(?, 'wss://relay.example.net', 1, 1)", + ("33" * 32,), + ) + with pytest.raises(sqlite3.IntegrityError): + connection.execute( + "INSERT INTO analytics_v2_delivery_state VALUES " + "(2, 0, 0, NULL, NULL, NULL, 1)" + ) + with pytest.raises(sqlite3.IntegrityError): + connection.execute( + "UPDATE analytics_v2_delivery_state " + "SET active_epoch_floor = NULL WHERE id = 1" + ) + with pytest.raises(sqlite3.IntegrityError): + connection.execute( + "INSERT INTO analytics_v2_outbox VALUES " + "(?, ?, 'invalid', 1, '2026-08-31', 0, '2026-08-31', " + "'ordinary:0:2026-08-31', 0, X'5B5D', " + "0, 0, 'pending', 1, 1, 1, 0, NULL, NULL)", + ("44" * 32, "11" * 32), + ) + with pytest.raises(sqlite3.IntegrityError): + connection.execute( + "INSERT INTO analytics_v2_outbox VALUES " + "(?, ?, 'routstr.analytics.v2:test:week:2026-08-31', 38422, " + "'2026-08-31', 1, '2026-08-31', " + "'ordinary:1:2026-08-31', 1, X'5B5D', " + "0, 0, 'pending', 2, 2, 2, 0, NULL, NULL)", + ("55" * 32, "11" * 32), + ) + with pytest.raises(sqlite3.IntegrityError): + connection.execute( + "INSERT INTO analytics_v2_relay_receipts VALUES " + "(?, 'wss://other.example.net', -1, 1)", + ("33" * 32,), + ) + connection.commit() + + _run_alembic(root, database_url, "downgrade", PREVIOUS_REVISION) + with sqlite3.connect(database_path) as connection: + assert not tables & _tables(connection) + assert "terminal_outcomes" in _tables(connection) + + _run_alembic(root, database_url, "upgrade", REVISION) + with sqlite3.connect(database_path) as connection: + assert tables <= _tables(connection) + assert all( + connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone() == (0,) + for table in tables + ) diff --git a/tests/unit/test_analytics_v2_producer.py b/tests/unit/test_analytics_v2_producer.py new file mode 100644 index 00000000..1ca42bdd --- /dev/null +++ b/tests/unit/test_analytics_v2_producer.py @@ -0,0 +1,947 @@ +from __future__ import annotations + +import asyncio +import json +from collections.abc import AsyncIterator, Awaitable, Callable +from datetime import UTC, date, datetime, timedelta +from pathlib import Path + +import pytest +import pytest_asyncio +from nostr_sdk import Keys +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlmodel import SQLModel, col, select +from sqlmodel.ext.asyncio.session import AsyncSession + +from routstr.core.db import ( + AnalyticsV2Outbox, + TerminalOutcome, + TerminalOutcomeEpoch, +) +from routstr.nostr.analytics_v2 import aggregate_ledger_week, encode_week_event +from routstr.nostr.analytics_v2_delivery import ( + AnalyticsV2Delivery, + AnalyticsV2Producer, + RelaySendResult, + RelayTarget, + activate_analytics_v2_sharing, + claim_analytics_v2_identity, + enqueue_signed_event, + run_analytics_v2_publisher, + transition_analytics_v2_sharing, +) + +PRIVATE_KEY = "11" * 32 +PUBLIC_KEY = Keys.parse(PRIVATE_KEY).public_key().to_hex() +WEEK = date(2026, 8, 31) + + +def _at_ms(day: date) -> int: + return int(datetime(day.year, day.month, day.day, tzinfo=UTC).timestamp() * 1000) + + +@pytest_asyncio.fixture +async def session_factory( + tmp_path: Path, +) -> AsyncIterator[async_sessionmaker[AsyncSession]]: + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'producer.db'}") + async with engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + yield factory + await engine.dispose() + + +def _outcome( + outcome_id: str, terminal_day: date, *, tokens: int = 1 +) -> TerminalOutcome: + return TerminalOutcome( + outcome_id=outcome_id, + terminal_at_ms=1, + terminal_day=terminal_day, + model_identifier="model/served", + input_tokens=tokens, + output_tokens=tokens, + cache_read_input_tokens=0, + cache_creation_input_tokens=0, + revenue_msats=tokens, + input_source="reported", + output_source="reported", + cache_read_source="missing", + cache_creation_source="missing", + ) + + +async def _activate( + factory: async_sessionmaker[AsyncSession], +) -> None: + await claim_analytics_v2_identity( + factory, + pubkey=PUBLIC_KEY, + provider_d="provider", + at_ms=1, + ) + await activate_analytics_v2_sharing( + factory, + coverage_day=WEEK - timedelta(days=1), + at_ms=2, + ) + + +@pytest.mark.asyncio +async def test_producer_recovers_each_current_epoch_week_once_and_corrects_sent_rows( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + session.add(_outcome("old-epoch", WEEK - timedelta(days=2))) + session.add(_outcome("week-one", WEEK)) + session.add(_outcome("week-two", WEEK + timedelta(days=8), tokens=2)) + await session.commit() + + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + recovery_time = datetime(2026, 9, 14, 0, 1, tzinfo=UTC) + assert await producer.produce_once(now=recovery_time) == 2 + assert await producer.produce_once(now=recovery_time) == 0 + + async with session_factory() as session: + result = await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.week)) + ) + initial = result.all() + assert [(row.week, row.epoch, row.finalized) for row in initial] == [ + (WEEK, 0, True), + (WEEK + timedelta(days=7), 0, True), + ] + first_payload = json.loads(json.loads(initial[0].frame)[1]["content"]) + second_payload = json.loads(json.loads(initial[1].frame)[1]["content"]) + assert first_payload["days"][WEEK.isoformat()][0] == 1 + assert second_payload["days"][(WEEK + timedelta(days=8)).isoformat()][0] == 1 + + async with session_factory() as session: + first = await session.get(AnalyticsV2Outbox, initial[0].event_id) + assert first is not None + first.first_send_attempt_at_ms = 3 + session.add(_outcome("late-week-one", WEEK + timedelta(days=1), tokens=3)) + await session.commit() + + assert await producer.produce_once(now=recovery_time) == 1 + async with session_factory() as session: + result = await session.exec( + select(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.week) == WEEK) + .order_by(col(AnalyticsV2Outbox.created_at)) + ) + versions = result.all() + assert len(versions) == 2 + correction_payload = json.loads(json.loads(versions[1].frame)[1]["content"]) + assert correction_payload["corrects"] == versions[0].event_id + assert correction_payload["corrected"] is True + assert correction_payload["days"][(WEEK + timedelta(days=1)).isoformat()][0] == 1 + + async with session_factory() as session: + session.add(_outcome("later-week-one", WEEK + timedelta(days=1), tokens=4)) + await session.commit() + assert await producer.produce_once(now=recovery_time) == 1 + async with session_factory() as session: + result = await session.exec( + select(AnalyticsV2Outbox) + .where(col(AnalyticsV2Outbox.week) == WEEK) + .order_by(col(AnalyticsV2Outbox.created_at)) + ) + versions = result.all() + assert [row.status for row in versions] == ["superseded", "superseded", "pending"] + replacement_payload = json.loads(json.loads(versions[2].frame)[1]["content"]) + assert replacement_payload["corrects"] == versions[0].event_id + assert replacement_payload["days"][(WEEK + timedelta(days=1)).isoformat()][0] == 2 + + +@pytest.mark.asyncio +async def test_unsent_changed_version_is_coalesced_without_public_correction( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + session.add(_outcome("first", WEEK)) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + recovery_time = datetime(2026, 9, 7, 0, 1, tzinfo=UTC) + assert await producer.produce_once(now=recovery_time) == 1 + + async with session_factory() as session: + session.add(_outcome("late", WEEK + timedelta(days=1))) + await session.commit() + assert await producer.produce_once(now=recovery_time) == 1 + + async with session_factory() as session: + result = await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + rows = result.all() + assert [row.status for row in rows] == ["superseded", "pending"] + latest_payload = json.loads(json.loads(rows[1].frame)[1]["content"]) + assert "corrects" not in latest_payload + assert "corrected" not in latest_payload + + +@pytest.mark.asyncio +async def test_loss_rotation_finalizes_closed_epoch_before_new_epoch_same_week( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + session.add(_outcome("initial", WEEK)) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + assert await producer.produce_once(now=datetime(2026, 9, 2, 0, 1, tzinfo=UTC)) == 1 + + async with session_factory() as session: + old_epoch = await session.get(TerminalOutcomeEpoch, 0) + assert old_epoch is not None + old_epoch.coverage_end_day = WEEK + timedelta(days=2) + old_epoch.current_slot = None + session.add( + TerminalOutcomeEpoch( + epoch=1, + coverage_start_day=WEEK + timedelta(days=4), + current_slot=1, + ) + ) + session.add(_outcome("safe-old", WEEK + timedelta(days=2))) + session.add(_outcome("new-epoch", WEEK + timedelta(days=4))) + await session.commit() + + concurrent = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + results = await asyncio.gather( + producer.produce_once(now=datetime(2026, 9, 6, 0, 1, tzinfo=UTC)), + concurrent.produce_once(now=datetime(2026, 9, 6, 0, 2, tzinfo=UTC)), + ) + assert sum(results) == 2 + async with session_factory() as session: + rows = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + ).all() + pending = [row for row in rows if row.status == "pending"] + assert [(row.epoch, row.finalized) for row in pending] == [ + (0, True), + (1, False), + ] + assert pending[0].d_tag != pending[1].d_tag + old_payload = json.loads(json.loads(pending[0].frame)[1]["content"]) + assert old_payload["days"][(WEEK + timedelta(days=2)).isoformat()][0] == 1 + + +@pytest.mark.asyncio +async def test_late_change_after_unsent_ordinary_corrects_attempted_predecessor( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + session.add(_outcome("first", WEEK)) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + assert await producer.produce_once(now=datetime(2026, 9, 1, 0, 1, tzinfo=UTC)) == 1 + async with session_factory() as session: + first = (await session.exec(select(AnalyticsV2Outbox))).one() + first.first_send_attempt_at_ms = 3 + await session.commit() + + next_day = datetime(2026, 9, 2, 0, 1, tzinfo=UTC) + assert await producer.produce_once(now=next_day) == 1 + async with session_factory() as session: + session.add(_outcome("late-first-day", WEEK, tokens=2)) + await session.commit() + assert await producer.produce_once(now=next_day) == 1 + + async with session_factory() as session: + rows = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + ).all() + assert [row.status for row in rows] == ["superseded", "superseded", "pending"] + payload = json.loads(json.loads(rows[2].frame)[1]["content"]) + assert payload["corrects"] == rows[0].event_id + assert payload["days"][WEEK.isoformat()][0] == 2 + + +@pytest.mark.asyncio +async def test_concurrent_producers_commit_one_ordinary_semantic_slot( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + session.add(_outcome("first", WEEK)) + await session.commit() + first_producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + second_producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + + results = await asyncio.gather( + first_producer.produce_once(now=datetime(2026, 9, 1, 0, 1, tzinfo=UTC)), + second_producer.produce_once(now=datetime(2026, 9, 1, 0, 2, tzinfo=UTC)), + ) + assert sorted(results) == [0, 1] + async with session_factory() as session: + rows = (await session.exec(select(AnalyticsV2Outbox))).all() + assert len(rows) == 1 + assert rows[0].status == "pending" + assert rows[0].semantic_slot == f"ordinary:0:{WEEK.isoformat()}" + + +@pytest.mark.asyncio +async def test_new_epoch_uses_a_distinct_coordinate_from_prior_epoch_same_week( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + aggregate = aggregate_ledger_week( + [], + epoch=0, + epoch_coverage_start=WEEK, + epoch_coverage_end=None, + week=WEEK, + today_utc=WEEK + timedelta(days=2), + ) + assert aggregate is not None + prior_created_at = 4_000_000_000 + prior = encode_week_event( + aggregate, + private_key_hex=PRIVATE_KEY, + provider_d="provider", + created_at=prior_created_at, + ) + await enqueue_signed_event(session_factory, prior, stored_at_ms=3) + + await transition_analytics_v2_sharing( + session_factory, + enabled=False, + at_ms=_at_ms(WEEK + timedelta(days=2)), + ) + await activate_analytics_v2_sharing( + session_factory, + coverage_day=WEEK + timedelta(days=3), + at_ms=5, + ) + async with session_factory() as session: + session.add(_outcome("new-epoch", WEEK + timedelta(days=4))) + await session.commit() + + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + assert await producer.produce_once(now=datetime(2026, 9, 6, 0, 1, tzinfo=UTC)) == 1 + async with session_factory() as session: + rows = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + ).all() + assert {row.epoch for row in rows} == {0, 1} + assert len({row.d_tag for row in rows}) == 2 + latest = next(row for row in rows if row.epoch == 1) + payload = json.loads(json.loads(latest.frame)[1]["content"]) + assert "corrects" not in payload + + +@pytest.mark.asyncio +async def test_multi_day_disable_excludes_gap_and_never_replays_old_epoch( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + session.add(_outcome("before-disable", WEEK)) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + assert await producer.produce_once(now=datetime(2026, 9, 2, 0, 1, tzinfo=UTC)) == 1 + + await transition_analytics_v2_sharing( + session_factory, + enabled=False, + at_ms=_at_ms(WEEK + timedelta(days=2)), + ) + reenable_day = WEEK + timedelta(days=4) + async with session_factory() as session: + for offset in (2, 3, 4): + session.add(_outcome(f"gap-{offset}", WEEK + timedelta(days=offset))) + session.add(_outcome("after-enable", WEEK + timedelta(days=5))) + await session.commit() + activation = await activate_analytics_v2_sharing( + session_factory, + coverage_day=reenable_day, + at_ms=_at_ms(reenable_day), + ) + assert activation.state.active_epoch_floor == 1 + + assert await producer.produce_once(now=datetime(2026, 9, 7, 0, 1, tzinfo=UTC)) == 1 + async with session_factory() as session: + epochs = ( + await session.exec( + select(TerminalOutcomeEpoch).order_by(col(TerminalOutcomeEpoch.epoch)) + ) + ).all() + rows = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + ).all() + assert [ + (epoch.epoch, epoch.coverage_start_day, epoch.coverage_end_day) + for epoch in epochs + ] == [ + # Private coverage continued while public sharing was off. + (0, WEEK, WEEK + timedelta(days=3)), + (1, WEEK + timedelta(days=5), None), + ] + assert [(row.epoch, row.status) for row in rows] == [ + (0, "cancelled"), + (1, "pending"), + ] + payload = json.loads(json.loads(rows[1].frame)[1]["content"]) + assert set(payload["days"]) == { + (WEEK + timedelta(days=5)).isoformat(), + (WEEK + timedelta(days=6)).isoformat(), + } + + +@pytest.mark.asyncio +async def test_publisher_loop_performs_no_production_or_send_while_disabled( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + sends = 0 + + async def sender( + target: RelayTarget, + event_id: str, + frame: bytes, + is_active: Callable[[], Awaitable[bool]], + ) -> RelaySendResult: + nonlocal sends + sends += 1 + return RelaySendResult(True, True) + + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + delivery = AnalyticsV2Delivery( + session_factory, operator_relays=("wss://relay.valid.net",), sender=sender + ) + task = asyncio.create_task( + run_analytics_v2_publisher(producer, delivery, interval_seconds=0.01) + ) + await asyncio.sleep(0.03) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + async with session_factory() as session: + result = await session.exec(select(AnalyticsV2Outbox)) + assert result.all() == [] + assert sends == 0 + + +@pytest.mark.asyncio +async def test_model_only_correction_keeps_totals_and_updates_daily_partition( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + session.add(_outcome("first", WEEK)) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + now = datetime(2026, 9, 7, 0, 1, tzinfo=UTC) + assert await producer.produce_once(now=now) == 1 + async with session_factory() as session: + original = (await session.exec(select(AnalyticsV2Outbox))).one() + original.first_send_attempt_at_ms = 3 + outcome = await session.get(TerminalOutcome, "first") + assert outcome is not None + outcome.model_identifier = "model/corrected" + await session.commit() + assert await producer.produce_once(now=now) == 1 + assert await producer.produce_once(now=now) == 0 + async with session_factory() as session: + rows = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + ).all() + before, after = [json.loads(json.loads(row.frame)[1]["content"]) for row in rows] + assert before["days"] == after["days"] + assert after["corrects"] == original.event_id + assert "model/corrected" in after["daily_models"][WEEK.isoformat()] + + +@pytest.mark.asyncio +async def test_database_groups_requests_without_losing_usage_provenance( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + for index in range(40): + session.add(_outcome(str(index), WEEK, tokens=index + 1)) + estimated = _outcome("estimated", WEEK, tokens=5) + estimated.input_source = "estimated" + session.add(estimated) + missing = _outcome("missing", WEEK, tokens=0) + missing.input_source = "missing" + missing.output_source = "missing" + session.add(missing) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + _, grouped = await producer._load_eligible_epochs(0, WEEK + timedelta(days=1)) + assert len(grouped) == 3 + assert sum(row.completed_requests for row in grouped) == 42 + assert await producer.produce_once(now=datetime(2026, 9, 1, 0, 1, tzinfo=UTC)) == 1 + async with session_factory() as session: + stored = (await session.exec(select(AnalyticsV2Outbox))).one() + payload = json.loads(json.loads(stored.frame)[1]["content"]) + values = payload["days"][WEEK.isoformat()] + assert values[0] == 42 + assert values[1] == 40 + assert values[5] == sum(range(1, 41)) + 5 + assert values[9] == sum(range(1, 41)) + 5 + assert values[10] == 1 + assert values[14] == 1 + assert values[18:] == [40, 2 * sum(range(1, 41))] + + +@pytest.mark.asyncio +async def test_database_grouping_keeps_measured_cache_cohort_separate( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + rows = [_outcome("plain-a", WEEK, tokens=10), _outcome("plain-b", WEEK, tokens=20)] + rows.append(_outcome("measured-zero", WEEK, tokens=0)) + cached = _outcome("cached", WEEK, tokens=30) + cached.cache_read_source = cached.cache_creation_source = "reported" + cached.cache_read_input_tokens = 8 + cached.cache_creation_input_tokens = 2 + rows.append(cached) + for side in ("cache_read", "cache_creation"): + missing_cache = _outcome(side, WEEK, tokens=100) + setattr(missing_cache, f"{side}_input_tokens", 5) + rows.append(missing_cache) + estimated = _outcome("estimated", WEEK, tokens=40) + estimated.input_source = "estimated" + rows.append(estimated) + input_only = _outcome("input-only", WEEK, tokens=50) + input_only.output_source = "missing" + input_only.output_tokens = 0 + rows.append(input_only) + output_only = _outcome("output-only", WEEK, tokens=60) + output_only.input_source = "missing" + output_only.input_tokens = 0 + rows.append(output_only) + async with session_factory() as session: + session.add_all(rows) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + _, grouped = await producer._load_eligible_epochs(0, WEEK + timedelta(days=1)) + assert len(grouped) == 6 + assert sum(row.completed_requests for row in grouped) == 9 + assert await producer.produce_once(now=datetime(2026, 9, 1, 0, 1, tzinfo=UTC)) == 1 + async with session_factory() as session: + stored = (await session.exec(select(AnalyticsV2Outbox))).one() + payload = json.loads(json.loads(stored.frame)[1]["content"]) + values = payload["days"][WEEK.isoformat()] + assert values[0] == 9 + assert values[18:] == [4, 130] + assert payload["daily_models"][WEEK.isoformat()]["model/served"] == values + + +@pytest.mark.asyncio +async def test_database_report_keeps_each_usage_source_apart( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + for source in ("reported", "estimated", "missing"): + row = _outcome(source, WEEK, tokens=10) + row.model_identifier = source + row.cache_read_input_tokens = 6 + row.cache_creation_input_tokens = 4 + for name in ("input", "output", "cache_read", "cache_creation"): + setattr(row, f"{name}_source", source) + session.add(row) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + assert await producer.produce_once(now=datetime(2026, 9, 1, 0, 1, tzinfo=UTC)) == 1 + async with session_factory() as session: + stored = (await session.exec(select(AnalyticsV2Outbox))).one() + payload = json.loads(json.loads(stored.frame)[1]["content"]) + models = payload["daily_models"][WEEK.isoformat()] + assert models["reported"][1:5] == [1] * 4 + assert models["reported"][18:] == [1, 30] + assert models["estimated"][10:14] == [1] * 4 + assert models["estimated"][18:] == [0, 0] + assert models["missing"][14:18] == [1] * 4 + assert models["missing"][18:] == [0, 0] + assert payload["days"][WEEK.isoformat()][18:] == [1, 30] + + +@pytest.mark.asyncio +async def test_measured_cohort_change_corrects_an_otherwise_identical_report( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + measured = _outcome("measured", WEEK, tokens=10) + missing = _outcome("missing", WEEK, tokens=0) + missing.input_source = missing.output_source = "missing" + async with session_factory() as session: + session.add_all([measured, missing]) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + now = datetime(2026, 9, 1, 0, 1, tzinfo=UTC) + assert await producer.produce_once(now=now) == 1 + async with session_factory() as session: + original = (await session.exec(select(AnalyticsV2Outbox))).one() + original.first_send_attempt_at_ms = 3 + input_only = await session.get(TerminalOutcome, "measured") + output_only = await session.get(TerminalOutcome, "missing") + assert input_only is not None and output_only is not None + input_only.output_source = "missing" + input_only.output_tokens = 0 + output_only.output_source = "reported" + output_only.output_tokens = 10 + await session.commit() + assert await producer.produce_once(now=now) == 1 + assert await producer.produce_once(now=now) == 0 + async with session_factory() as session: + versions = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + ).all() + before, after = [ + json.loads(json.loads(row.frame)[1]["content"]) for row in versions + ] + assert before["days"][WEEK.isoformat()][:18] == after["days"][WEEK.isoformat()][:18] + assert before["days"][WEEK.isoformat()][18:] == [1, 20] + assert after["days"][WEEK.isoformat()][18:] == [0, 0] + assert after["corrects"] == versions[0].event_id + + +@pytest.mark.asyncio +async def test_history_limit_preserves_the_oldest_whole_week( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + now = datetime(2027, 9, 3, 0, 1, tzinfo=UTC) + cutoff = now.date() - timedelta(days=365) + monday = cutoff - timedelta(days=cutoff.weekday()) + assert monday < cutoff + async with session_factory() as session: + session.add(_outcome("before-window-within-week", monday, tokens=9)) + session.add(_outcome("inside-window", cutoff, tokens=3)) + session.add(_outcome("too-old", monday - timedelta(days=1), tokens=100)) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + assert await producer.produce_once(now=now) > 0 + assert await producer.produce_once(now=now) == 0 + async with session_factory() as session: + rows = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.week)) + ) + ).all() + assert rows[0].week == monday + payload = json.loads(json.loads(rows[0].frame)[1]["content"]) + assert payload["coverage_start"] == monday.isoformat() + assert payload["days"][monday.isoformat()][9] == 9 + assert payload["days"][cutoff.isoformat()][9] == 3 + + +@pytest.mark.asyncio +async def test_private_writer_continues_through_public_sharing_toggles( + session_factory: async_sessionmaker[AsyncSession], + monkeypatch: pytest.MonkeyPatch, +) -> None: + from routstr.core import terminal_outcomes as ledger + + clock = [_at_ms(WEEK - timedelta(days=1))] + writer = ledger.TerminalOutcomeWriter( + session_factory=session_factory, + clock=lambda: clock[0], + heartbeat_seconds=0.01, + lease_timeout_seconds=100, + ) + monkeypatch.setattr(ledger, "terminal_outcome_writer", writer) + assert await writer.start() + try: + await _activate(session_factory) + clock[0] = _at_ms(WEEK) + ledger.record_terminal_outcome( + ledger.TerminalOutcomeContext("before", "model/served"), + input_tokens=1, + output_tokens=1, + revenue_msats=1, + cache_read_input_tokens=0, + cache_creation_input_tokens=0, + terminal_at_ms=clock[0], + ) + assert await writer.flush() + clock[0] = _at_ms(WEEK + timedelta(days=1)) + await transition_analytics_v2_sharing( + session_factory, enabled=False, at_ms=clock[0] + ) + assert writer.running + ledger.record_terminal_outcome( + ledger.TerminalOutcomeContext("private", "model/served"), + input_tokens=2, + output_tokens=2, + revenue_msats=2, + cache_read_input_tokens=0, + cache_creation_input_tokens=0, + terminal_at_ms=clock[0], + ) + assert await writer.flush() + clock[0] = _at_ms(WEEK + timedelta(days=2)) + activation = await activate_analytics_v2_sharing( + session_factory, + coverage_day=WEEK + timedelta(days=2), + at_ms=clock[0], + ) + assert activation.transitioned + assert writer.running + ledger.record_terminal_outcome( + ledger.TerminalOutcomeContext("after", "model/served"), + input_tokens=3, + output_tokens=3, + revenue_msats=3, + cache_read_input_tokens=0, + cache_creation_input_tokens=0, + terminal_at_ms=clock[0], + ) + assert await writer.flush() + async with session_factory() as session: + rows = (await session.exec(select(TerminalOutcome))).all() + assert {row.outcome_id for row in rows} == { + ledger._outcome_id(name) for name in ("before", "private", "after") + } + assert sum(row.revenue_msats for row in rows) == 6 + finally: + assert await writer.stop() + + +@pytest.mark.asyncio +async def test_midnight_publication_waits_for_every_writer_to_drain( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + from routstr.core.db import TerminalOutcomeWriterRun + + await _activate(session_factory) + day_after = WEEK + timedelta(days=2) + async with session_factory() as session: + session.add(_outcome("saved-first-day", WEEK)) + session.add( + TerminalOutcomeWriterRun( + run_id="other-worker", + status="active", + started_at_ms=_at_ms(WEEK), + heartbeat_at_ms=_at_ms(day_after), + flushed_through_ms=_at_ms(WEEK + timedelta(days=1)), + ) + ) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + now = datetime(2026, 9, 2, 0, 1, tzinfo=UTC) + assert await producer.produce_once(now=now) == 1 + async with session_factory() as session: + prior = (await session.exec(select(AnalyticsV2Outbox))).one() + payload = json.loads(json.loads(prior.frame)[1]["content"]) + assert payload["through"] == WEEK.isoformat() + session.add( + _outcome("queued-before-midnight", WEEK + timedelta(days=1), tokens=50) + ) + run = await session.get(TerminalOutcomeWriterRun, "other-worker") + assert run is not None + run.flushed_through_ms = _at_ms(day_after) + await session.commit() + assert await producer.produce_once(now=now) == 1 + async with session_factory() as session: + latest = ( + await session.exec( + select(AnalyticsV2Outbox).where( + col(AnalyticsV2Outbox.status) == "pending" + ) + ) + ).one() + payload = json.loads(json.loads(latest.frame)[1]["content"]) + assert payload["through"] == (WEEK + timedelta(days=1)).isoformat() + assert payload["days"][(WEEK + timedelta(days=1)).isoformat()][9] == 50 + + +@pytest.mark.asyncio +async def test_new_relay_size_limit_replaces_oversize_frames_without_mutating_them( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + async with session_factory() as session: + for index in range(100): + row = _outcome(str(index), WEEK, tokens=index + 1) + row.model_identifier = f"model/{index}/" + "x" * 80 + session.add(row) + await session.commit() + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + now = datetime(2026, 9, 7, 0, 1, tzinfo=UTC) + assert await producer.produce_once(now=now) == 1 + async with session_factory() as session: + original = (await session.exec(select(AnalyticsV2Outbox))).one() + original.first_send_attempt_at_ms = 3 + original_bytes = bytes(original.frame) + await session.commit() + limit = len(original_bytes) // 2 + assert await producer.produce_once(now=now, max_frame_bytes=limit) == 1 + assert await producer.produce_once(now=now, max_frame_bytes=limit) == 0 + async with session_factory() as session: + rows = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + ).all() + assert bytes(rows[0].frame) == original_bytes + assert len(rows[1].frame) <= limit + before, after = [json.loads(json.loads(row.frame)[1]["content"]) for row in rows] + assert before["days"] == after["days"] + assert after["corrects"] == original.event_id + assert after["daily_models"][WEEK.isoformat()]["_other"][0] > 0 + + async with session_factory() as session: + row = _outcome("second-week", WEEK + timedelta(days=7)) + session.add(row) + await session.commit() + assert ( + await producer.produce_once( + now=datetime(2026, 9, 8, 0, 1, tzinfo=UTC), max_frame_bytes=limit + ) + == 1 + ) + async with session_factory() as session: + second_week = ( + await session.exec( + select(AnalyticsV2Outbox).where( + col(AnalyticsV2Outbox.week) == WEEK + timedelta(days=7) + ) + ) + ).one() + assert len(second_week.frame) <= limit + + +@pytest.mark.asyncio +async def test_epoch_closure_finalizes_existing_last_day_without_new_requests( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + await _activate(session_factory) + producer = AnalyticsV2Producer( + session_factory, + private_key_hex=PRIVATE_KEY, + public_key_hex=PUBLIC_KEY, + provider_d="provider", + ) + now = datetime(2026, 9, 2, 0, 1, tzinfo=UTC) + assert await producer.produce_once(now=now) == 1 + async with session_factory() as session: + first = (await session.exec(select(AnalyticsV2Outbox))).one() + first.first_send_attempt_at_ms = 3 + epoch = await session.get(TerminalOutcomeEpoch, 0) + assert epoch is not None + epoch.coverage_end_day = WEEK + timedelta(days=1) + epoch.current_slot = None + await session.commit() + assert await producer.produce_once(now=now) == 1 + assert await producer.produce_once(now=now) == 0 + async with session_factory() as session: + rows = ( + await session.exec( + select(AnalyticsV2Outbox).order_by(col(AnalyticsV2Outbox.created_at)) + ) + ).all() + assert rows[-1].finalized + assert rows[-1].through_day == rows[0].through_day diff --git a/tests/unit/test_terminal_outcomes.py b/tests/unit/test_terminal_outcomes.py index 6a3dbf25..9ffed82c 100644 --- a/tests/unit/test_terminal_outcomes.py +++ b/tests/unit/test_terminal_outcomes.py @@ -796,6 +796,84 @@ async def test_failed_writer_start_cannot_backfill_missed_days_as_zero( assert await writer.stop(timeout=1) +@pytest.mark.parametrize("failed_startup", [False, True]) +async def test_fresh_restart_does_not_publish_zero_for_unattended_days( + ledger: tuple[AsyncEngine, SessionFactory], + failed_startup: bool, +) -> None: + import json + + from nostr_sdk import Keys + + from routstr.core.db import AnalyticsV2Outbox + from routstr.nostr.analytics_v2_delivery import ( + AnalyticsV2Producer, + activate_analytics_v2_sharing, + claim_analytics_v2_identity, + ) + + _, sessions = ledger + first_day, clean_day, recovery_day = ( + date(2026, 9, 1), + date(2026, 9, 3), + date(2026, 9, 7), + ) + clock = MutableClock(_timestamp(first_day)) + pubkey = Keys.parse("11" * 32).public_key().to_hex() + await claim_analytics_v2_identity( + sessions, pubkey=pubkey, provider_d="provider", at_ms=clock.value + ) + await activate_analytics_v2_sharing( + sessions, coverage_day=first_day, at_ms=clock.value + ) + first = TerminalOutcomeWriter(session_factory=sessions, clock=clock) + assert await first.start() + clock.value = _timestamp(clean_day) + assert await first.flush(timeout=1) + assert await first.stop(timeout=1) + + if failed_startup: + + @asynccontextmanager + async def unavailable() -> AsyncGenerator[AsyncSession, None]: + raise OSError("synthetic stats database unavailable at startup") + yield # pragma: no cover + + failed = TerminalOutcomeWriter( + session_factory=unavailable, + clock=MutableClock(_timestamp(clean_day + timedelta(days=1))), + ) + assert not await failed.start() + assert failed.loss_pending + + clock.value = _timestamp(recovery_day) + recovered = TerminalOutcomeWriter(session_factory=sessions, clock=clock) + assert await recovered.start() + producer = AnalyticsV2Producer( + sessions, + private_key_hex="11" * 32, + public_key_hex=pubkey, + provider_d="provider", + ) + try: + assert ( + await producer.produce_once( + now=datetime.fromtimestamp(clock.value / 1000, UTC) + ) + == 1 + ) + async with sessions() as session: + reports = (await session.exec(select(AnalyticsV2Outbox))).all() + reported_days = { + day + for report in reports + for day in json.loads(json.loads(report.frame)[1]["content"])["days"] + } + assert reported_days == {"2026-09-02"} + finally: + assert await recovered.stop(timeout=1) + + async def test_restart_alongside_live_writer_preserves_continuous_coverage( ledger: tuple[AsyncEngine, SessionFactory], ) -> None: