diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..a0e9c39 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,50 @@ +name: connector-ci + +on: + pull_request: + push: + branches: [main] + +permissions: + contents: read + +jobs: + relay: + name: Relay connector + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 + + - name: Install Rust 1.95 + uses: dtolnay/rust-toolchain@6c977a6ca4077a0ceb28ffbe03f59d46e9ac8772 + with: + toolchain: 1.95.0 + components: clippy, rustfmt + + - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 + + - name: Verify the packaged conformance vector + run: | + actual="$( + openssl dgst -sha256 relay/tests/fixtures/connector-v1-vectors.json | + awk '{print $NF}' + )" + test "${actual}" = "18d7218047bd77b599c0639b69c4fead31149beccbc13156fd387512bc818b91" + + - name: Format + run: cargo fmt --all -- --check + + - name: Test + run: cargo test --workspace --all-features --locked + + - name: Lint + run: cargo clippy --workspace --all-targets --all-features --locked -- -D warnings + + - name: Build documentation + env: + RUSTDOCFLAGS: -D warnings + run: cargo doc --workspace --no-deps --all-features --locked + + - name: Verify the crates.io package + run: cargo publish --manifest-path relay/Cargo.toml --dry-run --locked diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..65376a4 --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,110 @@ +name: publish-relay-connector + +on: + release: + types: [published] + workflow_dispatch: + inputs: + release_tag: + description: Existing version-matched release tag to resume + required: true + type: string + +permissions: + contents: read + +concurrency: + group: crates-io-publish + cancel-in-progress: false + +jobs: + validate: + if: >- + github.event_name == 'workflow_dispatch' || + ( + github.event_name == 'release' && + github.event.release.prerelease == false && + startsWith(github.event.release.tag_name, 'pocketstation-relay-v') + ) + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 + with: + fetch-depth: 0 + + - name: Install Rust 1.95 + uses: dtolnay/rust-toolchain@6c977a6ca4077a0ceb28ffbe03f59d46e9ac8772 + with: + toolchain: 1.95.0 + components: clippy, rustfmt + + - name: Validate the release tag + env: + EVENT_NAME: ${{ github.event_name }} + RELEASE_TAG: ${{ github.event.release.tag_name || inputs.release_tag }} + run: | + set -euo pipefail + version="$( + cargo metadata --format-version 1 --no-deps | + jq -r '.packages[] | select(.name == "pocketstation-relay") | .version' + )" + expected_tag="pocketstation-relay-v${version}" + if [[ "${RELEASE_TAG}" != "${expected_tag}" ]]; then + echo "release tag ${RELEASE_TAG} must equal ${expected_tag}" >&2 + exit 1 + fi + tag_commit="$(git rev-list -n 1 "${RELEASE_TAG}")" + git fetch origin main + if ! git merge-base --is-ancestor "${tag_commit}" origin/main; then + echo "release commit is not contained in origin/main" >&2 + exit 1 + fi + if [[ "${EVENT_NAME}" == "release" && "${tag_commit}" != "${GITHUB_SHA}" ]]; then + echo "release tag does not resolve to the checked-out commit" >&2 + exit 1 + fi + + - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 + + - name: Verify the package + run: | + actual="$( + openssl dgst -sha256 relay/tests/fixtures/connector-v1-vectors.json | + awk '{print $NF}' + )" + test "${actual}" = "18d7218047bd77b599c0639b69c4fead31149beccbc13156fd387512bc818b91" + cargo fmt --all -- --check + cargo test --workspace --all-features --locked + cargo clippy --workspace --all-targets --all-features --locked -- -D warnings + RUSTDOCFLAGS='-D warnings' cargo doc --workspace --no-deps --all-features --locked + cargo publish --manifest-path relay/Cargo.toml --dry-run --locked + + publish: + needs: validate + runs-on: ubuntu-latest + timeout-minutes: 15 + environment: crates-io + steps: + - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 + + - uses: dtolnay/rust-toolchain@6c977a6ca4077a0ceb28ffbe03f59d46e9ac8772 + with: + toolchain: 1.95.0 + + - name: Publish pocketstation-relay + env: + CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }} + run: | + set -euo pipefail + version="$( + cargo metadata --format-version 1 --no-deps | + jq -r '.packages[] | select(.name == "pocketstation-relay") | .version' + )" + if curl --fail --silent --show-error \ + "https://crates.io/api/v1/crates/pocketstation-relay/${version}" \ + >/dev/null; then + echo "pocketstation-relay ${version} is already published" + exit 0 + fi + cargo publish --manifest-path relay/Cargo.toml --locked diff --git a/Cargo.lock b/Cargo.lock index 7a75c0f..853847b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -866,7 +866,9 @@ checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" [[package]] name = "pocketstation" -version = "1.1.1" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69ef0a8643e866a639647afe98fd83208096983480bcd6978f8e85e2e55a9c4c" dependencies = [ "cc", "hound", @@ -881,7 +883,7 @@ dependencies = [ [[package]] name = "pocketstation-relay" -version = "0.1.1" +version = "0.1.2" dependencies = [ "base64", "pocketstation", diff --git a/README.md b/README.md index 0e8f9cc..fb944b9 100644 --- a/README.md +++ b/README.md @@ -1,166 +1,56 @@ -# PocketStation Connector Registry +# PocketStation connectors -Find the first-party packages that connect a PocketStation Session to an -external service. +Use a Connector to send audio from a PocketStation `Session` to an external +service. Connectors are separate packages, so their network and provider +dependencies do not become Core dependencies. -This repository is the source registry and shared verification workspace for -connectors maintained by the PocketStation project. Each connector is an -independent package with its own protocol scope, compatibility contract, -documentation, tests, and release lifecycle. +## Available connector -The repository root is a catalog. Package-specific setup and implementation -guidance belongs inside each connector directory. +PocketStation Relay is the only first-party connector currently available. -## Available connectors +| Package | Sends | Destination | +|---|---|---| +| [`pocketstation-relay`](https://crates.io/crates/pocketstation-relay) | independent named audio buses | [PocketStation Relay](https://github.com/pocketstation-io/relay) over WebRTC | -| Connector | Package | Direction | Connects to | Release | Documentation | -|---|---|---|---|---|---| -| PocketStation Relay | [`pocketstation-relay`](https://crates.io/crates/pocketstation-relay) | outbound audio | [PocketStation Relay](https://github.com/pocketstation-io/relay) over WebRTC | `0.1.1` | [Guide](relay/README.md) · [Rust API](https://docs.rs/pocketstation-relay) | +Install the Rust packages: -That is the complete first-party registry today. A connector not listed here -does not inherit PocketStation maintenance, compatibility, or evidence claims. - -## Choose a connector - -Use `pocketstation-relay` when you need to publish independent, named audio -buses from a Rust PocketStation Session to PocketStation Relay. - -There is not currently a first-party LiveKit, generic WHIP, OpenAI, Deepgram, -or arbitrary WebRTC connector in this registry. Those services have different -authentication, negotiation, lifecycle, and outcome contracts. Support requires -a dedicated adapter; changing a URL is not sufficient. - -For installation and a complete application-plus-microphone example, go -directly to the [PocketStation Relay connector guide](relay/README.md). - -## What “first-party” means - -A connector in this registry must have all of the following: - -- a named maintainer and an active product or ecosystem requirement; -- a finite, typed configuration contract with secret redaction; -- an explicit provider/protocol compatibility boundary; -- canonical PocketStation Connector and Endpoint lifecycle integration; -- bounded preparation, delivery, cancellation, drain/abort, and shutdown; -- stable provider error classification and observable terminal outcomes; -- executable conformance, saturation, rollback, and failure tests; -- real protocol integration evidence; -- an independently installable package and isolated consumer proof; -- an intentional versioning and compatibility policy. - -Passing component tests does not automatically establish remote production -readiness, every network topology, every platform, or competitive superiority. -Those claims require separately identified evidence. - -## Architecture boundary - -```text -PocketStation Core - Session + Graph + Endpoint lifecycle - ↓ - Connector contract - ↓ -independently packaged provider adapter - ↓ - external service or protocol -``` - -Responsibilities stay deliberately separated: - -| Owner | Responsibility | -|---|---| -| [`pocketstation`](https://github.com/pocketstation-io/pocketstation) | provider-neutral graph, bounded routing, lineage, lifecycle, recording, observations, Connector contract | -| This registry | first-party connector packages, compatibility, conformance, packaging, release ownership | -| Provider/service repository | wire protocol, server behavior, authentication authority, remote delivery | - -Connectors are outbound Endpoint specializations. Inbound media remains a -PocketStation `Source`; transformations remain `Operator`s. A larger -bidirectional integration may compose all three without introducing another -Session or runtime. - -Core never gains a closed provider enum. Adding a connector must not add its -WebRTC, SDK, authentication, or protocol dependencies to Core. - -## Registry policy - -This is a curated first-party registry, not a collection of every possible -integration. - -A proposed connector moves through these stages: - -1. **Scope** — identify a real user workflow and the exact protocol boundary. -2. **Ownership** — assign maintainers, security ownership, and compatibility - responsibilities. -3. **Contract** — declare inputs, capabilities, configuration, credentials, - limits, readiness, errors, and outcomes. -4. **Implementation** — use the canonical Core lifecycle without duplicating - graph, queue, or Session authority. -5. **Conformance** — prove rollback, saturation, discontinuity, cancellation, - drain/abort, failure containment, and exact destruction. -6. **Integration** — exercise the real external service and record the evidence - boundary honestly. -7. **Distribution** — package, inspect, install, and run from an isolated - consumer before release. - -An example or experimental adapter is not promoted into this registry merely -because it compiles. - -## Repository layout - -```text -connectors/ -├── README.md this registry and its policies -├── Cargo.toml shared verification workspace only -└── relay/ - ├── README.md package setup, behavior, and operational limits - ├── Cargo.toml independently released crate - ├── src/ Relay-specific implementation - └── tests/ package and portable-semantics conformance +```bash +cargo add pocketstation pocketstation-relay ``` -Package directories own their user documentation. The repository README owns -only discovery, support status, shared boundaries, and registry policy. +Then follow the [Relay connector guide](relay/README.md) to publish application +and microphone audio as separate buses. -## Versioning and compatibility +See the [release notes](relay/RELEASE_NOTES.md) before upgrading. -Connector packages version independently from this repository and from -PocketStation Core. +## What the Relay connector handles -- Published crate versions and Git tags are immutable. -- Each package declares the Core versions it supports. -- Provider protocol compatibility is proved by that package, not inferred from - Core's trait definitions. -- A breaking provider or public Rust API change requires the package's normal - semantic-versioning process. -- A repository commit is not a release until its package artifact, tag, and - isolated consumer agree on the same source. +The package owns the Relay-specific work: -Current compatibility: +- source capability authentication; +- WebRTC signaling, ICE, DTLS, Opus, and RTP; +- named AudioBus publication; +- finite startup and shutdown deadlines; +- redacted credentials and structured failures. -| Package | Connector version | PocketStation Core | Evidence boundary | -|---|---:|---:|---| -| `pocketstation-relay` | `0.1.1` | `1.1.1` | component and same-host integration; remote production breadth not implied | +PocketStation Core continues to own capture, graph compilation, bounded +routing, recording, and Session lifecycle. -## Develop and verify the registry - -Run the complete workspace gate from the repository root: - -```bash -cargo fmt --all -- --check -cargo test --workspace --all-targets --all-features --locked -cargo clippy --workspace --all-targets --all-features --locked -- -D warnings -RUSTDOCFLAGS="-D warnings" cargo doc --workspace --no-deps -``` +## Current limits -Package and real-service verification requirements remain in each connector's -own guide and release process. +There are no first-party LiveKit, OpenAI, Deepgram, Twilio, generic WHIP, or +generic WebRTC connectors in this repository. Each service requires its own +authentication, media negotiation, lifecycle, and error handling; changing a +URL is not enough. -## Community connectors +The Relay connector's published evidence covers component and same-host +integration tests. It does not claim every NAT topology, platform, or production +load. -Third-party connectors can implement the same open Core contract without -living in this repository. Their maintainers own distribution, provider -compatibility, security response, and support claims. +## Build another connector -If a community connector is later considered for first-party support, it must -pass the registry policy above. Adoption is an ownership commitment—not only a -directory move. +Third-party packages can implement PocketStation's open Connector API without +living in this repository. Start with the +[Core Connector guide](https://github.com/pocketstation-io/pocketstation/blob/main/docs/guides/connectors.md). +The package author owns provider compatibility, security updates, distribution, +and support. diff --git a/relay/Cargo.toml b/relay/Cargo.toml index ba2a4c1..1cf3210 100644 --- a/relay/Cargo.toml +++ b/relay/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "pocketstation-relay" -version = "0.1.1" +version = "0.1.2" edition.workspace = true rust-version.workspace = true license.workspace = true @@ -9,7 +9,13 @@ readme = "README.md" description = "Bounded WebRTC relay publishing connector for PocketStation" keywords = ["audio", "realtime", "relay", "webrtc", "pocketstation"] categories = ["multimedia::audio", "network-programming"] -include = ["/Cargo.toml", "/README.md", "/src/**", "/tests/**"] +include = [ + "/Cargo.toml", + "/README.md", + "/RELEASE_NOTES.md", + "/src/**", + "/tests/**", +] [package.metadata.docs.rs] default-target = "x86_64-unknown-linux-gnu" @@ -20,7 +26,7 @@ workspace = true [dependencies] base64 = "0.22" -pocketstation = { path = "../../pocketstation", version = "1.1.1", default-features = false } +pocketstation = { version = "1.1.3", default-features = false } serde = { version = "1", features = ["derive"] } serde_json = "1" str0m = "0.20" @@ -29,4 +35,4 @@ tungstenite = { version = "0.29", features = ["native-tls"] } url = "2" [dev-dependencies] -pocketstation = { path = "../../pocketstation", version = "1.1.1", default-features = false, features = ["conformance-fixtures"] } +pocketstation = { version = "1.1.3", default-features = false, features = ["conformance-fixtures"] } diff --git a/relay/README.md b/relay/README.md index 5bd9933..7a06837 100644 --- a/relay/README.md +++ b/relay/README.md @@ -1,7 +1,7 @@ # PocketStation Relay connector Publish independent PocketStation audio stems to PocketStation Relay through -the canonical `pocketstation::connector` lifecycle. +the `pocketstation::connector` lifecycle. `pocketstation-relay` owns the client-side Opus, RTP, WebRTC, and Relay signaling implementation. PocketStation Core continues to own graph @@ -24,7 +24,12 @@ You also need a reachable valid RelaySession source credential. The crate never starts hidden infrastructure. -## Publish two source-aware buses +Before running the example, create a `RelaySession` in the control plane (or a +standalone Relay), keep its source capability, and confirm that the Relay URL +is reachable. The example uses placeholder values; it does not contact a +PocketStation-operated service by default. + +## Declare two bus publications ```rust,no_run use pocketstation::connector::ConnectorSecret; @@ -32,9 +37,12 @@ use pocketstation::{ApplicationSelector, EdgeContract, Session, Source}; use pocketstation_relay::{RelayConnector, RelayRouteConfiguration}; # fn main() -> Result<(), Box> { +let application_name = std::env::args() + .nth(1) + .ok_or("usage: publish_to_relay ")?; let session = Session::builder().recording_root("recordings").build(); let application = session.capture(Source::application( - ApplicationSelector::name("PocketStation Demo"), + ApplicationSelector::name(application_name), ))?; let microphone = session.capture(Source::microphone_default())?; @@ -78,6 +86,12 @@ assert!(stop.is_success()); # } ``` +This program demonstrates configuration, grouped preparation, Session start, +and joined shutdown. Replace the placeholder URL, Session ID, and source +capability before running it. A receiver-visible publication also requires the +application and microphone to produce media while the Session remains running; +the snippet stops immediately so it can stay focused on declaration. + The two route configurations share a Relay origin, Session, credential, publisher group, ICE configuration, and startup deadline. They therefore prepare and run as one publisher while retaining distinct source, stem, route, @@ -144,15 +158,15 @@ Session prepare The crate does not create a second graph, Session, queue policy, or retry engine. Provider retry/reconnect behavior must remain finite and explicit. -## Inspect publication outcomes +## Inspect publication results `RelayConnector` retains bounded publication receipts. A receipt key identifies the endpoint/route publication, and the final result reports stable outcome state and unit-bearing statistics. Missing receipts are not interpreted as success. -Use Core's Session and route observations for generic delivery truth; use the -connector receipt for Relay-specific publication truth. +Use Core's Session and route observations for route delivery. Use the connector +receipt for Relay-specific publication results. ## Current boundaries diff --git a/relay/RELEASE_NOTES.md b/relay/RELEASE_NOTES.md new file mode 100644 index 0000000..e00b824 --- /dev/null +++ b/relay/RELEASE_NOTES.md @@ -0,0 +1,27 @@ +# PocketStation Relay connector release notes + +## 0.1.2 — Stop cancelled output before Relay sends it + +When a person interrupts generated speech, stopping the provider task is not +enough. Encoded audio may still be waiting in the Connector queue and can reach +the receiver after the application has moved to a new response. + +PocketStation Relay 0.1.2 carries Core's output generation identity through PCM +packetization and Opus encoding. If the application cancels that output, the +Connector discards its queued frames before RTP publication while other +AudioBuses continue. + +This release requires `pocketstation 1.1.3`, which introduced output generation +ownership for application-provided audio. + +Cancellation cannot recall RTP packets already sent to Relay or audio already +buffered by a receiver. A complete interruption path must also clear receiver +playout when that receiver provides the capability. + +### Upgrade + +This update requires no Connector configuration migration. + +```console +cargo update -p pocketstation -p pocketstation-relay +``` diff --git a/relay/src/audio/opus_worker.rs b/relay/src/audio/opus_worker.rs index ebd2711..d663995 100644 --- a/relay/src/audio/opus_worker.rs +++ b/relay/src/audio/opus_worker.rs @@ -1,6 +1,7 @@ //! Opus encoder thread and the frame/counter types it produces. use pocketstation::codec::{OpusConfig, OpusEncodeError, OpusEncoder, OPUS_MAX_PACKET_BYTES}; +use pocketstation::OutputGeneration; #[cfg(any(target_os = "macos", target_os = "windows", target_os = "linux"))] use std::time::Instant; use str0m::media::{Frequency, MediaTime}; @@ -27,6 +28,9 @@ pub(crate) trait EncodableFrame: Send + 'static { /// Monotonic nanoseconds at which the first sample of this frame was captured. /// Returns 0 when the source has no capture timestamp (e.g. sine generator). fn capture_timestamp_ns(&self) -> u64; + fn output_generation(&self) -> Option { + None + } } /// Zero-copy bridge: `AudioFrame`'s pool-backed buffer stays alive until the @@ -74,6 +78,10 @@ impl EncodableFrame for pocketstation::EndpointAudioFrame { fn channels(&self) -> u8 { self.channels() } + + fn output_generation(&self) -> Option { + self.output_generation().cloned() + } } /// Budget for the age of a frame at encode time: 2× the 20 ms frame period. @@ -100,8 +108,12 @@ pub(crate) struct EncoderCounters { pub(crate) opus_encode_errors: AtomicU64, /// Frames dropped because the encoder→publisher channel was full. pub(crate) encoded_channel_drops: AtomicU64, - /// Encoded frames superseded by a fresher frame after an RTC-loop stall. + /// Encoded frames discarded in favor of a fresher frame after an RTC-loop stall. pub(crate) publisher_stale_drops: AtomicU64, + /// Output frames discarded after their application operation was cancelled. + pub(crate) cancelled_output_frames: AtomicU64, + /// Partial PCM samples removed when output ownership changes. + pub(crate) cancelled_output_samples: AtomicU64, /// Cumulative capture age in nanoseconds (for mean computation). pub(crate) capture_age_sum_ns: AtomicU64, /// Maximum capture age observed, in nanoseconds. @@ -128,6 +140,7 @@ pub(crate) struct EncodedAudioFrame { /// 0 when `capture_timestamp_ns` is 0. #[allow(dead_code)] pub(crate) capture_age_ns: u64, + pub(crate) output_generation: Option, } /// Delivery behavior at the bounded encoder-to-publisher boundary. @@ -165,6 +178,16 @@ fn deliver_encoded_frame( delivery_policy: EncodedDeliveryPolicy, counters: &EncoderCounters, ) -> bool { + if encoded_frame + .output_generation + .as_ref() + .is_some_and(|generation| !generation.is_active()) + { + counters + .cancelled_output_frames + .fetch_add(1, Ordering::Relaxed); + return true; + } match delivery_policy { EncodedDeliveryPolicy::PreserveWithBackpressure => encoded_tx.send(encoded_frame).is_ok(), EncodedDeliveryPolicy::DropNewest => match encoded_tx.try_send(encoded_frame) { @@ -201,10 +224,20 @@ impl EncoderWorker { &mut self, samples: &[f32], capture_timestamp_ns: u64, + output_generation: Option, encoded_tx: &mpsc::SyncSender, delivery_policy: EncodedDeliveryPolicy, counters: &EncoderCounters, ) -> bool { + if output_generation + .as_ref() + .is_some_and(|generation| !generation.is_active()) + { + counters + .cancelled_output_frames + .fetch_add(1, Ordering::Relaxed); + return true; + } let square_sum: f32 = samples.iter().map(|sample| sample * sample).sum(); let sample_count = u16::try_from(samples.len()).unwrap_or(u16::MAX); let rms = if sample_count == 0 { @@ -242,6 +275,7 @@ impl EncoderWorker { audio_level, capture_timestamp_ns, capture_age_ns, + output_generation, }; self.rtp_sample_count = self.rtp_sample_count.saturating_add(self.rtp_step_samples); deliver_encoded_frame(encoded_tx, encoded_frame, delivery_policy, counters) @@ -302,21 +336,48 @@ pub(crate) fn spawn_opus_encoder( counters.opus_encode_errors.fetch_add(1, Ordering::Relaxed); return; }; + let mut packet_output: Option = None; while let Ok(frame) = frame_rx.recv() { if frame.sample_rate_hz() != 48_000 || frame.channels() != channels { counters.opus_encode_errors.fetch_add(1, Ordering::Relaxed); continue; } + let output_generation = frame.output_generation(); + if output_generation + .as_ref() + .is_some_and(|generation| !generation.is_active()) + { + counters + .cancelled_output_frames + .fetch_add(1, Ordering::Relaxed); + continue; + } + let output_generation_id = output_generation + .as_ref() + .map(|generation| generation.id().get()); let samples = frame.samples(); - let Ok(discarded) = packetizer.begin_frame(frame.sequence_number(), samples.len()) - else { + let Ok(boundary) = packetizer.begin_frame( + frame.sequence_number(), + samples.len(), + output_generation_id, + ) else { counters.opus_encode_errors.fetch_add(1, Ordering::Relaxed); continue; }; - counters - .normalization_partial_samples_discarded - .fetch_add(discarded as u64, Ordering::Relaxed); + if packetizer.output_generation_id() + != packet_output.as_ref().map(|value| value.id().get()) + { + packet_output = output_generation; + } + counters.normalization_partial_samples_discarded.fetch_add( + boundary.discontinuity_discarded_samples as u64, + Ordering::Relaxed, + ); + counters.cancelled_output_samples.fetch_add( + boundary.cancelled_output_discarded_samples as u64, + Ordering::Relaxed, + ); let mut offset = 0; while offset < samples.len() { @@ -326,6 +387,7 @@ pub(crate) fn spawn_opus_encoder( if !worker.encode_and_deliver( packet, timestamp_ns, + packet_output.clone(), &encoded_tx, delivery_policy, counters.as_ref(), @@ -345,6 +407,7 @@ pub(crate) fn spawn_opus_encoder( let _ = worker.encode_and_deliver( packet, timestamp_ns, + packet_output, &encoded_tx, delivery_policy, counters.as_ref(), @@ -358,6 +421,24 @@ pub(crate) fn spawn_opus_encoder( #[cfg(test)] mod tests { use super::*; + use pocketstation::{AudioInputConfig, OutputCancelResult, SampleFormat, SampleSpec, Session}; + + fn output_generation() -> OutputGeneration { + let session = Session::builder() + .sample_spec(SampleSpec::new(48_000, 1, SampleFormat::F32Interleaved)) + .build(); + let input = session + .audio_input( + AudioInputConfig::new( + SampleSpec::new(48_000, 1, SampleFormat::F32Interleaved), + 2, + 960, + ) + .expect("valid output input"), + ) + .expect("output input"); + input.begin_output_generation().expect("output generation") + } fn encoded_frame(sequence: u8) -> EncodedAudioFrame { EncodedAudioFrame { @@ -368,6 +449,7 @@ mod tests { audio_level: None, capture_timestamp_ns: 0, capture_age_ns: 0, + output_generation: None, } } @@ -430,6 +512,29 @@ mod tests { assert_eq!(counters.encoded_channel_drops.load(Ordering::Relaxed), 1); } + #[test] + fn given_cancelled_output_when_encoded_delivery_runs_then_frame_is_discarded() { + let counters = EncoderCounters::default(); + let (encoded_tx, encoded_rx) = mpsc::sync_channel(1); + let generation = output_generation(); + let mut frame = encoded_frame(1); + frame.output_generation = Some(generation.clone()); + + assert_eq!(generation.cancel(), OutputCancelResult::Cancelled); + assert!(deliver_encoded_frame( + &encoded_tx, + frame, + EncodedDeliveryPolicy::PreserveWithBackpressure, + &counters, + )); + + assert!(matches!( + encoded_rx.try_recv(), + Err(mpsc::TryRecvError::Empty) + )); + assert_eq!(counters.cancelled_output_frames.load(Ordering::Relaxed), 1); + } + // Given an EncodedAudioFrame with capture_timestamp_ns == 0, // When capture_age_ns is read, // Then it is also 0 (unknown-source frames carry no age). @@ -443,6 +548,7 @@ mod tests { audio_level: None, capture_timestamp_ns: 0, capture_age_ns: 0, + output_generation: None, }; assert_eq!(frame.capture_timestamp_ns, 0); assert_eq!(frame.capture_age_ns, 0); @@ -466,6 +572,7 @@ mod tests { audio_level: None, capture_timestamp_ns: capture_ts, capture_age_ns: age_ns, + output_generation: None, }; assert!( frame.capture_age_ns > 0, diff --git a/relay/src/audio/pcm_packetizer.rs b/relay/src/audio/pcm_packetizer.rs index f03a8f2..c0bf2dd 100644 --- a/relay/src/audio/pcm_packetizer.rs +++ b/relay/src/audio/pcm_packetizer.rs @@ -17,6 +17,12 @@ pub(crate) enum PcmPacketizerError { MisalignedInput { samples: usize, channels: u8 }, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct PcmPacketizerBegin { + pub(crate) discontinuity_discarded_samples: usize, + pub(crate) cancelled_output_discarded_samples: usize, +} + /// Bounded assembler for one 20 ms, 48 kHz mono or stereo transport packet. pub(crate) struct PcmPacketizer { samples: [f32; MAX_INTERLEAVED_SAMPLES], @@ -25,6 +31,7 @@ pub(crate) struct PcmPacketizer { channels: u8, packet_timestamp_ns: u64, expected_sequence_number: Option, + output_generation_id: Option, } impl PcmPacketizer { @@ -42,32 +49,46 @@ impl PcmPacketizer { channels, packet_timestamp_ns: 0, expected_sequence_number: None, + output_generation_id: None, }) } - /// Starts one source frame and returns the number of partial samples - /// discarded at a sequence discontinuity. + /// Starts one source frame and reports any partial packet removed at its boundary. pub(crate) fn begin_frame( &mut self, sequence_number: u64, sample_count: usize, - ) -> Result { + output_generation_id: Option, + ) -> Result { if !sample_count.is_multiple_of(usize::from(self.channels)) { return Err(PcmPacketizerError::MisalignedInput { samples: sample_count, channels: self.channels, }); } - let discarded = if self + let sequence_changed = self .expected_sequence_number - .is_some_and(|expected| expected != sequence_number) - { + .is_some_and(|expected| expected != sequence_number); + let output_changed = self.written > 0 && self.output_generation_id != output_generation_id; + let discarded = if sequence_changed || output_changed { self.discard_partial() } else { 0 }; self.expected_sequence_number = Some(sequence_number.saturating_add(1)); - Ok(discarded) + self.output_generation_id = output_generation_id; + Ok(PcmPacketizerBegin { + discontinuity_discarded_samples: if sequence_changed && !output_changed { + discarded + } else { + 0 + }, + cancelled_output_discarded_samples: if output_changed { discarded } else { 0 }, + }) + } + + pub(crate) const fn output_generation_id(&self) -> Option { + self.output_generation_id } /// Copies as much of `input` as fits and returns the number of interleaved @@ -151,9 +172,15 @@ mod tests { #[test] fn arbitrary_stereo_callback_widths_form_exact_twenty_millisecond_packets() { let mut packetizer = PcmPacketizer::new(48_000, 2).unwrap(); - assert_eq!(packetizer.begin_frame(1, 1_024).unwrap(), 0); + assert_eq!( + packetizer.begin_frame(1, 1_024, None).unwrap(), + PcmPacketizerBegin::default() + ); assert_eq!(packetizer.push(&[0.5; 1_024], 10, 0), 1_024); - assert_eq!(packetizer.begin_frame(2, 1_024).unwrap(), 0); + assert_eq!( + packetizer.begin_frame(2, 1_024, None).unwrap(), + PcmPacketizerBegin::default() + ); assert_eq!(packetizer.push(&[0.25; 1_024], 20, 0), 896); assert_eq!(packetizer.complete_packet().unwrap().0.len(), 1_920); } @@ -161,8 +188,30 @@ mod tests { #[test] fn discontinuity_discards_only_the_bounded_partial_packet() { let mut packetizer = PcmPacketizer::new(48_000, 1).unwrap(); - packetizer.begin_frame(7, 480).unwrap(); + packetizer.begin_frame(7, 480, None).unwrap(); packetizer.push(&[0.0; 480], 10, 0); - assert_eq!(packetizer.begin_frame(9, 480).unwrap(), 480); + assert_eq!( + packetizer.begin_frame(9, 480, None).unwrap(), + PcmPacketizerBegin { + discontinuity_discarded_samples: 480, + cancelled_output_discarded_samples: 0, + } + ); + } + + #[test] + fn given_partial_packet_when_output_changes_then_old_samples_are_discarded() { + let mut packetizer = PcmPacketizer::new(48_000, 1).unwrap(); + packetizer.begin_frame(7, 480, Some(1)).unwrap(); + packetizer.push(&[0.0; 480], 10, 0); + + assert_eq!( + packetizer.begin_frame(8, 480, Some(2)).unwrap(), + PcmPacketizerBegin { + discontinuity_discarded_samples: 0, + cancelled_output_discarded_samples: 480, + } + ); + assert_eq!(packetizer.output_generation_id(), Some(2)); } } diff --git a/relay/src/rtc/publisher.rs b/relay/src/rtc/publisher.rs index ef17873..1ab9bb6 100644 --- a/relay/src/rtc/publisher.rs +++ b/relay/src/rtc/publisher.rs @@ -222,6 +222,19 @@ pub(crate) fn run_publish_loop( .publisher_stale_drops .fetch_add(stale_drops, Ordering::Relaxed); } + if frame + .output_generation + .as_ref() + .is_some_and(|generation| !generation.is_active()) + { + streams[index] + .stream + .counters + .cancelled_output_frames + .fetch_add(1, Ordering::Relaxed); + round_robin_start = (index + 1) % streams.len(); + continue 'publish; + } let mid = streams[index].stream.mid; let payload_type = streams[index].payload_type; let payload_bytes = frame.payload.len() as u64; @@ -313,6 +326,7 @@ mod tests { audio_level: None, capture_timestamp_ns: 0, capture_age_ns: 0, + output_generation: None, } } diff --git a/relay/src/runtime.rs b/relay/src/runtime.rs index 5a50c96..924ad1f 100644 --- a/relay/src/runtime.rs +++ b/relay/src/runtime.rs @@ -44,6 +44,8 @@ pub struct RelayPublishStatistics { pub encoder_channel_drops_total: u64, pub publisher_stale_drops_total: u64, pub ingress_queue_drops_total: u64, + pub cancelled_output_frames_total: u64, + pub cancelled_output_samples_total: u64, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -770,6 +772,8 @@ fn statistics( encoder_channel_drops_total: counters.encoded_channel_drops.load(Ordering::Relaxed), publisher_stale_drops_total: counters.publisher_stale_drops.load(Ordering::Relaxed), ingress_queue_drops_total: state.ingress_queue_drops_total.load(Ordering::Relaxed), + cancelled_output_frames_total: counters.cancelled_output_frames.load(Ordering::Relaxed), + cancelled_output_samples_total: counters.cancelled_output_samples.load(Ordering::Relaxed), } } diff --git a/relay/tests/fixtures/README.md b/relay/tests/fixtures/README.md new file mode 100644 index 0000000..c78078f --- /dev/null +++ b/relay/tests/fixtures/README.md @@ -0,0 +1,9 @@ +# Connector conformance data + +`connector-v1-vectors.json` is the packaged copy of +`protocol/conformance/connector/v1/vectors.json`. The source file remains the +protocol authority. Connector CI verifies this copy against the canonical +SHA-256 value before running the portable semantics test. + +When the protocol corpus changes, update the packaged file and its CI hash in +the same reviewed change. diff --git a/relay/tests/fixtures/connector-v1-vectors.json b/relay/tests/fixtures/connector-v1-vectors.json new file mode 100644 index 0000000..e9d5062 --- /dev/null +++ b/relay/tests/fixtures/connector-v1-vectors.json @@ -0,0 +1,163 @@ +{ + "schema_revision": 1, + "limits": { + "maximum_manifest_entries": 128, + "maximum_manifest_text_bytes": 4096, + "maximum_configuration_fields": 128, + "maximum_configuration_text_bytes": 16384, + "maximum_error_message_bytes": 4096 + }, + "cases": [ + { + "id": "valid_outbound_audio_endpoint", + "expected": "accept", + "manifest": { + "api_revision": 1, + "manifest_revision": 1, + "package_id": "dev.pocketstation.relay", + "package_version": "0.1.0", + "configuration": { + "revision": 1, + "fields": [ + { + "name": "source_token", + "kind": "secret", + "requirement": "required", + "documentation": "Relay source authentication token." + }, + { + "name": "startup_timeout_ms", + "kind": "duration_milliseconds", + "requirement": "defaulted", + "default": 10000, + "unsigned_range": [1, 60000], + "documentation": "Finite publisher startup deadline." + } + ] + }, + "components": [ + { + "component_id": "relay.publisher", + "kind": "endpoint", + "ports": [ + { + "name": "audio", + "direction": "input", + "signal_spec_id": "pocketstation.signal.pcm-audio.v1", + "multiplicity": "many", + "required": true + } + ] + } + ] + }, + "endpoint": { + "component_id": "relay.publisher", + "maximum_inflight_items": 256, + "maximum_payload_bytes": 1048576, + "startup_deadline_ms": 10000, + "shutdown_deadline_ms": 5000, + "probe_interval_ms": 25, + "success_threshold": 1, + "failure_threshold": 3 + } + }, + { + "id": "valid_composed_source_operator_endpoint_package", + "expected": "accept", + "manifest": { + "api_revision": 1, + "manifest_revision": 2, + "package_id": "dev.pocketstation.translation", + "package_version": "1.0.0", + "configuration": {"revision": 1, "fields": []}, + "components": [ + { + "component_id": "capture", + "kind": "source", + "ports": [{"name": "audio", "direction": "output", "signal_spec_id": "pocketstation.signal.pcm-audio.v1", "multiplicity": "one", "required": true}] + }, + { + "component_id": "transcribe", + "kind": "operator", + "ports": [ + {"name": "audio", "direction": "input", "signal_spec_id": "pocketstation.signal.pcm-audio.v1", "multiplicity": "one", "required": true}, + {"name": "transcript", "direction": "output", "signal_spec_id": "dev.pocketstation.signal.transcript.v1", "schema_ref": "proto:pocketstation.example.Transcript", "multiplicity": "many", "required": true} + ] + }, + { + "component_id": "publisher", + "kind": "endpoint", + "ports": [{"name": "transcript", "direction": "input", "signal_spec_id": "dev.pocketstation.signal.transcript.v1", "schema_ref": "proto:pocketstation.example.Transcript", "multiplicity": "many", "required": true}] + } + ] + } + }, + { + "id": "invalid_zero_manifest_revision", + "expected": "reject", + "error_code": "connector.manifest.invalid_revision", + "manifest": {"api_revision": 1, "manifest_revision": 0, "package_id": "dev.pocketstation.invalid", "package_version": "1.0.0", "configuration": {"revision": 1, "fields": []}, "components": []} + }, + { + "id": "invalid_duplicate_component_id", + "expected": "reject", + "error_code": "connector.manifest.duplicate_component", + "manifest": { + "api_revision": 1, + "manifest_revision": 1, + "package_id": "dev.pocketstation.invalid", + "package_version": "1.0.0", + "configuration": {"revision": 1, "fields": []}, + "components": [ + {"component_id": "publisher", "kind": "endpoint", "ports": []}, + {"component_id": "publisher", "kind": "endpoint", "ports": []} + ] + } + }, + { + "id": "invalid_secret_default", + "expected": "reject", + "error_code": "connector.configuration.secret_default_forbidden", + "manifest": { + "api_revision": 1, + "manifest_revision": 1, + "package_id": "dev.pocketstation.invalid", + "package_version": "1.0.0", + "configuration": { + "revision": 1, + "fields": [{"name": "token", "kind": "secret", "requirement": "defaulted", "default": "must-never-serialize", "documentation": "Invalid secret default."}] + }, + "components": [] + } + }, + { + "id": "invalid_unbounded_startup_deadline", + "expected": "reject", + "error_code": "connector.endpoint.invalid_deadline", + "manifest": { + "api_revision": 1, + "manifest_revision": 1, + "package_id": "dev.pocketstation.invalid", + "package_version": "1.0.0", + "configuration": {"revision": 1, "fields": []}, + "components": [{"component_id": "publisher", "kind": "endpoint", "ports": []}] + }, + "endpoint": {"component_id": "publisher", "maximum_inflight_items": 1, "maximum_payload_bytes": 1, "startup_deadline_ms": 0, "shutdown_deadline_ms": 1, "probe_interval_ms": 1, "success_threshold": 1, "failure_threshold": 1} + }, + { + "id": "valid_orthogonal_service_status", + "expected": "accept", + "service_status": { + "delivery_readiness": "not_ready", + "health": "degraded", + "recovery": "reconnecting", + "readiness_reason_code": "relay.ice.reconnecting", + "health_reason_code": "relay.network.degraded", + "recovery_reason_code": "relay.ice.reconnecting", + "revision": 4, + "last_transition_elapsed_ns": 125000000 + } + } + ] +} diff --git a/relay/tests/portable_semantics.rs b/relay/tests/portable_semantics.rs index d5ec87e..b196687 100644 --- a/relay/tests/portable_semantics.rs +++ b/relay/tests/portable_semantics.rs @@ -2,7 +2,7 @@ use std::collections::BTreeSet; use serde_json::Value; -const VECTORS: &str = include_str!("../../../protocol/conformance/connector/v1/vectors.json"); +const VECTORS: &str = include_str!("fixtures/connector-v1-vectors.json"); fn positive(value: Option<&Value>) -> bool { value.and_then(Value::as_u64).is_some_and(|value| value > 0)