rtt

rtt ¶

NIP-66 RTT metadata container with relay probe capabilities.

Tests a relay's round-trip time by measuring connection open, event read, and event write latencies as defined by NIP-66. Results are stored as millisecond integers alongside detailed logs for each phase.

Note

The RTT test executes three sequential phases:

Open -- measures WebSocket connection establishment time via connect_relay.
Read -- measures time to receive the first event matching a filter via stream_events.
Write -- measures time to publish an event and verify it can be retrieved back from the relay.

If the open phase fails, read and write are automatically marked as failed with the same reason (cascading failure). The write phase includes a verification step that re-fetches the published event to confirm the relay actually stored it.

Classes¶

Nip66RttDependencies ¶

Bases: NamedTuple

Grouped dependencies for RTT probe tests.

Bundles the signing keys, event builder, and read filter required by the RTT measurement phases (open, read, write).

Nip66RttMetadata ¶

Bases: BaseNipMetadata

Container for RTT measurement data and multi-phase probe logs.

Provides the execute() class method that connects to a relay and measures open, read, and write round-trip times.

Warning

The execute() method never raises exceptions for transport errors or missing dependencies. All failures are captured in the Nip66RttMultiPhaseLogs fields. Overlay network relays tested without a proxy URL receive an immediate failure result (all phases marked as failed).

Functions¶

execute `async` `classmethod` ¶

execute(
    relay: Relay,
    deps: Nip66RttDependencies,
    timeout: float | None = None,
    proxy_url: str | None = None,
    *,
    allow_insecure: bool = False,
) -> Self

Test a relay's round-trip times across three phases.

Phases are executed sequentially: open -> read -> write. If the open phase fails, read and write are marked as failed with the same reason (cascading failure).

Parameters:

relay (Relay) –

Relay to test.
deps (Nip66RttDependencies) –

Grouped dependencies (keys, event_builder, read_filter).
timeout (float | None, default: None ) –

Connection timeout in seconds (default: 10.0).
proxy_url (str | None, default: None ) –

Optional SOCKS5 proxy URL for overlay networks.
allow_insecure (bool, default: False ) –

Fall back to unverified SSL (default: False).

Returns:

Self –

An Nip66RttMetadata instance with measurement data and logs.

Source code in src/bigbrotr/nips/nip66/rtt.py

@classmethod
async def execute(
    cls,
    relay: Relay,
    deps: Nip66RttDependencies,
    timeout: float | None = None,  # noqa: ASYNC109
    proxy_url: str | None = None,
    *,
    allow_insecure: bool = False,
) -> Self:
    """Test a relay's round-trip times across three phases.

    Phases are executed sequentially: open -> read -> write.
    If the open phase fails, read and write are marked as failed
    with the same reason (cascading failure).

    Args:
        relay: Relay to test.
        deps: Grouped dependencies (keys, event_builder, read_filter).
        timeout: Connection timeout in seconds (default: 10.0).
        proxy_url: Optional SOCKS5 proxy URL for overlay networks.
        allow_insecure: Fall back to unverified SSL (default: False).

    Returns:
        An ``Nip66RttMetadata`` instance with measurement data and logs.
    """
    timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
    logger.debug("rtt_started relay=%s timeout_s=%s proxy=%s", relay.url, timeout, proxy_url)

    overlay_networks = (NetworkType.TOR, NetworkType.I2P, NetworkType.LOKI)
    if proxy_url is None and relay.network in overlay_networks:
        reason = f"overlay network {relay.network.value} requires proxy"
        return cls(
            data=Nip66RttData(),
            logs=Nip66RttMultiPhaseLogs(
                open_success=False,
                open_reason=reason,
                read_success=False,
                read_reason=reason,
                write_success=False,
                write_reason=reason,
            ),
        )

    rtt_data = cls._empty_rtt_data()
    logs = cls._empty_logs()
    relay_url = RelayUrl.parse(relay.url)

    # Phase 1: Open connection
    client, open_rtt = await cls._test_open(
        relay, deps.keys, proxy_url, timeout, allow_insecure=allow_insecure, logs=logs
    )
    if client is None:
        return cls._build_result(rtt_data, logs)

    rtt_data["rtt_open"] = open_rtt
    logs["open_success"] = True

    try:
        # Phase 2: Read capability
        read_result = await cls._test_read(client, deps.read_filter, timeout, relay.url)
        rtt_data["rtt_read"] = read_result.get("rtt_read")
        logs["read_success"] = read_result["read_success"]
        logs["read_reason"] = read_result.get("read_reason")

        # Phase 3: Write capability
        write_result = await cls._test_write(
            client, deps.event_builder, relay_url, timeout, relay.url
        )
        rtt_data["rtt_write"] = write_result.get("rtt_write")
        logs["write_success"] = write_result["write_success"]
        logs["write_reason"] = write_result.get("write_reason")
    finally:
        await cls._cleanup(client)

    logger.debug(
        "rtt_completed relay=%s rtt_data=%s open_success=%s read_success=%s write_success=%s",
        relay.url,
        rtt_data,
        logs.get("open_success"),
        logs.get("read_success"),
        logs.get("write_success"),
    )
    return cls._build_result(rtt_data, logs)

rtt