diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..d274e6b --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,48 @@ +version: 2 +updates: + - package-ecosystem: cargo + directory: "/" + schedule: + interval: weekly + day: monday + time: "06:00" + timezone: "UTC" + open-pull-requests-limit: 5 + cooldown: + default-days: 7 + commit-message: + prefix: "chore" + include: scope + labels: + - dependencies + groups: + cargo: + applies-to: version-updates + patterns: + - "*" + cargo-security: + applies-to: security-updates + patterns: ["*"] + - package-ecosystem: github-actions + directory: "/" + schedule: + interval: weekly + day: monday + time: "06:00" + timezone: "UTC" + open-pull-requests-limit: 5 + cooldown: + default-days: 7 + commit-message: + prefix: "chore" + include: scope + labels: + - dependencies + groups: + github-actions: + applies-to: version-updates + patterns: + - "*" + github-actions-security: + applies-to: security-updates + patterns: ["*"] diff --git a/.github/rulesets/tags-release.json b/.github/rulesets/tags-release.json new file mode 100644 index 0000000..0639cd7 --- /dev/null +++ b/.github/rulesets/tags-release.json @@ -0,0 +1,20 @@ +{ + "name": "Protect release tags", + "target": "tag", + "enforcement": "active", + "bypass_actors": [], + "conditions": { + "ref_name": { + "include": ["refs/tags/[0-9]*.[0-9]*.[0-9]*"], + "exclude": [] + } + }, + "rules": [ + { + "type": "deletion" + }, + { + "type": "non_fast_forward" + } + ] +} diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8bea1b7..f630b77 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,65 +7,43 @@ on: # Two supported targets: Linux x86_64 (Ubuntu) and macOS arm64. # GitHub-hosted runners only — this is a public module. +# +# The build/test/lint jobs are the pinned `rust-ci` reusable from +# NDDev-OpenNetwork/ci-workflows; one pin per repository (see +# supply-chain.yml, codeql.yml, release.yml — same SHA). -env: - CARGO_TERM_COLOR: always +permissions: {} -jobs: - fmt: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 - - uses: dtolnay/rust-toolchain@6bed0761d98439e5a578e2877258200ad565ba87 # stable - with: - components: rustfmt - - run: cargo fmt --check - - check: - strategy: - matrix: - os: [ubuntu-latest, macos-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 - - uses: dtolnay/rust-toolchain@6bed0761d98439e5a578e2877258200ad565ba87 # stable - with: - components: clippy - - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2 - - run: cargo clippy --workspace --all-targets -- -D warnings - - name: clippy with owned transport backend (noq) and owned relay - run: >- - cargo clippy --workspace --all-targets - --features rds-net/transport-noq,rds-agent/transport-noq,rds-cli/transport-noq,rds-bench/transport-noq,rds-relay/owned-relay - -- -D warnings - - name: clippy with metrics export feature - run: >- - cargo clippy -p rds-net --all-targets --features metrics - -- -D warnings - - name: clippy with x11/desktop feature (linux) - if: matrix.os == 'ubuntu-latest' - run: cargo clippy --workspace --all-targets --features rds-desktop/x11 -- -D warnings - - name: clippy with desktop feature on agent/cli - run: >- - cargo clippy --workspace --all-targets - --features rds-agent/desktop,rds-cli/desktop -- -D warnings +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true - test: - strategy: - matrix: - os: [ubuntu-latest, macos-latest] - runs-on: ${{ matrix.os }} - steps: - - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 - - uses: dtolnay/rust-toolchain@6bed0761d98439e5a578e2877258200ad565ba87 # stable - - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2 - - run: cargo test --workspace - - name: test owned transport backend (noq) and owned relay - run: >- - cargo test -p rds-net -p rds-agent -p rds-relay - --features rds-net/transport-noq,rds-agent/transport-noq,rds-relay/owned-relay - - name: test metrics (known-traffic + prometheus render) - run: cargo test -p rds-net --features metrics --test metrics - - name: test with x11/desktop feature (linux) - if: matrix.os == 'ubuntu-latest' - run: cargo test --workspace --features rds-desktop/x11 +jobs: + rust: + name: rust + permissions: + contents: read + uses: NDDev-OpenNetwork/ci-workflows/.github/workflows/rust-ci.yml@a7b90bdf1ac12465cbd07072e3360442cfc8eec6 # 0.1.26 + with: + runner: ubuntu-latest + toolchain: stable + components: "clippy,rustfmt" + build_command: "cargo build --locked --workspace --all-targets" + # Tests run per-OS. The noq/owned-relay and metrics lanes run on both + # targets; the x11 lane is Linux-only (no X11 on macOS runners). + test_matrix_os: '["ubuntu-latest", "macos-latest"]' + test_command: >- + cargo test --locked --workspace + && cargo test --locked -p rds-net -p rds-agent -p rds-relay --features rds-net/transport-noq,rds-agent/transport-noq,rds-relay/owned-relay + && cargo test --locked -p rds-net --features metrics --test metrics + && { [ "$(uname -s)" != "Linux" ] || cargo test --locked --workspace --features rds-desktop/x11; } + fmt_command: "cargo fmt --all -- --check" + # Clippy runs once on Linux: default features, then every + # feature-gated lane (owned transport, metrics export, x11/desktop). + clippy_command: >- + cargo clippy --locked --workspace --all-targets -- -D warnings + && cargo clippy --locked --workspace --all-targets --features rds-net/transport-noq,rds-agent/transport-noq,rds-cli/transport-noq,rds-bench/transport-noq,rds-relay/owned-relay -- -D warnings + && cargo clippy --locked -p rds-net --all-targets --features metrics -- -D warnings + && cargo clippy --locked --workspace --all-targets --features rds-desktop/x11 -- -D warnings + && cargo clippy --locked --workspace --all-targets --features rds-agent/desktop,rds-cli/desktop -- -D warnings + timeout_minutes: 45 diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml new file mode 100644 index 0000000..b8b2634 --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,34 @@ +name: codeql + +# Code scanning via the pinned public reusable. Default setup stays +# `not-configured` on purpose: a pinned workflow is reviewable in a diff, +# and enabling default setup later means removing this file first. +# +# Rust needs no build step (CodeQL build-mode `none`, GA): extraction runs +# rust-analyzer over the workspace directly, so no build_command is passed. +# The weekly schedule keeps scanning a repository that stops changing. + +on: + push: + branches: [main] + pull_request: + schedule: + - cron: "41 5 * * 4" + +permissions: {} + +concurrency: + group: codeql-${{ github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + +jobs: + codeql: + name: codeql + permissions: + actions: read # the callee reads this run for SARIF upload bookkeeping + contents: read # check out the tree being analysed + security-events: write # publish CodeQL results to code scanning + uses: NDDev-OpenNetwork/ci-workflows/.github/workflows/public-codeql.yml@a7b90bdf1ac12465cbd07072e3360442cfc8eec6 # 0.1.26 + with: + languages: '["rust", "actions"]' + runner: ubuntu-latest diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..bbe9d13 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,97 @@ +name: release + +# Tag-driven immutable release: push a signed tag `X.Y.Z` where VERSION and +# the CHANGELOG `## [X.Y.Z]` heading match, and the pinned supply-chain +# reusable publishes one GitHub Release carrying a deterministic source +# archive, an SPDX SBOM of it, release notes, a manifest and SHA256SUMS — +# plus SLSA build-provenance and SBOM attestations (public repo: artifact +# attestations are included on every plan). +# +# Graph: resolve (read-only) → authorize (the `release` environment is the +# authority seam — add a required reviewer in Settings → Environments to +# make it a human gate) → publish (the only job holding write scopes). +# +# workflow_dispatch must run from the tag ref: the reusable fails closed +# when the checked-out HEAD is not the tagged commit. + +on: + push: + tags: + - "[0-9]+.[0-9]+.[0-9]+" + workflow_dispatch: + inputs: + version: + description: "Version to release. Run from the tag ref; must equal VERSION and the tag." + required: true + type: string + +permissions: {} + +concurrency: + group: release-${{ github.ref }} + cancel-in-progress: false + +jobs: + resolve: + name: Resolve release version + runs-on: ubuntu-latest + timeout-minutes: 5 + permissions: {} + outputs: + version: ${{ steps.v.outputs.version }} + steps: + - name: Resolve and shape-check the version + id: v + env: + INPUT_VERSION: ${{ inputs.version }} + EVENT_NAME: ${{ github.event_name }} + REF_NAME: ${{ github.ref_name }} + run: | + set -euo pipefail + version="$REF_NAME" + if [ "$EVENT_NAME" = "workflow_dispatch" ]; then + version="$INPUT_VERSION" + fi + if ! [[ "$version" =~ ^(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$ ]]; then + echo "release: version must be numeric SemVer X.Y.Z" >&2 + exit 1 + fi + echo "version=$version" >> "$GITHUB_OUTPUT" + + authorize: + name: Authorize release + needs: resolve + runs-on: ubuntu-latest + timeout-minutes: 5 + environment: release + permissions: {} + steps: + - name: Record the authorized candidate + env: + RELEASE_VERSION: ${{ needs.resolve.outputs.version }} + run: | + set -euo pipefail + { + echo "### Release authorized" + echo + echo "- version: \`${RELEASE_VERSION}\`" + echo "- approver: recorded by the \`release\` environment" + } >> "$GITHUB_STEP_SUMMARY" + + publish: + name: Build and publish release + needs: [resolve, authorize] + permissions: + contents: write # create the GitHub Release and upload assets + id-token: write # OIDC identity for SLSA build provenance + attestations: write # attest the SBOM and the release archive + artifact-metadata: write # actions/attest artifact storage record + uses: NDDev-OpenNetwork/ci-workflows/.github/workflows/release-supply-chain.yml@a7b90bdf1ac12465cbd07072e3360442cfc8eec6 # 0.1.26 + with: + runner: ubuntu-latest + version: ${{ needs.resolve.outputs.version }} + package_name: remote-device-sync + archive_paths: >- + README.md LICENSE VERSION CHANGELOG.md AGENTS.md deny.toml + rust-toolchain.toml Cargo.toml Cargo.lock crates deploy docs + scripts .github diff --git a/.github/workflows/supply-chain.yml b/.github/workflows/supply-chain.yml new file mode 100644 index 0000000..8972699 --- /dev/null +++ b/.github/workflows/supply-chain.yml @@ -0,0 +1,35 @@ +name: supply-chain + +on: + push: + branches: [main] + pull_request: + schedule: + # Weekly: RustSec advisories land continuously — catch one that + # appeared after the last commit. + - cron: "23 4 * * 3" + +# cargo-deny (deny.toml: bans/licenses/advisories/sources), cargo-audit +# (RustSec) and cargo-machete (unused deps) via the pinned reusable. + +permissions: {} + +concurrency: + group: supply-chain-${{ github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + +jobs: + supply-chain: + name: rust supply chain + permissions: + contents: read + uses: NDDev-OpenNetwork/ci-workflows/.github/workflows/rust-supply-chain.yml@a7b90bdf1ac12465cbd07072e3360442cfc8eec6 # 0.1.26 + with: + runner: ubuntu-latest + toolchain: stable + enable_deny: true + enable_audit: true + enable_machete: true + working_directory: "." + deny_arguments: "--all-features --config deny.toml" + deny_command: "check bans licenses advisories sources" diff --git a/CHANGELOG.md b/CHANGELOG.md index f968d47..a2a6bbc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,91 @@ ## [Unreleased] +- Stability/latency hardening across the workspace: + - `rds-net`: the uni demux hands each accepted stream its own + tag-read task under a 10s bound — a peer that opens a stream and + never writes its `UniHello` can no longer stall routing of every + stream behind it, and a failed send into a reclaimed inbox no + longer removes the *new* route (channel-identity checked). + - `rds-desktop`: the decoder is session-local (a shared static one + cross-contaminated reference chains between sessions); decode + failure auto-requests an IDR, rate-limited at 500ms so corrupt + stretches can't storm; session ack and frame stream header/body + reads are bounded at 30s, frame bodies at 32 MiB. On the serving + side a frame that goes stale *while sending* is reset mid-write + (MoQ-style) — a stale delta's tail no longer consumes path + capacity; the reset re-arms the producer's IDR flag and drains the + undecodable deltas. Frame streams carry an explicit constant + priority below control (escalating per-frame priorities that + starved in-flight sends are not reintroduced). Empty encode-failure + placeholders are skipped instead of sent. + - `openh264` codec: the `keyframe` flag is read off the emitted NAL + units instead of assumed from `seq % 240` (which was wrong — + `intra_frame_period` defaulted to `auto`, so periodic IDRs never + fired on schedule); the period is now pinned at 240 (~8s at 30fps) + as a bound on undecodable time. `set_bitrate` actually rebuilds the + encoder past a 15% deadband instead of being a silent no-op, and + the rebuild's first frame is a real IDR. + - `rds-sync`: manifests stream through `StreamCDC` (memory bounded at + one max-size chunk, proven identical to slice chunking); manifest + scans, journal open/verify, and assembly run on `spawn_blocking`; + verified chunks are written by a dedicated blocking-pool sink + behind a bounded queue; every protocol read and chunk body is + bounded by a 300s stall; completion counts chunks on the wire — + not the sink's lagging `present` counter, which used to park the + receive loop in a 300s stall after the last chunk. + - `rds-agent`: stream hello bounded at 15s; state/grant/watcher + mutexes recover from poisoning instead of denying service forever; + request paths refuse with `HelloAck::Error` instead of + `unreachable!`. + - `rds-discovery`: `/v1/registry` PUT verifies freshness and stores + under one write lock — two racing valid PUTs can no longer leave + the older snapshot stored (regression test + `concurrent_registry_puts_cannot_regress`). + - `rds-relay`: register/control-stream wait bounded at 15s — a + connection that never registers no longer parks a task. + - `rds-cli`: connect bounded at 30s; every `HelloAck` wait and the + ping echo bounded at 15s. +- Protocol v3 + review hardening: every uni-directional stream now + opens with a `UniHello` tag (`Desktop`/`Sync`/`Audio`), and the + accepting side routes it through a single per-connection demux + (`Connection::uni_streams`) — desktop and sync can share a + connection without racing `accept_uni`. Inboxes end cleanly: + `UniStreams::recv` returns `None` once the connection dies — the + demux's exit drops every registered sender instead of leaving + consumers parked. `PROTOCOL_VERSION` is 3; v2 peers won't interop + on uni streams. +- Security fixes: sync confinement is now resolved, not just lexical — + `resolve_under` canonicalizes the deepest existing ancestor of every + destination and requires it to stay under the canonical sync root, + so symlinked components can't redirect pulls, the `.rds-sync` + journal, or assembly outside the root; assembly's temp file is a + suffixed `*.rds-part` created exclusively after unlinking any stale + one, and `.rds-sync` as a first path component is refused outright. + `ChunkHdr.len` is checked against the manifest before it sizes the + receive buffer (a forged u32 no longer forces a huge allocation). + X11 scroll injection clamps deltas to 32 clicks per event. Input + events naming a display other than the session's are dropped + un-acked. The directory's `/v1/revocations` PUT now verifies and + stores under one write lock (no check-then-store regression window). +- Correctness fix: `mailbox::Sender` decrements an explicit sender + count and then notifies on drop — a consumer parked in `recv` now + observes the last sender leaving instead of sleeping forever, and a + cross-thread wake can't observe a stale count and re-park + (regression tests `parked_recv_wakes_when_last_sender_drops`, + `parked_recv_survives_drop_wake_race`). +- CI/CD: the hand-rolled `ci.yml` is replaced by pinned + `ci-workflows` reusables (0.1.26): `rust-ci` (locked build, fmt, + five clippy lanes, ubuntu+macos test matrix), `rust-supply-chain` + (cargo-deny per `deny.toml`, cargo-audit, cargo-machete — weekly + advisory sweep), `public-codeql` (rust + actions, build-mode none) + and `release-supply-chain` (tag `X.Y.Z` → immutable release: source + archive + SPDX SBOM + SHA256SUMS + SLSA/SBOM attestations, behind a + `release` environment). `deny.toml` now allows `Unlicense` and + `CDLA-Permissive-2.0` (iroh transitive deps) and ignores the two + unfixable unmaintained advisories; 13 unused crate dependencies + removed; `VERSION` file added for the release contract; dependabot + tracks cargo + github-actions weekly. - WS8 deployment: `deploy/systemd/rds-server.service` and `rds-agent.service` — hardened units (ProtectSystem=strict, NoNewPrivileges, PrivateTmp/Devices, ProtectKernel*/ControlGroups, @@ -50,8 +135,8 @@ atomic rename after root verification. `rel_path` rejects traversal, absolute paths, NUL and oversize. `rds send`/`rds recv` push/pull through `rds_cli::open_sync`; the agent serves `Sync` - under `--sync-dir` with a one-session-per-connection guard (chunk - streams share the `accept_uni` queue) and advertises `Sync` in + under `--sync-dir` with a one-session-per-connection guard and + advertises `Sync` in `Info` only when configured. E2E: byte-identical push/pull, zero-chunk resend, corrupt-part refetch, torn-journal and mid-transfer kill resume, repeated kill/resume convergence, diff --git a/Cargo.lock b/Cargo.lock index fdda94a..a75863a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2835,7 +2835,6 @@ dependencies = [ "rds-discovery", "rds-net", "rds-sync", - "serde", "tokio", "tracing", "tracing-subscriber", @@ -2859,13 +2858,11 @@ dependencies = [ "noq", "rds-agent", "rds-cli", - "rds-core", "rds-discovery", "rds-net", "serde", "serde_json", "tokio", - "tracing", "tracing-subscriber", ] @@ -2874,7 +2871,6 @@ name = "rds-cli" version = "0.1.0" dependencies = [ "anyhow", - "bytes", "clap", "rds-core", "rds-desktop", @@ -2905,14 +2901,13 @@ dependencies = [ name = "rds-desktop" version = "0.1.0" dependencies = [ - "async-trait", "bytes", "noq", "openh264", "rds-bench", "rds-core", "rds-net", - "serde", + "rds-sync", "thiserror 2.0.20", "tokio", "tracing", @@ -2932,7 +2927,6 @@ dependencies = [ "serde_json", "thiserror 2.0.20", "tokio", - "tracing", ] [[package]] @@ -2953,11 +2947,9 @@ dependencies = [ "rds-core", "rds-discovery", "serde", - "thiserror 2.0.20", "tokio", "tokio-stream", "tracing", - "tracing-subscriber", "turmoil", ] @@ -2966,15 +2958,12 @@ name = "rds-relay" version = "0.1.0" dependencies = [ "anyhow", - "bytes", "clap", "iroh", "iroh-relay", "postcard", - "rand 0.9.5", "rds-core", "rds-net", - "serde", "tokio", "tracing", "tracing-subscriber", @@ -3006,7 +2995,6 @@ dependencies = [ "fastcdc", "noq", "postcard", - "proptest", "rds-bench", "rds-core", "rds-net", diff --git a/VERSION b/VERSION new file mode 100644 index 0000000..6e8bf73 --- /dev/null +++ b/VERSION @@ -0,0 +1 @@ +0.1.0 diff --git a/crates/rds-agent/Cargo.toml b/crates/rds-agent/Cargo.toml index 0474f94..e667877 100644 --- a/crates/rds-agent/Cargo.toml +++ b/crates/rds-agent/Cargo.toml @@ -23,7 +23,6 @@ rds-desktop = { workspace = true, optional = true } rds-net.workspace = true rds-discovery.workspace = true rds-sync.workspace = true -serde.workspace = true tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true diff --git a/crates/rds-agent/src/lib.rs b/crates/rds-agent/src/lib.rs index 751acdf..5ac6198 100644 --- a/crates/rds-agent/src/lib.rs +++ b/crates/rds-agent/src/lib.rs @@ -45,6 +45,19 @@ fn next_session_id() -> u64 { SESSION_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed) } +/// A peer that opens a stream but never writes its `StreamHello` would +/// otherwise park a task per stream for the connection's lifetime — +/// bounded here so silent streams cost seconds, not the session. +const HELLO_TIMEOUT: Duration = Duration::from_secs(15); + +/// Mutex acquisition that survives a poisoned lock: every mutex here +/// guards plain data (a state word, an `Option`, a `HashSet`) whose +/// invariants a panic cannot corrupt, so refusing service forever +/// after one panicked holder would be the worse failure. +fn lock(m: &Mutex) -> std::sync::MutexGuard<'_, T> { + m.lock().unwrap_or_else(|e| e.into_inner()) +} + /// Runtime policy for the agent. #[derive(Clone)] pub struct AgentPolicy { @@ -258,8 +271,9 @@ struct ConnAuthz { /// The active grant id, released back into `active_grants` on /// connection teardown so the slot frees for a future session. grant_id: Mutex>, - /// One sync session per connection: chunk streams arrive on - /// `accept_uni`, which concurrent sessions would race on. + /// One sync session per connection: the journal is per-destination + /// and the uni demux serves a single `Sync` claim at a time, so a + /// concurrent session gets a clean refusal instead of a race. sync_busy: std::sync::atomic::AtomicBool, } @@ -292,7 +306,7 @@ impl ConnAuthz { /// Scope the connection currently has for service streams. fn scope(&self) -> Result>, &'static str> { - match &*self.state.lock().unwrap() { + match &*lock(&self.state) { AuthzState::Open => Ok(None), AuthzState::Pending => Err("grant required: send Authz first"), AuthzState::Granted(g) => Ok(Some(g.clone())), @@ -344,11 +358,11 @@ async fn serve_connection( /// Connection teardown: stop the expiry/revocation watcher and release /// the grant slot so the same grant may authorize a future session. fn teardown(_conn: &Connection, policy: &AgentPolicy, authz: &ConnAuthz) { - if let Some(w) = authz.watcher.lock().unwrap().take() { + if let Some(w) = lock(&authz.watcher).take() { w.abort(); } - if let Some(id) = authz.grant_id.lock().unwrap().take() { - policy.active_grants.lock().unwrap().remove(&id); + if let Some(id) = lock(&authz.grant_id).take() { + lock(&policy.active_grants).remove(&id); } } @@ -365,7 +379,11 @@ async fn serve_stream( authz: Arc, desktop: bool, ) -> anyhow::Result<()> { - let hello: StreamHello = read_frame(&mut recv).await?; + let hello: StreamHello = match tokio::time::timeout(HELLO_TIMEOUT, read_frame(&mut recv)).await + { + Ok(h) => h?, + Err(_) => anyhow::bail!("stream hello timed out"), + }; if let StreamHello::Authz(grant) = hello { return authorize(&conn, send, recv, grant, &policy, &authz).await; } @@ -473,7 +491,19 @@ async fn serve_stream( } } #[cfg(not(feature = "desktop"))] - unreachable!() + { + // `desktop` is `cfg!(feature = "desktop")` so this + // is unreachable today — but a request path must + // refuse, never panic, if that coupling breaks. + write_frame( + &mut send, + &HelloAck::Error { + message: "desktop service not compiled in".into(), + }, + ) + .await?; + anyhow::bail!("desktop requested but not compiled in"); + } } else { let _ = hello; write_frame( @@ -523,7 +553,19 @@ async fn serve_stream( .await?; anyhow::bail!("audio service not implemented"); } - StreamHello::Authz(_) => unreachable!("Authz handled above"), + StreamHello::Authz(_) => { + // `authorize` early-returns on Authz, so this is + // unreachable — a request path still refuses rather + // than panic if that ever stops holding. + write_frame( + &mut send, + &HelloAck::Error { + message: "authz is not a service".into(), + }, + ) + .await?; + anyhow::bail!("authz stream on the service path"); + } } Ok(()) } @@ -557,7 +599,7 @@ async fn authorize( { // A second Authz stream is never valid — either still pending // (fine, this is the first) or already granted (refuse). - if matches!(*authz.state.lock().unwrap(), AuthzState::Granted(_)) { + if matches!(*lock(&authz.state), AuthzState::Granted(_)) { write_frame( &mut send, &HelloAck::Error { @@ -602,11 +644,7 @@ async fn authorize( deny(conn, "grant revoked"); anyhow::bail!("grant revoked"); } - let admitted = policy - .active_grants - .lock() - .map(|mut a| a.insert(verified.id)) - .unwrap_or(false); + let admitted = lock(&policy.active_grants).insert(verified.id); if !admitted { write_frame( &mut send, @@ -619,12 +657,12 @@ async fn authorize( anyhow::bail!("grant replay on concurrent connection"); } let grant = Arc::new(verified); - *authz.state.lock().unwrap() = AuthzState::Granted(grant.clone()); - *authz.grant_id.lock().unwrap() = Some(grant.id); + *lock(&authz.state) = AuthzState::Granted(grant.clone()); + *lock(&authz.grant_id) = Some(grant.id); write_frame(&mut send, &HelloAck::Ok).await?; send.finish()?; info!(%peer, grant = %blake3::Hash::from(grant.id), "grant authorized"); - *authz.watcher.lock().unwrap() = Some(tokio::spawn(watch_grant( + *lock(&authz.watcher) = Some(tokio::spawn(watch_grant( conn.clone(), grant, policy.denylist.subscribe(), diff --git a/crates/rds-bench/Cargo.toml b/crates/rds-bench/Cargo.toml index eea1bed..ba5ed2a 100644 --- a/crates/rds-bench/Cargo.toml +++ b/crates/rds-bench/Cargo.toml @@ -19,13 +19,11 @@ iroh-relay = { workspace = true, features = ["server"] } noq = { workspace = true, optional = true } rds-agent.workspace = true rds-cli.workspace = true -rds-core.workspace = true rds-discovery.workspace = true rds-net.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true -tracing.workspace = true tracing-subscriber.workspace = true [lints] diff --git a/crates/rds-cli/Cargo.toml b/crates/rds-cli/Cargo.toml index 599b50e..ef0af28 100644 --- a/crates/rds-cli/Cargo.toml +++ b/crates/rds-cli/Cargo.toml @@ -14,7 +14,6 @@ transport-noq = ["rds-net/transport-noq"] [dependencies] anyhow.workspace = true -bytes.workspace = true clap.workspace = true rds-core.workspace = true rds-desktop = { workspace = true, optional = true } diff --git a/crates/rds-cli/src/lib.rs b/crates/rds-cli/src/lib.rs index 863ca3d..5ad56ac 100644 --- a/crates/rds-cli/src/lib.rs +++ b/crates/rds-cli/src/lib.rs @@ -2,18 +2,35 @@ use std::net::SocketAddr; use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; use anyhow::Context; use rds_core::{AgentInfo, HelloAck, StreamHello, read_frame, write_frame}; use rds_net::{Connection, Endpoint, EndpointAddr}; use tokio::net::TcpListener; +/// Every acknowledgement wait is bounded — a peer that opens the +/// stream but never answers must not hang the CLI forever. Generous: +/// grant verification and the far side's TCP connect gate on it. +const ACK_TIMEOUT: Duration = Duration::from_secs(15); + +/// One `HelloAck` read, bounded by [`ACK_TIMEOUT`]. +async fn read_ack(recv: &mut rds_net::RecvStream) -> anyhow::Result { + match tokio::time::timeout(ACK_TIMEOUT, read_frame::<_, HelloAck>(recv)).await { + Ok(r) => r.map_err(Into::into), + Err(_) => anyhow::bail!("peer did not answer within {ACK_TIMEOUT:?}"), + } +} + +/// Bound on the whole dial — hole punching and relay fallback retry +/// internally, so the CLI gives them room but not forever. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(30); + /// Open a connection to `target` and return it. pub async fn connect(endpoint: &Endpoint, target: EndpointAddr) -> anyhow::Result { - endpoint - .connect(target, rds_core::ALPN) + tokio::time::timeout(CONNECT_TIMEOUT, endpoint.connect(target, rds_core::ALPN)) .await + .context("connect timed out")? .context("connect to peer") } @@ -29,7 +46,7 @@ pub async fn connect_authorized( let conn = connect(endpoint, target).await?; let (mut send, mut recv) = conn.open_bi().await?; write_frame(&mut send, &StreamHello::Authz(grant.clone())).await?; - match read_frame::<_, HelloAck>(&mut recv).await? { + match read_ack(&mut recv).await? { HelloAck::Ok => Ok(conn), HelloAck::Error { message } => anyhow::bail!("grant rejected: {message}"), other => anyhow::bail!("unexpected ack {other:?}"), @@ -37,17 +54,20 @@ pub async fn connect_authorized( } /// Send a `Ping` and measure the full round trip. -pub async fn ping(conn: &Connection, nonce: u64) -> anyhow::Result { +pub async fn ping(conn: &Connection, nonce: u64) -> anyhow::Result { let start = Instant::now(); let (mut send, mut recv) = conn.open_bi().await?; write_frame(&mut send, &StreamHello::Ping { nonce }).await?; - match read_frame::<_, HelloAck>(&mut recv).await? { + match read_ack(&mut recv).await? { HelloAck::Ok => {} HelloAck::Error { message } => anyhow::bail!("ping rejected: {message}"), other => anyhow::bail!("unexpected ack {other:?}"), } let mut buf = [0u8; 8]; - recv.read_exact(&mut buf).await?; + match tokio::time::timeout(ACK_TIMEOUT, recv.read_exact(&mut buf)).await { + Ok(r) => r?, + Err(_) => anyhow::bail!("ping echo timed out"), + } let echoed = u64::from_be_bytes(buf); if echoed != nonce { anyhow::bail!("ping echo mismatch: {echoed} != {nonce}"); @@ -59,7 +79,7 @@ pub async fn ping(conn: &Connection, nonce: u64) -> anyhow::Result anyhow::Result { let (mut send, mut recv) = conn.open_bi().await?; write_frame(&mut send, &StreamHello::Info).await?; - match read_frame::<_, HelloAck>(&mut recv).await? { + match read_ack(&mut recv).await? { HelloAck::Info(info) => Ok(info), HelloAck::Error { message } => anyhow::bail!("info rejected: {message}"), other => anyhow::bail!("unexpected ack {other:?}"), @@ -81,7 +101,7 @@ pub async fn open_tcp( }, ) .await?; - match read_frame::<_, HelloAck>(&mut recv).await? { + match read_ack(&mut recv).await? { HelloAck::Ok => Ok((send, recv)), HelloAck::Error { message } => anyhow::bail!("forward rejected: {message}"), other => anyhow::bail!("unexpected ack {other:?}"), @@ -95,7 +115,7 @@ pub async fn open_sync( ) -> anyhow::Result<(rds_net::SendStream, rds_net::RecvStream)> { let (mut send, mut recv) = conn.open_bi().await?; write_frame(&mut send, &StreamHello::Sync).await?; - match read_frame::<_, HelloAck>(&mut recv).await? { + match read_ack(&mut recv).await? { HelloAck::Ok => Ok((send, recv)), HelloAck::Error { message } => anyhow::bail!("sync rejected: {message}"), other => anyhow::bail!("unexpected ack {other:?}"), diff --git a/crates/rds-core/src/lib.rs b/crates/rds-core/src/lib.rs index 73e7383..d69e391 100644 --- a/crates/rds-core/src/lib.rs +++ b/crates/rds-core/src/lib.rs @@ -21,11 +21,32 @@ pub const ALPN: &[u8] = b"rds/0"; /// Wire protocol version. Peers refuse mismatched majors. /// v2: FrameHeader carries capture/encode/send timestamps, InputEvent /// carries metadata, control stream gains heartbeat + input acks. -pub const PROTOCOL_VERSION: u16 = 2; +/// v3: every uni-directional stream opens with a [`UniHello`] tag so a +/// single per-connection demux can route it — v2 consumers each called +/// `accept_uni` directly and could steal each other's streams. +pub const PROTOCOL_VERSION: u16 = 3; /// Upper bound for a serialized greeting, guard against abusive peers. pub const MAX_MESSAGE_LEN: u32 = 64 * 1024; +/// First frame on every uni-directional stream (v3): routes the stream +/// to the service that owns it. The accepting side runs one +/// `accept_uni` demux per connection and hands each stream to the +/// consumer registered for its tag — two services on one connection can +/// no longer consume each other's streams. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub enum UniHello { + /// Desktop video frame stream: a [`FrameHeader`] then the encoded + /// payload follow. + Desktop, + /// Sync chunk stream: `SyncMsg` frames (`ChunkSet`, `ChunkHdr` + + /// bytes, `SetDone`) follow. + Sync, + /// Audio packet stream: [`AudioFrame`] records follow (codec + /// support lands in v0.3). + Audio, +} + /// First frame on every bi-directional stream. #[derive(Debug, Clone, Serialize, Deserialize)] pub enum StreamHello { diff --git a/crates/rds-desktop/Cargo.toml b/crates/rds-desktop/Cargo.toml index a4cc2dc..3ceab79 100644 --- a/crates/rds-desktop/Cargo.toml +++ b/crates/rds-desktop/Cargo.toml @@ -11,12 +11,10 @@ default = [] x11 = ["dep:x11rb", "dep:openh264"] [dependencies] -async-trait.workspace = true bytes.workspace = true openh264 = { version = "0.9", optional = true } rds-core.workspace = true rds-net.workspace = true -serde.workspace = true thiserror.workspace = true tokio.workspace = true tracing.workspace = true @@ -25,6 +23,7 @@ x11rb = { version = "0.14", features = ["shm", "xtest", "xfixes"], optional = tr [dev-dependencies] noq.workspace = true rds-bench = { workspace = true, features = ["transport-noq"] } +rds-sync.workspace = true [lints] workspace = true diff --git a/crates/rds-desktop/src/client.rs b/crates/rds-desktop/src/client.rs index 905c734..89a093b 100644 --- a/crates/rds-desktop/src/client.rs +++ b/crates/rds-desktop/src/client.rs @@ -18,6 +18,109 @@ use tokio::sync::mpsc; use crate::{DesktopError, RawFrame, SessionClock, mailbox}; +/// Largest frame payload accepted off the wire. A real H.264 frame is +/// orders of magnitude smaller; the bound exists so a hostile or +/// broken peer cannot grow the receive buffer without limit. +const MAX_FRAME_BYTES: usize = 32 * 1024 * 1024; + +/// Minimum interval between decode-failure IDR requests — a corrupt +/// stretch must not turn into an IDR storm. +const IDR_MIN_INTERVAL_MS: u64 = 500; + +/// Bound on waiting for a frame stream's header or body, and on the +/// session handshake ack — a peer that opens a tagged stream and +/// stalls mid-send would otherwise park a task per stream. +const FRAME_STREAM_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + +/// Decode state owned by one session. Codec reference frames are chain +/// state: a decoder shared across sessions would cross-contaminate +/// streams, so it lives here and is dropped with the session. +struct Delivery { + #[cfg(feature = "x11")] + decoder: Option, + /// Session-clock ms of the last decode-failure IDR request. + last_idr_req_ms: u64, +} + +/// What `Delivery::decode` made of one payload. +enum DecodeOutcome { + /// A frame came out of the decoder (x11 build only — headless + /// builds have no decoder). + #[cfg(feature = "x11")] + Decoded(RawFrame), + /// Decoder buffered the input without producing a frame — happens + /// on the first frames of a chain; not an error. + Buffered, + /// Payload could not be decoded — the reference chain is broken + /// until the next keyframe. + Failed, +} + +impl Delivery { + fn new() -> Self { + Self { + #[cfg(feature = "x11")] + decoder: None, + last_idr_req_ms: 0, + } + } + + fn decode(&mut self, header: &FrameHeader, body: &[u8]) -> DecodeOutcome { + #[cfg(feature = "x11")] + { + use crate::Decoder; + if body.is_empty() { + return DecodeOutcome::Failed; + } + if self.decoder.is_none() { + match crate::H264Decoder::new() { + Ok(d) => self.decoder = Some(d), + Err(e) => { + tracing::warn!("decoder init failed: {e}"); + return DecodeOutcome::Failed; + } + } + } + let encoded = crate::EncodedFrame { + codec: header.codec, + data: bytes::Bytes::copy_from_slice(body), + keyframe: header.keyframe, + }; + match self.decoder.as_mut().unwrap().decode(&encoded) { + Ok(Some(raw)) => { + tracing::debug!( + "frame seq={} {}x{} decoded", + header.seq, + raw.width, + raw.height + ); + DecodeOutcome::Decoded(raw) + } + Ok(None) => { + tracing::debug!("frame seq={} buffered", header.seq); + DecodeOutcome::Buffered + } + Err(e) => { + tracing::debug!("decode seq={} failed: {e}", header.seq); + DecodeOutcome::Failed + } + } + } + #[cfg(not(feature = "x11"))] + { + let _ = header; + // No decoder in this build: a non-empty payload is simply + // consumed; an empty one still marks a broken delivery and + // deserves the IDR resync below. + if body.is_empty() { + DecodeOutcome::Failed + } else { + DecodeOutcome::Buffered + } + } + } +} + /// A live desktop session on the client side. pub struct DesktopSession { /// Decoded frames in arrival order; stale ones never arrive here. @@ -98,7 +201,15 @@ impl DesktopSession { let clock = opts.clock.unwrap_or_default(); let (mut send, mut recv) = conn.open_bi().await?; write_frame(&mut send, &StreamHello::Desktop(hello)).await?; - let caps = match read_frame::<_, HelloAck>(&mut recv).await? { + let caps = + match tokio::time::timeout(FRAME_STREAM_TIMEOUT, read_frame::<_, HelloAck>(&mut recv)) + .await + { + Ok(Ok(ack)) => ack, + Ok(Err(e)) => return Err(e.into()), + Err(_) => return Err(DesktopError::Capture("session ack timed out".into())), + }; + let caps = match caps { HelloAck::Desktop(caps) => caps, HelloAck::Ok => DesktopCaps { displays: vec![], @@ -144,37 +255,54 @@ impl DesktopSession { } }); - // Frame receiver task: accept uni streams, drop stale, decode - // newest. A delivered-seq gap means a delta chain broke — the - // session auto-requests an IDR so decode can resync. - let conn = conn.clone(); + // Frame receiver task: desktop-tagged uni streams from the + // connection demux, drop stale, decode newest. A delivered-seq + // gap means a delta chain broke — the session auto-requests an + // IDR so decode can resync. + let mut uni = conn + .uni_streams(rds_core::UniHello::Desktop) + .map_err(|e| DesktopError::Io(std::io::Error::other(e.to_string())))?; // `next_seq` is the lowest seq still acceptable — the next // expected frame. Init 0 accepts the stream's first frame // (seq 0 is fresh, not stale) and lets the gap check catch a // first arrival above it. `latest_seq()` derives from it. let seq_marker = next_seq.clone(); let gap_ctrl = ctrl_tx.clone(); + // Per-session decode state: the decoder's reference chain is + // session state, never global. + let delivery = Arc::new(std::sync::Mutex::new(Delivery::new())); + let deliver_clock = clock.clone(); // Serializes claim+delivery so the order frames reach the // consumer is strictly the order of their seq numbers. let deliver_lock = Arc::new(tokio::sync::Mutex::new(())); let frame_task = tokio::spawn(async move { - while let Ok(mut stream) = conn.accept_uni().await { + while let Some(mut stream) = uni.recv().await { let frame_tx = frame_tx.clone(); let header_tx = header_tx.clone(); let seq_marker = seq_marker.clone(); let gap_ctrl = gap_ctrl.clone(); let deliver_lock = deliver_lock.clone(); + let delivery = delivery.clone(); + let clock = deliver_clock.clone(); tokio::spawn(async move { - let header: FrameHeader = match read_frame(&mut stream).await { - Ok(h) => h, - Err(_) => return, - }; + let header: FrameHeader = + match tokio::time::timeout(FRAME_STREAM_TIMEOUT, read_frame(&mut stream)) + .await + { + Ok(Ok(h)) => h, + _ => return, + }; if header.seq < seq_marker.load(Ordering::Relaxed) { return; } - let body = match stream.read_to_end(64 * 1024 * 1024).await { - Ok(b) => b, - Err(_) => return, + let body = match tokio::time::timeout( + FRAME_STREAM_TIMEOUT, + stream.read_to_end(MAX_FRAME_BYTES), + ) + .await + { + Ok(Ok(b)) => b, + _ => return, }; // Delivered order must match seq order: claim the // watermark and publish under one lock so a slower @@ -191,7 +319,14 @@ impl DesktopSession { seq_marker.store(header.seq + 1, Ordering::Relaxed); // Wire-level tap: every complete, non-stale frame. header_tx.send(header.clone()); - deliver(header, body, frame_tx.clone()).await; + deliver( + &header, + &body, + &frame_tx, + &delivery, + &gap_ctrl, + clock.now_ms(), + ); }); } }); @@ -279,50 +414,39 @@ impl DesktopSession { } } -async fn deliver(header: FrameHeader, body: Vec, tx: mailbox::Sender) { - #[cfg(feature = "x11")] - { - use crate::Decoder; - use std::sync::Mutex; - static DECODER: Mutex> = Mutex::new(None); - let encoded = crate::EncodedFrame { - codec: header.codec, - data: bytes::Bytes::from(body), - keyframe: header.keyframe, - }; - let mut guard = match DECODER.lock() { - Ok(g) => g, - Err(_) => return, - }; - if guard.is_none() { - *guard = match crate::H264Decoder::new() { - Ok(d) => Some(d), - Err(e) => { - tracing::warn!("decoder init failed: {e}"); - return; - } - }; +/// Decode one complete frame under the session's serialized publish +/// lock and forward the result. A decode failure means the reference +/// chain is broken — request an IDR so the encoder resyncs instead of +/// decoding deltas against a corrupt reference until the next periodic +/// keyframe. The request is rate-limited so a corrupt stretch cannot +/// turn into an IDR storm. +fn deliver( + header: &FrameHeader, + body: &[u8], + #[allow(unused_variables)] tx: &mailbox::Sender, + delivery: &std::sync::Mutex, + ctrl: &mpsc::Sender, + now_ms: u64, +) { + let mut delivery = match delivery.lock() { + Ok(d) => d, + Err(_) => return, + }; + match delivery.decode(header, body) { + // Newest-frame-wins: a full queue evicts the oldest frame + // rather than dropping the fresh one. + #[cfg(feature = "x11")] + DecodeOutcome::Decoded(raw) => { + tx.send(raw); } - match guard.as_mut().unwrap().decode(&encoded) { - Ok(Some(raw)) => { - tracing::debug!( - "frame seq={} {}x{} decoded", - header.seq, - raw.width, - raw.height - ); - // Newest-frame-wins: a full queue evicts the oldest - // frame rather than dropping the fresh one. - tx.send(raw); + DecodeOutcome::Buffered => {} + DecodeOutcome::Failed => { + if now_ms.saturating_sub(delivery.last_idr_req_ms) >= IDR_MIN_INTERVAL_MS { + delivery.last_idr_req_ms = now_ms; + let _ = ctrl.try_send(DesktopControl::RequestIdr); } - Ok(None) => tracing::debug!("frame seq={} buffered", header.seq), - Err(e) => tracing::debug!("decode seq={} failed: {e}", header.seq), } } - #[cfg(not(feature = "x11"))] - { - let _ = (header, body, tx); - } } /// Headless desktop run: connects, prints capabilities, streams decode diff --git a/crates/rds-desktop/src/codec/openh264.rs b/crates/rds-desktop/src/codec/openh264.rs index 3aeaafa..d32cc29 100644 --- a/crates/rds-desktop/src/codec/openh264.rs +++ b/crates/rds-desktop/src/codec/openh264.rs @@ -6,7 +6,8 @@ use bytes::Bytes; use openh264::decoder::{Decoder as OhDecoder, DecoderConfig}; use openh264::encoder::{ - BitRate, Encoder as OhEncoder, EncoderConfig, FrameRate, RateControlMode, UsageType, + BitRate, EncodedBitStream, Encoder as OhEncoder, EncoderConfig, FrameRate, IntraFramePeriod, + RateControlMode, UsageType, }; use openh264::formats::{YUVBuffer, YUVSource}; use openh264::{Error as OhError, OpenH264API}; @@ -17,33 +18,62 @@ use crate::{Decoder, DesktopError, EncodedFrame, Encoder, RawFrame}; /// Real-time OpenH264 encoder feeding `EncodedFrame`s. pub struct H264Encoder { inner: OhEncoder, + /// Rate the live encoder is configured for. bitrate: u64, + /// Requested rate waiting to be applied — `openh264` has no live + /// bitrate setter, so a change rebuilds the encoder lazily at the + /// next `encode` (the rebuilt encoder's first frame is an IDR with + /// fresh SPS/PPS, which is exactly the resync a rate shift wants). + pending_bitrate: Option, + fps: f32, want_idr: bool, - seq: u64, } impl H264Encoder { pub fn new(bitrate_bps: u64, fps: f32) -> Result { + let inner = Self::build(bitrate_bps as u32, fps)?; + Ok(Self { + inner, + bitrate: bitrate_bps, + pending_bitrate: None, + fps, + want_idr: true, + }) + } + + fn build(bitrate_bps: u32, fps: f32) -> Result { let config = EncoderConfig::new() .usage_type(UsageType::CameraVideoRealTime) .rate_control_mode(RateControlMode::Bitrate) - .bitrate(BitRate::from_bps(bitrate_bps as u32)) + .bitrate(BitRate::from_bps(bitrate_bps)) .max_frame_rate(FrameRate::from_hz(fps)) + // ~8s at 30fps: a bound on how long a client that missed + // every resync hint stays undecodable. + .intra_frame_period(IntraFramePeriod::from_num_frames(240)) .skip_frames(true); let api = OpenH264API::from_source(); - let inner = OhEncoder::with_api_config(api, config) - .map_err(|e| DesktopError::Encode(e.to_string()))?; - Ok(Self { - inner, - bitrate: bitrate_bps, - want_idr: true, - seq: 0, - }) + OhEncoder::with_api_config(api, config).map_err(|e| DesktopError::Encode(e.to_string())) } } +/// Smallest relative bitrate change worth an encoder rebuild — smaller +/// steps are carried by the writer's token-bucket pacing alone, so the +/// controller's gentle recovery probes don't keep resetting rate +/// control state. +const BITRATE_REBUILD_MIN_PCT: u64 = 15; + impl Encoder for H264Encoder { fn encode(&mut self, frame: &RawFrame) -> Result { + if let Some(bps) = self.pending_bitrate.take() { + match Self::build(bps, self.fps) { + Ok(inner) => { + self.inner = inner; + self.bitrate = u64::from(bps); + self.want_idr = true; + } + Err(e) => tracing::warn!("encoder rebuild failed, keeping old rate: {e}"), + } + } if self.want_idr { self.inner.force_intra_frame(); self.want_idr = false; @@ -53,8 +83,10 @@ impl Encoder for H264Encoder { .inner .encode(&yuv) .map_err(|e| DesktopError::Encode(e.to_string()))?; - let keyframe = self.seq.is_multiple_of(240); - self.seq += 1; + // The flag the writer's collapse trusts must be the truth on + // the wire, not a schedule assumption: OpenH264 decides when + // forced and periodic IDRs actually land, so read the NALs. + let keyframe = bitstream_has_idr(&stream); Ok(EncodedFrame { codec: Codec::H264, data: Bytes::from(stream.to_vec()), @@ -67,15 +99,48 @@ impl Encoder for H264Encoder { } fn set_bitrate(&mut self, bps: u32) { - if u64::from(bps) == self.bitrate { + let bps = u64::from(bps); + if bps == 0 || bps == self.bitrate { + return; + } + if bps.abs_diff(self.bitrate) * 100 < self.bitrate * BITRATE_REBUILD_MIN_PCT { return; } - self.bitrate = u64::from(bps); - // OpenH264 requires a rebuild for rate changes; keep the newest - // request applied lazily on the next IDR boundary. + self.pending_bitrate = Some(bps as u32); } } +/// Whether the emitted bitstream carries an IDR picture — the flag the +/// writer's collapse trusts. Read off the NALs rather than assumed: +/// OpenH264 decides when forced and periodic IDRs actually land. +/// NAL units arrive Annex-B wrapped: a 3- or 4-byte start code, then +/// the header byte whose low five bits are the unit type (5 = IDR). +fn bitstream_has_idr(stream: &EncodedBitStream<'_>) -> bool { + for l in 0..stream.num_layers() { + let Some(layer) = stream.layer(l) else { + continue; + }; + for n in 0..layer.nal_count() { + let Some(nal) = layer.nal_unit(n) else { + continue; + }; + // First non-zero byte is the start code's trailing 0x01; + // the byte right after it is the NAL header. + let Some(hdr) = nal + .iter() + .position(|&b| b != 0) + .and_then(|i| nal.get(i + 1)) + else { + continue; + }; + if hdr & 0x1f == 5 { + return true; + } + } + } + false +} + /// OpenH264 decoder producing RGB8 frames. pub struct H264Decoder { inner: OhDecoder, @@ -153,3 +218,52 @@ pub fn bgra_to_i420(frame: &RawFrame) -> YUVBuffer { fn clamp8(v: i32) -> u8 { v.clamp(0, 255) as u8 } + +#[cfg(test)] +mod tests { + use super::*; + + fn frame() -> RawFrame { + let (w, h) = (64u32, 64u32); + RawFrame { + width: w, + height: h, + stride: w * 4, + data: Bytes::from(vec![0x5Au8; (w * h * 4) as usize]), + } + } + + /// The keyframe flag must describe the emitted bitstream: IDR on + /// frame 0 (fresh encoder) and after `request_idr`, deltas between. + #[test] + fn keyframe_flag_matches_emitted_nals() { + let mut enc = H264Encoder::new(1_000_000, 30.0).unwrap(); + assert!(enc.encode(&frame()).unwrap().keyframe, "first frame"); + for _ in 0..3 { + assert!( + !enc.encode(&frame()).unwrap().keyframe, + "delta flagged as keyframe" + ); + } + enc.request_idr(); + assert!( + enc.encode(&frame()).unwrap().keyframe, + "forced IDR not flagged" + ); + assert!(!enc.encode(&frame()).unwrap().keyframe); + } + + /// A bitrate change beyond the deadband rebuilds the encoder — the + /// next frame is an IDR carrying fresh SPS/PPS. + #[test] + fn bitrate_change_rebuilds_with_idr() { + let mut enc = H264Encoder::new(1_000_000, 30.0).unwrap(); + enc.encode(&frame()).unwrap(); + // Below the deadband: no rebuild, no forced keyframe. + enc.set_bitrate(1_100_000); + assert!(!enc.encode(&frame()).unwrap().keyframe); + // Past it: the rebuilt encoder's first frame is an IDR. + enc.set_bitrate(2_000_000); + assert!(enc.encode(&frame()).unwrap().keyframe); + } +} diff --git a/crates/rds-desktop/src/input/x11.rs b/crates/rds-desktop/src/input/x11.rs index 3185632..271c948 100644 --- a/crates/rds-desktop/src/input/x11.rs +++ b/crates/rds-desktop/src/input/x11.rs @@ -17,6 +17,12 @@ const BUTTON_PRESS: u8 = 4; const BUTTON_RELEASE: u8 = 5; const MOTION_NOTIFY: u8 = 6; +/// Max wheel clicks injected per scroll event. Remote-supplied deltas +/// are unbounded f64s — without a cap a single event can loop billions +/// of paired XTEST calls and wedge the injector permanently. 32 lines +/// is already a page-scale scroll. +const MAX_SCROLL_CLICKS: f64 = 32.0; + /// XTEST input sink: injects events into the default X11 session. pub struct XtestInput { conn: RustConnection, @@ -74,7 +80,9 @@ impl InputSink for XtestInput { ), InputKind::Scroll { dx, dy } => { // Emulate wheel clicks: 4 up, 5 down, 6 left, 7 right. - for _ in 0..dy.abs().round() as u32 { + // `.min` bounds the loop; even a NaN delta caps at the + // limit — `f64::min` ignores NaN rather than propagating. + for _ in 0..dy.abs().min(MAX_SCROLL_CLICKS).round() as u32 { let b = if dy > 0.0 { 4 } else { 5 }; self.conn .xtest_fake_input(BUTTON_PRESS, b, 0, self.root, 0, 0, 0) @@ -83,7 +91,7 @@ impl InputSink for XtestInput { .xtest_fake_input(BUTTON_RELEASE, b, 0, self.root, 0, 0, 0) .map_err(|e| DesktopError::Input(e.to_string()))?; } - for _ in 0..dx.abs().round() as u32 { + for _ in 0..dx.abs().min(MAX_SCROLL_CLICKS).round() as u32 { let b = if dx > 0.0 { 6 } else { 7 }; self.conn .xtest_fake_input(BUTTON_PRESS, b, 0, self.root, 0, 0, 0) diff --git a/crates/rds-desktop/src/mailbox.rs b/crates/rds-desktop/src/mailbox.rs index a141821..0557355 100644 --- a/crates/rds-desktop/src/mailbox.rs +++ b/crates/rds-desktop/src/mailbox.rs @@ -8,6 +8,7 @@ use std::collections::VecDeque; use std::sync::Mutex; +use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::sync::Notify; /// Shared state for one queue; both ends clone the `Arc`. @@ -15,6 +16,13 @@ struct State { queue: Mutex>, notify: Notify, capacity: usize, + /// Live `Sender` halves. `recv` returns `None` when this reaches + /// zero. `Arc::strong_count` can't serve here: a `Sender`'s `Drop` + /// runs *before* its `Arc` field is released, so the wake would + /// land while the dying sender still counts — a receiver woken in + /// that window reads a stale non-zero count and re-parks forever. + /// The explicit counter is decremented first, then the wake fires. + senders: AtomicUsize, } /// Producer half: `send` never blocks and never fails while a receiver @@ -30,16 +38,29 @@ pub fn channel(capacity: usize) -> (Sender, Receiver) { queue: Mutex::new(VecDeque::with_capacity(capacity)), notify: Notify::new(), capacity: capacity.max(1), + senders: AtomicUsize::new(1), }); (Sender(state.clone()), Receiver(state)) } impl Clone for Sender { fn clone(&self) -> Self { + self.0.senders.fetch_add(1, Ordering::Relaxed); Self(self.0.clone()) } } +impl Drop for Sender { + /// A parked `recv` must observe the last sender leaving. The count + /// falls BEFORE the wake: a receiver woken early (cross-thread + /// scheduling) that still saw the old count would re-park and never + /// be woken again — the permanent hang this exists to prevent. + fn drop(&mut self) { + self.0.senders.fetch_sub(1, Ordering::AcqRel); + self.0.notify.notify_one(); + } +} + impl Sender { /// Enqueue `item`; evicts the oldest queued item when full. /// Returns the evicted item, if any. @@ -78,8 +99,8 @@ impl Receiver { return Some(item); } } - // All senders dropped → only our own Arc remains. - if std::sync::Arc::strong_count(&self.0) == 1 { + // All senders gone → the queue reads as closed. + if self.0.senders.load(Ordering::Acquire) == 0 { return None; } self.0.notify.notified().await; @@ -115,6 +136,44 @@ mod tests { assert_eq!(rx.recv().await, None); } + #[tokio::test] + async fn parked_recv_wakes_when_last_sender_drops() { + // The session-end order: the consumer is already parked inside + // `recv` when the producer's task finishes and drops the sender. + let (tx, mut rx) = channel::(4); + let waiter = tokio::spawn(async move { rx.recv().await }); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + drop(tx); + assert_eq!( + tokio::time::timeout(std::time::Duration::from_secs(2), waiter) + .await + .expect("parked recv hung after last sender dropped") + .unwrap(), + None + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn parked_recv_survives_drop_wake_race() { + // Drop must decrement the sender count BEFORE waking: a parked + // receiver polled on another worker inside that window used to + // see a stale `strong_count`, re-park, and hang forever. The + // window is a few instructions — loop to give it chances. + for _ in 0..200 { + let (tx, mut rx) = channel::(1); + let waiter = tokio::spawn(async move { rx.recv().await }); + tokio::task::yield_now().await; + drop(tx); + assert_eq!( + tokio::time::timeout(std::time::Duration::from_secs(5), waiter) + .await + .expect("parked recv hung after last sender dropped") + .unwrap(), + None + ); + } + } + #[tokio::test] async fn recv_awaits_produce() { let (tx, mut rx) = channel(4); diff --git a/crates/rds-desktop/src/session.rs b/crates/rds-desktop/src/session.rs index b0f340a..8beb05a 100644 --- a/crates/rds-desktop/src/session.rs +++ b/crates/rds-desktop/src/session.rs @@ -1,10 +1,13 @@ //! Serving side of a desktop session. //! //! Frame delivery follows the MoQ pattern: every encoded frame goes out on -//! its own uni-directional stream carrying a `FrameHeader` v2, newer frames -//! get higher stream priority, and the peer resets streams overtaken by -//! fresher ones. Input events, encoder steering and heartbeats arrive on -//! the bi-directional control stream, which outranks every frame stream. +//! its own uni-directional stream carrying a `FrameHeader`, a fresher +//! queued frame always supersedes a stale one, and a stale frame still +//! in flight is reset mid-send rather than finishing on the wire. +//! Keyframes are never superseded — every delta behind them depends on +//! their landing. Input events, encoder steering and heartbeats arrive +//! on the bi-directional control stream, which outranks every frame +//! stream. use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; @@ -19,6 +22,10 @@ use crate::DesktopError; /// Highest input/control priority; video frames rank below. const CONTROL_PRIORITY: i32 = i32::MAX; +/// Frame streams sit at the midpoint: strictly below control, above +/// QUIC's default so they can't be starved by lower-priority traffic +/// the connection might one day carry. +const FRAME_PRIORITY: i32 = i32::MAX / 2; /// Pacing sample interval for the bitrate controller. const PACING_INTERVAL: Duration = Duration::from_millis(250); @@ -278,18 +285,19 @@ pub async fn serve_desktop_with( }) }; - // Writer task: one uni stream per frame, sent inline — the send - // itself is the only in-flight bound. A continuous producer means - // any frame queued behind an in-progress send is already stale: - // serializing sends keeps the collapse fresh (each transmitted - // frame is the newest available) and bounds concurrent streams - // to one, so opened-but-unsent streams can't pile up. - // The collapse is decode-aware: a queued keyframe always survives - // (deltas produced after it can't decode without it), otherwise - // the newest frame wins. + // Writer task: one uni stream per frame. A continuous producer + // means any frame queued behind an in-progress send is already + // stale — the collapse keeps only the newest (decode-aware: a + // queued keyframe always survives since deltas behind it can't + // decode without it), and a send still in flight when a fresher + // frame arrives is reset mid-write rather than allowed to finish + // (MoQ-style stale reset): the client would drop the tail anyway, + // so its unsent bytes only consume path capacity the fresh frame + // needs. let writer_conn = conn.clone(); let writer_clock = clock.clone(); let writer_bitrate = Arc::clone(&controls.bitrate); + let writer_idr = Arc::clone(&controls.idr); let mut writer = tokio::spawn(async move { // Token bucket on the paced bitrate: offering faster than the // path sustains only backlogs QUIC's send buffer with frames @@ -298,15 +306,21 @@ pub async fn serve_desktop_with( // keyframe can't stall the writer. let mut budget = 0.0f64; let mut last = Instant::now(); - while let Some(mut produced) = rx.recv().await { - let mut have_keyframe = produced.header.keyframe; - while let Ok(newer) = rx.try_recv() { - if newer.header.keyframe || !have_keyframe { - have_keyframe |= newer.header.keyframe; - produced = newer; - } - // A non-keyframe newer than a queued keyframe is - // undecodable without it — skip it, not the keyframe. + let mut pending: Option = None; + 'writer: loop { + let mut produced = match pending.take() { + Some(p) => p, + None => match rx.recv().await { + Some(p) => p, + None => break, + }, + }; + produced = collapse(produced, &mut rx); + // Encode-failure placeholders carry no payload: sending one + // decodes to garbage on the client, while a skipped seq is + // what the client's gap→IDR resync is for. + if produced.payload.is_empty() { + continue; } let bps = writer_bitrate.load(Ordering::Relaxed).max(50_000) as f64 / 8.0; let now = Instant::now(); @@ -317,13 +331,32 @@ pub async fn serve_desktop_with( let wait = ((cost - budget) / bps).min(0.5); tokio::time::sleep(Duration::from_secs_f64(wait)).await; budget = (budget - cost).max(-bps * 0.5); + // Frames produced during the pacing wait are fresher — + // collapse once more before committing to the wire. + produced = collapse(produced, &mut rx); + if produced.payload.is_empty() { + continue; + } } else { budget -= cost; } produced.header.send_ts_ms = writer_clock.now_ms(); - if let Err(e) = write_frame_stream(&writer_conn, produced).await { - tracing::debug!("frame send failed, ending writer: {e}"); - break; + match send_frame(&writer_conn, &produced, &mut rx).await { + SendOutcome::Sent => {} + SendOutcome::Superseded(newer) => pending = Some(newer), + SendOutcome::ResetStale => { + // The dropped tail broke the delta chain — the next + // produced frame must be an IDR, and the queued + // deltas in front of it are undecodable. + writer_idr.store(true, Ordering::Relaxed); + while let Ok(queued) = rx.try_recv() { + if queued.header.keyframe { + pending = Some(queued); + continue 'writer; + } + } + } + SendOutcome::Done | SendOutcome::Failed => break 'writer, } } }); @@ -331,10 +364,23 @@ pub async fn serve_desktop_with( // Control loop: input + encoder steering + heartbeat, until the // peer goes away. `send` also carries DesktopEvent replies. let send_clock = clock.clone(); + let session_display = hello.display; let control = async { loop { match read_frame::<_, DesktopControl>(&mut recv).await { Ok(DesktopControl::Input(ev)) => { + // The grant/display constraint was scoped to the + // hello's display — an event targeting another + // display is out of scope. Skip it (and don't ack: + // an ack reports the event handled). + if ev.display_id != session_display { + tracing::warn!( + event_display = ev.display_id, + session_display, + "input event for out-of-scope display dropped" + ); + continue; + } #[cfg(all(target_os = "linux", feature = "x11"))] if let Err(e) = crate::input::x11::inject(&ev) { tracing::warn!("input injection failed: {e}"); @@ -582,12 +628,109 @@ mod x11 { } } -async fn write_frame_stream(conn: &Connection, produced: Produced) -> Result<(), DesktopError> { - let mut stream = conn.open_uni().await?; - write_frame(&mut stream, &produced.header).await?; - stream.write_all(&produced.payload).await?; - stream.finish()?; - Ok(()) +/// Drain queued frames newest-wins. Decode-aware: once a keyframe is +/// in the mix it absorbs everything — deltas produced after it cannot +/// decode without it, so the keyframe is kept and later deltas are +/// skipped rather than the other way around. +fn collapse(mut produced: Produced, rx: &mut mpsc::Receiver) -> Produced { + let mut have_keyframe = produced.header.keyframe; + while let Ok(newer) = rx.try_recv() { + if newer.header.keyframe || !have_keyframe { + have_keyframe |= newer.header.keyframe; + produced = newer; + } + } + produced +} + +/// Reset code for a frame stream abandoned mid-send — the frame went +/// stale while still in flight, so its tail is dropped instead of +/// consuming path capacity the fresher frame needs. +const STALE_FRAME_RESET: u32 = 0x1; + +/// How one frame send ended. +enum SendOutcome { + /// Frame fully sent. + Sent, + /// A fresher decodable frame supersedes — send it next. + Superseded(Produced), + /// A stale delta was reset mid-send: the reference chain is broken + /// on the client and only an IDR resyncs it, so queued deltas are + /// worthless and the next produced frame must be a keyframe. + ResetStale, + /// Producer closed mid-send; the final frame was finished. + Done, + /// Transport failure — the writer ends. + Failed, +} + +/// Send one frame on its own tagged uni stream, aborting mid-write if +/// a fresher frame lands: an in-flight keyframe is finished (the chain +/// behind it depends on it), a stale delta is reset. +async fn send_frame( + conn: &Connection, + produced: &Produced, + rx: &mut mpsc::Receiver, +) -> SendOutcome { + let mut stream = match conn.open_uni().await { + Ok(s) => s, + Err(e) => { + tracing::debug!("frame stream open failed: {e}"); + return SendOutcome::Failed; + } + }; + // Frame streams rank below the control stream — a stale frame + // must never delay an input event or a resync request. + if let Err(e) = stream.set_priority(FRAME_PRIORITY) { + tracing::debug!("frame stream priority failed: {e}"); + } + // Every uni stream leads with its UniHello tag — the receiver's + // per-connection demux routes on it. + if let Err(e) = write_frame(&mut stream, &rds_core::UniHello::Desktop).await { + tracing::debug!("frame tag write failed: {e}"); + return SendOutcome::Failed; + } + if let Err(e) = write_frame(&mut stream, &produced.header).await { + tracing::debug!("frame header write failed: {e}"); + return SendOutcome::Failed; + } + tokio::select! { + res = async { + stream.write_all(&produced.payload).await.map_err(std::io::Error::other)?; + stream.finish().map_err(std::io::Error::other) + } => match res { + Ok(()) => SendOutcome::Sent, + Err(e) => { + tracing::debug!("frame send failed: {e}"); + SendOutcome::Failed + } + }, + newer = rx.recv() => match newer { + // The producer ended: this is the freshest frame that will + // ever exist — finish it, then the writer drains out. + None => match stream.write_all(&produced.payload).await { + Ok(()) => { + let _ = stream.finish(); + SendOutcome::Done + } + Err(_) => SendOutcome::Failed, + }, + Some(newer) => { + if produced.header.keyframe { + let _ = stream.write_all(&produced.payload).await; + let _ = stream.finish(); + SendOutcome::Superseded(newer) + } else { + let _ = stream.reset(STALE_FRAME_RESET.into()); + if newer.header.keyframe { + SendOutcome::Superseded(newer) + } else { + SendOutcome::ResetStale + } + } + } + }, + } } #[cfg(test)] diff --git a/crates/rds-desktop/tests/session_v2.rs b/crates/rds-desktop/tests/session_v2.rs index 328ebd7..043a19d 100644 --- a/crates/rds-desktop/tests/session_v2.rs +++ b/crates/rds-desktop/tests/session_v2.rs @@ -442,6 +442,130 @@ async fn soak_60fps() { h.server_task.abort(); } +/// v3 uni demux: a live desktop session and a sync pull share ONE +/// connection. Both consume uni streams — Desktop-tagged frame streams +/// and Sync-tagged chunk streams — which the connection demux routes +/// to their own consumer. Pre-v3, two `accept_uni` callers raced and +/// each could swallow the other's streams. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn desktop_and_sync_share_one_connection() { + let clock = SessionClock::default(); + let (server_ep, client_ep, _imp, target) = endpoints(None).await; + + // Sync root with a multi-chunk file to pull while frames stream. + let sync_root = std::env::temp_dir().join(format!( + "rds-coexist-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&sync_root).unwrap(); + let data: Vec = (0..400_000u32).map(|i| (i * 31 % 251) as u8).collect(); + std::fs::write(sync_root.join("media.bin"), &data).unwrap(); + + // Server: dispatch each control stream — Desktop session or Sync — + // like the agent does. + let server_task = tokio::spawn({ + let sync_root = sync_root.clone(); + let clock = clock.clone(); + async move { + let conn = server_ep.accept().await.unwrap().await.unwrap(); + while let Ok((mut send, mut recv)) = conn.accept_bi().await { + let conn = conn.clone(); + let dir = sync_root.clone(); + let clock = clock.clone(); + tokio::spawn(async move { + match read_frame::<_, StreamHello>(&mut recv).await { + Ok(StreamHello::Desktop(hello)) => { + rds_core::write_frame( + &mut send, + &HelloAck::Desktop(rds_core::DesktopCaps { + displays: vec![], + codecs: vec![Codec::H264], + }), + ) + .await + .unwrap(); + let _ = serve_desktop_with( + conn, + send, + recv, + hello, + SessionConfig { + producer: Some(Box::new( + SyntheticProducer::new(90, 320, 240, 1500) + .keyframe_every(30), + )), + clock: Some(clock), + ..Default::default() + }, + ) + .await; + } + Ok(StreamHello::Sync) => { + let _ = rds_sync::engine::serve(conn, send, recv, dir).await; + } + _ => {} + } + }); + } + } + }); + + let conn = client_ep.connect(target, rds_core::ALPN).await.unwrap(); + let mut session = DesktopSession::connect_opts( + &conn, + DesktopHello { + display: 0, + max_fps: 90, + codec: Codec::H264, + input_acks: false, + }, + SessionOpts { + clock: Some(clock.clone()), + }, + ) + .await + .unwrap(); + + // Pull while frames stream: the sync engine claims `UniHello::Sync` + // streams; the session owns `UniHello::Desktop`. A misrouted stream + // either stalls the pull or kills the frame task — both would fail + // the asserts below. + let dest_dir = sync_root.join("dest"); + let conn2 = conn.clone(); + let pull = tokio::spawn(async move { + let (mut send, recv) = conn2.open_bi().await.unwrap(); + rds_core::write_frame(&mut send, &StreamHello::Sync) + .await + .unwrap(); + rds_sync::engine::recv_file(&conn2, "media.bin", &dest_dir, send, recv).await + }); + + // Consume frames while the pull runs. + let mut frames = 0usize; + let deadline = Instant::now() + Duration::from_secs(15); + let mut pull_done = false; + while Instant::now() < deadline && !(pull_done && frames >= 30) { + tokio::select! { + h = session.frame_headers.recv() => if h.is_some() { frames += 1; }, + _ = tokio::time::sleep(Duration::from_millis(50)) => { + pull_done |= pull.is_finished(); + } + } + } + let (dest, stats) = pull.await.unwrap().unwrap(); + assert_eq!(std::fs::read(&dest).unwrap(), data, "pull bytes differ"); + assert!(stats.fetched > 0); + assert!( + frames >= 30, + "desktop starved by concurrent sync: {frames} frames" + ); + server_task.abort(); +} + #[cfg(target_os = "linux")] fn self_rss_kb() -> Option { let status = std::fs::read_to_string("/proc/self/status").ok()?; diff --git a/crates/rds-discovery/Cargo.toml b/crates/rds-discovery/Cargo.toml index 9be616c..85adf2f 100644 --- a/crates/rds-discovery/Cargo.toml +++ b/crates/rds-discovery/Cargo.toml @@ -15,7 +15,6 @@ serde.workspace = true serde_json.workspace = true thiserror.workspace = true tokio.workspace = true -tracing.workspace = true [dev-dependencies] proptest.workspace = true diff --git a/crates/rds-discovery/src/service.rs b/crates/rds-discovery/src/service.rs index 6e30fd9..8f8040c 100644 --- a/crates/rds-discovery/src/service.rs +++ b/crates/rds-discovery/src/service.rs @@ -104,6 +104,22 @@ fn hex16(b: &[u8]) -> String { b.iter().map(|x| format!("{x:02x}")).collect() } +/// Lock acquisition that survives a poisoned lock: every lock here +/// guards plain data (instants, counters, `Option` snapshots) whose +/// invariants a panic cannot tear, so one panicked holder must not +/// fail every request the directory serves from then on. +fn lock(m: &Mutex) -> std::sync::MutexGuard<'_, T> { + m.lock().unwrap_or_else(|e| e.into_inner()) +} + +fn read(l: &RwLock) -> std::sync::RwLockReadGuard<'_, T> { + l.read().unwrap_or_else(|e| e.into_inner()) +} + +fn write(l: &RwLock) -> std::sync::RwLockWriteGuard<'_, T> { + l.write().unwrap_or_else(|e| e.into_inner()) +} + /// Cap on distinct writer labels before accounting folds into `other` /// — the scrape stays bounded under a writer flood. const MAX_WRITER_LABELS: usize = 4096; @@ -120,7 +136,7 @@ impl RateLimiter { /// bounds the verification CPU an unauthenticated peer can burn. fn check_global(&self, limits: &Limits) -> bool { { - let mut start = self.window_start.lock().unwrap(); + let mut start = lock(&self.window_start); if start.elapsed() >= Duration::from_secs(60) { *start = Instant::now(); self.window_count.store(0, Ordering::Relaxed); @@ -131,7 +147,7 @@ impl RateLimiter { /// Per-key interval between accepted PUTs. fn check_key(&self, key: &EndpointKey, limits: &Limits) -> bool { - let mut last = self.last_put.lock().unwrap(); + let mut last = lock(&self.last_put); if let Some(t) = last.get(key) && t.elapsed() < limits.put_min_interval { @@ -301,7 +317,7 @@ fn put_record(state: &State, req: &Request) -> Response { match state.store.put(&record) { Ok(()) => { state.metrics.puts_ok.fetch_add(1, Ordering::Relaxed); - let mut per = state.metrics.endpoint_puts.lock().unwrap(); + let mut per = lock(&state.metrics.endpoint_puts); let label = if per.len() >= MAX_WRITER_LABELS { "other".to_string() } else { @@ -367,7 +383,7 @@ fn get_name(state: &State, name: &str) -> Response { &DiscoveryError::InvalidRecord("invalid device name".into()), ); } - let registry = state.registry.read().unwrap(); + let registry = read(&state.registry); match registry.as_ref().and_then(|r| r.entries.get(name)) { Some(key) => Response::json(200, serde_json::json!({ "key": key.to_string() })), None => Response::error(404, &DiscoveryError::NotFound), @@ -388,11 +404,14 @@ fn put_registry(state: &State, req: &Request) -> Response { return Response::error(400, &DiscoveryError::InvalidRecord(e.to_string())); } }; - let current = state.registry.read().unwrap(); + // Hold the write lock across verify+store — same reason as + // `put_revocations`: the monotonic check runs against `current`, + // and a dropped read lock would let two valid PUTs race so the + // older snapshot wins. + let mut current = write(&state.registry); match snap.verify_fresh(key, current.as_ref()) { Ok(payload) => { - drop(current); - *state.registry.write().unwrap() = Some(payload); + *current = Some(payload); state.metrics.registry_puts.fetch_add(1, Ordering::Relaxed); Response::json(200, serde_json::json!({ "stored": true })) } @@ -404,7 +423,7 @@ fn put_registry(state: &State, req: &Request) -> Response { /// signature themselves, so the stored signed bytes are what travel. /// 404 until the estate publishes the first snapshot. fn get_revocations(state: &State) -> Response { - match &*state.revocations.read().unwrap() { + match &*read(&state.revocations) { Some((snap, _)) => Response::json(200, snap), None => Response::error(404, &DiscoveryError::NotFound), } @@ -424,11 +443,13 @@ fn put_revocations(state: &State, req: &Request) -> Response { return Response::error(400, &DiscoveryError::InvalidRecord(e.to_string())); } }; - let current = state.revocations.read().unwrap(); + // Hold the write lock across verify+store: the monotonic check runs + // against `current`, so it must be the same snapshot we replace — + // a dropped read lock would let two valid PUTs race and regress. + let mut current = write(&state.revocations); match snap.verify_fresh(key, current.as_ref().map(|(_, p)| p)) { Ok(payload) => { - drop(current); - *state.revocations.write().unwrap() = Some((snap, payload)); + *current = Some((snap, payload)); state .metrics .revocations_puts @@ -465,7 +486,7 @@ fn metrics(state: &State) -> Response { ); // Per-endpoint accounting: PUT counts by anonymized writer label // (blake3(key)[..8] — never the key itself; C7 security). - let per = m.endpoint_puts.lock().unwrap(); + let per = lock(&m.endpoint_puts); body.push_str(&format!("rds_directory_writers_distinct {}\n", per.len())); for (writer, count) in per.iter() { body.push_str(&format!( diff --git a/crates/rds-discovery/tests/directory_e2e.rs b/crates/rds-discovery/tests/directory_e2e.rs index a2d0ffd..24c2505 100644 --- a/crates/rds-discovery/tests/directory_e2e.rs +++ b/crates/rds-discovery/tests/directory_e2e.rs @@ -380,6 +380,62 @@ async fn revocations_roundtrip_and_authz() { assert!(matches!(err, DiscoveryError::Http { status: 409, .. })); } +/// Two racing valid PUTs must never leave the older snapshot stored: +/// `verify_fresh`'s monotonic check has to run against the same +/// snapshot the write lock replaces. +#[tokio::test] +async fn concurrent_registry_puts_cannot_regress() { + use rds_discovery::registry::RegistryPayload; + + let reg_key = key(50); + let store = Arc::new(MemoryStore::default()); + let dir = service::serve( + "127.0.0.1:0".parse().unwrap(), + store, + ServiceConfig { + registry_key: Some(reg_key.verifying_key()), + ..Default::default() + }, + ) + .await + .unwrap(); + let client = Arc::new(Client::new(dir.addr())); + + let now = now_unix().unwrap(); + let n = 8u64; + let mut tasks = Vec::new(); + for i in 0..n { + // Snapshot i maps "device" to key byte i, issued_at strictly + // increasing — the identifiable winner is n-1. + let snap = SignedRegistry::sign( + &RegistryPayload { + entries: BTreeMap::from([("device".into(), EndpointKey([i as u8; 32]))]), + issued_at: now + 1000 + i, + expires_at: now + 3600, + }, + ®_key, + ) + .unwrap(); + let client = client.clone(); + tasks.push(tokio::spawn( + async move { client.update_registry(&snap).await }, + )); + } + let mut accepted = 0; + for t in tasks { + if t.await.unwrap().is_ok() { + accepted += 1; + } + } + assert!(accepted >= 1, "every registry PUT failed"); + // However the PUTs interleaved, the stored snapshot is the newest. + assert_eq!( + client.resolve_name("device").await.unwrap(), + EndpointKey([(n - 1) as u8; 32]), + "registry regressed under concurrent PUTs" + ); +} + #[tokio::test] async fn registry_put_refused_without_configured_key() { // No estate key configured: the name API is off and PUTs are 401. diff --git a/crates/rds-net/Cargo.toml b/crates/rds-net/Cargo.toml index bceef4e..18b5b74 100644 --- a/crates/rds-net/Cargo.toml +++ b/crates/rds-net/Cargo.toml @@ -20,14 +20,12 @@ postcard = { workspace = true, features = ["alloc", "use-std"] } rds-core.workspace = true rds-discovery.workspace = true serde.workspace = true -thiserror.workspace = true tokio.workspace = true tokio-stream = { workspace = true, optional = true } tracing.workspace = true [dev-dependencies] iroh-relay = { workspace = true, features = ["server"] } -tracing-subscriber.workspace = true turmoil = "0.7.2" [features] diff --git a/crates/rds-net/src/lib.rs b/crates/rds-net/src/lib.rs index 6b76b68..42ad0f3 100644 --- a/crates/rds-net/src/lib.rs +++ b/crates/rds-net/src/lib.rs @@ -331,7 +331,12 @@ impl Future for Incoming { /// Stream futures and error types are shared — both backends return /// the same `noq` types. #[derive(Clone)] -pub struct Connection(ConnectionInner); +pub struct Connection { + inner: ConnectionInner, + /// Routes inbound uni streams to the consumer that claimed their + /// `UniHello` tag — see [`Connection::uni_streams`]. + demux: std::sync::Arc, +} #[derive(Clone)] enum ConnectionInner { @@ -340,19 +345,136 @@ enum ConnectionInner { Noq(backends::noq::Connection), } +/// Per-connection uni-stream router: one `accept_uni` owner that reads +/// each stream's `UniHello` tag and hands the stream to the consumer +/// that claimed the tag. Without it, independent consumers racing on +/// `accept_uni` steal each other's streams. +#[derive(Default)] +struct UniDemux { + state: std::sync::Mutex, +} + +#[derive(Default)] +struct UniDemuxState { + routes: std::collections::HashMap>, + task: Option>, +} + +/// Demux queue depth per kind. Desktop frame streams arrive one per +/// frame and the consumer drains them into per-stream tasks +/// immediately, so 128 covers bursts without letting a wedged consumer +/// grow memory unboundedly. +const UNI_DEMUX_DEPTH: usize = 128; + +/// Inbound uni streams of one [`rds_core::UniHello`] kind — see +/// [`Connection::uni_streams`]. +pub struct UniStreams { + kind: rds_core::UniHello, + rx: tokio::sync::mpsc::Receiver, +} + +impl UniStreams { + /// The kind this inbox serves. + pub fn kind(&self) -> rds_core::UniHello { + self.kind + } + + /// Next inbound stream of this kind; `None` once the connection + /// dies. + pub async fn recv(&mut self) -> Option { + self.rx.recv().await + } +} + +/// How long an inbound uni stream may sit before writing its `UniHello` +/// tag. A peer that opens streams and never tags them would otherwise +/// park a demux task per stream until the connection dies. +const UNI_TAG_TIMEOUT: Duration = Duration::from_secs(10); + +/// The demux body: accept, then hand each stream its own tag-read task +/// so a peer that stalls before writing the `UniHello` cannot block +/// routing of the streams queued behind it (head-of-line). Runs until +/// the connection dies; a route whose consumer dropped is removed so a +/// later `uni_streams` can reclaim the kind. On exit every registered +/// sender is dropped so parked [`UniStreams::recv`] callers observe +/// `None` — a dead connection ends its inboxes, it does not leave them +/// waiting forever. +async fn uni_demux(conn: Connection, demux: std::sync::Arc) { + loop { + let stream = match conn.accept_uni().await { + Ok(s) => s, + Err(_) => break, + }; + let demux = std::sync::Arc::clone(&demux); + tokio::spawn(async move { + route_uni(stream, &demux).await; + }); + } + let mut st = demux.state.lock().unwrap(); + st.task = None; + st.routes.clear(); +} + +/// Read one stream's tag and hand it to the claimed inbox. Stream order +/// within a kind is not the accept order under parallel tag reads; +/// consumers order by their own wire sequencing (frame `seq`, chunk +/// index). +async fn route_uni(mut stream: RecvStream, demux: &UniDemux) { + let kind = match tokio::time::timeout( + UNI_TAG_TIMEOUT, + rds_core::read_frame::<_, rds_core::UniHello>(&mut stream), + ) + .await + { + Ok(Ok(k)) => k, + Ok(Err(e)) => { + tracing::debug!("uni stream dropped, unreadable tag: {e}"); + return; + } + Err(_) => { + tracing::debug!("uni stream dropped: tag timeout"); + return; + } + }; + let tx = demux.state.lock().unwrap().routes.get(&kind).cloned(); + match tx { + Some(tx) => { + // Backpressure, not loss: a full queue parks the router task + // until the consumer drains it (sync transfers must never + // silently lose a chunk stream). Other streams keep routing. + if tx.send(stream).await.is_err() { + // Only remove the route if the map still holds *this* + // channel — a re-claimed kind must not be clobbered by + // a stale sender's failure. + let mut st = demux.state.lock().unwrap(); + if st.routes.get(&kind).is_some_and(|t| t.same_channel(&tx)) { + st.routes.remove(&kind); + } + } + } + None => tracing::debug!("uni {kind:?} stream dropped: no consumer"), + } +} + impl Connection { fn new_iroh(inner: iroh::endpoint::Connection) -> Self { - Self(ConnectionInner::Iroh(inner)) + Self { + inner: ConnectionInner::Iroh(inner), + demux: Default::default(), + } } #[cfg(feature = "transport-noq")] fn new_noq(inner: backends::noq::Connection) -> Self { - Self(ConnectionInner::Noq(inner)) + Self { + inner: ConnectionInner::Noq(inner), + demux: Default::default(), + } } /// Verified peer identity. pub fn remote_id(&self) -> EndpointId { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c.remote_id(), #[cfg(feature = "transport-noq")] ConnectionInner::Noq(c) => c.remote_id(), @@ -361,7 +483,7 @@ impl Connection { /// Open a bidirectional stream. pub fn open_bi(&self) -> OpenBi<'_> { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c.open_bi(), #[cfg(feature = "transport-noq")] ConnectionInner::Noq(c) => c.open_bi(), @@ -370,7 +492,7 @@ impl Connection { /// Accept the next bidirectional stream opened by the peer. pub fn accept_bi(&self) -> AcceptBi<'_> { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c.accept_bi(), #[cfg(feature = "transport-noq")] ConnectionInner::Noq(c) => c.accept_bi(), @@ -379,7 +501,7 @@ impl Connection { /// Open a unidirectional stream. pub fn open_uni(&self) -> OpenUni<'_> { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c.open_uni(), #[cfg(feature = "transport-noq")] ConnectionInner::Noq(c) => c.open_uni(), @@ -387,17 +509,45 @@ impl Connection { } /// Accept the next unidirectional stream opened by the peer. + /// + /// Prefer [`uni_streams`](Self::uni_streams): on a connection whose + /// services use tagged streams, a direct `accept_uni` races the + /// demux and can steal tagged streams from their consumers. pub fn accept_uni(&self) -> AcceptUni<'_> { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c.accept_uni(), #[cfg(feature = "transport-noq")] ConnectionInner::Noq(c) => c.accept_uni(), } } + /// Claim inbound uni streams tagged `kind` (the `UniHello` first + /// frame every v3 sender writes). The first claim on a connection + /// spawns the shared demux that owns `accept_uni`; each kind allows + /// one live claim — a second registration while the first inbox is + /// alive fails rather than splitting the queue, and a dropped + /// inbox frees the kind for re-claim. + /// + /// Must be called inside a tokio runtime. + pub fn uni_streams(&self, kind: rds_core::UniHello) -> anyhow::Result { + let (tx, rx) = tokio::sync::mpsc::channel(UNI_DEMUX_DEPTH); + let mut st = self.demux.state.lock().unwrap(); + if st.routes.get(&kind).is_some_and(|s| !s.is_closed()) { + anyhow::bail!("uni stream kind {kind:?} already claimed"); + } + st.routes.insert(kind, tx); + if st.task.is_none() { + st.task = Some(tokio::spawn(uni_demux( + self.clone(), + std::sync::Arc::clone(&self.demux), + ))); + } + Ok(UniStreams { kind, rx }) + } + /// Send an unreliable datagram. pub fn send_datagram(&self, data: bytes::Bytes) -> Result<(), SendDatagramError> { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c.send_datagram(data), #[cfg(feature = "transport-noq")] ConnectionInner::Noq(c) => c.send_datagram(data), @@ -406,7 +556,7 @@ impl Connection { /// Receive the next unreliable datagram. pub fn read_datagram(&self) -> ReadDatagram<'_> { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c.read_datagram(), #[cfg(feature = "transport-noq")] ConnectionInner::Noq(c) => c.read_datagram(), @@ -415,7 +565,7 @@ impl Connection { /// Close the connection. pub fn close(&self, error_code: VarInt, reason: &[u8]) { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c.close(error_code, reason), #[cfg(feature = "transport-noq")] ConnectionInner::Noq(c) => c.close(error_code, reason), @@ -425,7 +575,7 @@ impl Connection { /// Whether the connection has closed (either side). Samplers use /// this as their stop condition. pub fn is_closed(&self) -> bool { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c.close_reason().is_some(), #[cfg(feature = "transport-noq")] ConnectionInner::Noq(c) => c.inner().close_reason().is_some(), @@ -435,7 +585,7 @@ impl Connection { /// Snapshot of every live path's transport counters, normalized /// across backends. Used by media pacing (WS5) and metrics (WS7). pub fn path_stats(&self) -> Vec { - match &self.0 { + match &self.inner { ConnectionInner::Iroh(c) => c .paths() .iter() diff --git a/crates/rds-net/tests/metrics.rs b/crates/rds-net/tests/metrics.rs index 97aefd2..8d36f25 100644 --- a/crates/rds-net/tests/metrics.rs +++ b/crates/rds-net/tests/metrics.rs @@ -97,8 +97,9 @@ async fn direct_traffic_counts_direct_not_relay() { assert!(counter(&s, "rds_net_datagrams_sent_total{via=\"direct\"}") > 0); } -/// Relay-only advertised address: every datagram must land on -/// `via=relay` — the split is real, not guessed. +/// Relay-only advertised address: the payload must land on +/// `via=relay` — the split is real, not guessed. Direct-path probes +/// may still appear in the counters (see the assertion comment). #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn relay_only_traffic_counts_relay_not_direct() { let mut relay_config = iroh_relay::server::ServerConfig::default(); @@ -164,13 +165,16 @@ async fn relay_only_traffic_counts_relay_not_direct() { let c = client.metrics(); assert!(counter(&c, "rds_net_datagrams_sent_total{via=\"relay\"}") > 0); - assert_eq!( - counter(&c, "rds_net_datagrams_sent_total{via=\"direct\"}"), - 0 - ); assert!(counter(&c, "rds_net_bytes_sent_total{via=\"relay\"}") >= 9); - assert_eq!(counter(&c, "rds_net_bytes_sent_total{via=\"direct\"}"), 0); + assert!(counter(&c, "rds_net_bytes_received_total{via=\"relay\"}") >= 9); assert!(counter(&c, "rds_net_paths_seen_total{via=\"relay\"}") >= 1); + // No `via=direct == 0` assertions: path pinning caps concurrent + // paths at one, but iroh still fires direct-path probes at + // candidates it learns via in-band address exchange — whether they + // land inside the test window is platform timing (4 datagrams on + // macOS runners, none observed on Linux). They are real datagrams + // and the counter is right to record them; the relay side above is + // what proves the split accounts payload traffic correctly. } /// Prometheus export exists under `metrics` and carries the names the diff --git a/crates/rds-net/tests/uni_demux.rs b/crates/rds-net/tests/uni_demux.rs new file mode 100644 index 0000000..6f7547a --- /dev/null +++ b/crates/rds-net/tests/uni_demux.rs @@ -0,0 +1,121 @@ +//! v3 uni demux lifecycle: a dead connection must end every claimed +//! `UniStreams` inbox (`recv` → `None`) — `rds-sync::receive` and the +//! desktop frame task both rely on end-of-stream to exit instead of +//! parking forever on a dead peer. + +use std::collections::BTreeSet; +use std::time::Duration; + +use rds_net::{EndpointAddr, EndpointConfig, TransportAddr, bind_endpoint}; + +/// Connected pair over the default (iroh) backend, discovery off — +/// returns the client-side connection plus both endpoints. +async fn connected_pair() -> (rds_net::Connection, rds_net::Endpoint, rds_net::Endpoint) { + let cfg = || EndpointConfig::default().without_discovery(); + let server = bind_endpoint(cfg()).await.unwrap(); + let client = bind_endpoint(cfg()).await.unwrap(); + + let server_ep = server.clone(); + let accept = tokio::spawn(async move { server_ep.accept().await.unwrap().await.unwrap() }); + + let mut addrs = BTreeSet::new(); + for a in server.addr().addrs { + if let TransportAddr::Ip(sa) = a { + addrs.insert(TransportAddr::Ip(sa)); + } + } + let conn = client + .connect( + EndpointAddr { + id: server.id(), + addrs, + }, + rds_core::ALPN, + ) + .await + .unwrap(); + let _server_conn = accept.await.unwrap(); + (conn, server, client) +} + +/// `recv` must yield `None` once the connection dies — the demux exits +/// and its registered senders drop, ending the inbox. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn uni_recv_ends_when_connection_dies() { + let (conn, _server, _client) = connected_pair().await; + let mut uni = conn.uni_streams(rds_core::UniHello::Sync).unwrap(); + + conn.close(0u32.into(), b"done"); + tokio::time::sleep(Duration::from_millis(200)).await; + + match tokio::time::timeout(Duration::from_secs(10), uni.recv()).await { + Ok(None) => {} + Ok(Some(_)) => panic!("uni stream arrived after close"), + Err(_) => panic!("uni.recv() hung after connection death"), + } +} + +/// A claim on an already-dead connection ends immediately — the demux +/// respawns, fails `accept_uni`, and drops the fresh sender. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn uni_streams_on_dead_connection_end() { + let (conn, _server, _client) = connected_pair().await; + conn.close(0u32.into(), b"done"); + tokio::time::sleep(Duration::from_millis(200)).await; + + let mut uni = conn.uni_streams(rds_core::UniHello::Desktop).unwrap(); + match tokio::time::timeout(Duration::from_secs(10), uni.recv()).await { + Ok(None) => {} + Ok(Some(_)) => panic!("uni stream arrived on dead connection"), + Err(_) => panic!("uni.recv() hung on dead connection"), + } +} + +/// A peer that opens a uni stream and never writes the `UniHello` tag +/// must not stall routing of the streams behind it — each accepted +/// stream gets its own tag-read task. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stalled_tag_does_not_block_routing() { + let cfg = || EndpointConfig::default().without_discovery(); + let server = bind_endpoint(cfg()).await.unwrap(); + let client = bind_endpoint(cfg()).await.unwrap(); + + let server_ep = server.clone(); + let accept = tokio::spawn(async move { server_ep.accept().await.unwrap().await.unwrap() }); + + let mut addrs = BTreeSet::new(); + for a in server.addr().addrs { + if let TransportAddr::Ip(sa) = a { + addrs.insert(TransportAddr::Ip(sa)); + } + } + let conn = client + .connect( + EndpointAddr { + id: server.id(), + addrs, + }, + rds_core::ALPN, + ) + .await + .unwrap(); + let server_conn = accept.await.unwrap(); + + let mut uni = conn.uni_streams(rds_core::UniHello::Desktop).unwrap(); + + // Stream one: opened, held open, tag never written. + let _stalled = server_conn.open_uni().await.unwrap(); + + // Stream two: tagged properly — must route despite the stalled + // predecessor sitting at the head of the accept queue. + let mut tagged = server_conn.open_uni().await.unwrap(); + rds_core::write_frame(&mut tagged, &rds_core::UniHello::Desktop) + .await + .unwrap(); + + match tokio::time::timeout(Duration::from_secs(5), uni.recv()).await { + Ok(Some(_)) => {} + Ok(None) => panic!("inbox ended while connection is alive"), + Err(_) => panic!("tagged stream stuck behind a stalled tag"), + } +} diff --git a/crates/rds-relay/Cargo.toml b/crates/rds-relay/Cargo.toml index 58a8eb4..d43d9c4 100644 --- a/crates/rds-relay/Cargo.toml +++ b/crates/rds-relay/Cargo.toml @@ -8,21 +8,16 @@ repository.workspace = true [dependencies] anyhow.workspace = true -bytes.workspace = true clap.workspace = true iroh.workspace = true iroh-relay = { workspace = true, features = ["server"] } postcard = { workspace = true, features = ["alloc", "use-std"] } rds-core.workspace = true rds-net = { workspace = true, optional = true } -serde.workspace = true tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true -[dev-dependencies] -rand.workspace = true - [features] default = [] ## Owned relay server on rds-net's noq backend (WS2). diff --git a/crates/rds-relay/src/server.rs b/crates/rds-relay/src/server.rs index 01ce2cc..0dce979 100644 --- a/crates/rds-relay/src/server.rs +++ b/crates/rds-relay/src/server.rs @@ -39,6 +39,9 @@ const RATE_BURST: f64 = 4.0 * 1024.0 * 1024.0; const MAX_RECENT_PEERS: usize = 64; /// Grace between `Drain` broadcast and closing the listener. const DRAIN_GRACE: Duration = Duration::from_secs(2); +/// A connection that never opens its control stream and registers +/// parks a task otherwise — bounded, like the agent's stream hello. +const REGISTER_TIMEOUT: Duration = Duration::from_secs(15); /// A running owned relay. Dropping it leaves tasks detached; call /// [`Relay::close`] for a clean stop or [`Relay::drain`] for a graceful @@ -224,9 +227,14 @@ async fn serve_conn(conn: rds_noq::Connection, state: std::sync::Arc) -> bail!("refused {id}: not on allowlist"); } - // Control stream: first bidi the client opens. - let (mut ctrl_send, mut ctrl_recv) = conn.accept_bi().await?; - let hello = read_control(&mut ctrl_recv).await?; + // Control stream: first bidi the client opens — bounded so a + // connection that never registers costs seconds, not a parked task. + let (mut ctrl_send, mut ctrl_recv) = tokio::time::timeout(REGISTER_TIMEOUT, conn.accept_bi()) + .await + .context("control stream never opened")??; + let hello = tokio::time::timeout(REGISTER_TIMEOUT, read_control(&mut ctrl_recv)) + .await + .context("register read timed out")??; if !matches!(hello, RelayControl::Register) { conn.close(0u32.into(), b"expected register"); bail!("{id} did not register"); diff --git a/crates/rds-sync/Cargo.toml b/crates/rds-sync/Cargo.toml index 2c957b7..d1a0ede 100644 --- a/crates/rds-sync/Cargo.toml +++ b/crates/rds-sync/Cargo.toml @@ -19,7 +19,6 @@ tokio.workspace = true tracing.workspace = true [dev-dependencies] -proptest.workspace = true rds-bench = { workspace = true, features = ["transport-noq"] } noq.workspace = true diff --git a/crates/rds-sync/src/engine.rs b/crates/rds-sync/src/engine.rs index 748766e..6f5fc20 100644 --- a/crates/rds-sync/src/engine.rs +++ b/crates/rds-sync/src/engine.rs @@ -11,19 +11,30 @@ //! dead stream only stalls its own indices and a dropped connection //! resumes from the receiver's journal. -use std::io::{Read, Seek, SeekFrom}; +use std::io::SeekFrom; use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; use anyhow::{Context, bail}; use rds_core::{read_frame, write_frame}; use rds_net::{Connection, RecvStream, SendStream}; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; +use tokio::sync::mpsc; use crate::journal::Journal; use crate::proto::{ CHUNKSET_BATCH, FETCH_STREAMS, MANIFEST_BATCH, MAX_CHUNKS, SyncMsg, bits_to_indices, - check_manifest, check_rel_path, need_bits, + check_manifest, check_rel_path, need_bits, resolve_under, }; -use crate::{Manifest, manifest_of}; +use crate::{Manifest, manifest_of_path}; + +/// No protocol read may stall longer than this — a peer that is alive +/// but silent still must not hang a transfer forever. Generous because +/// reads gate on the peer's disk work (manifest scans, journal +/// rescans); a dead connection ends them regardless. +const READ_STALL: Duration = Duration::from_secs(300); /// Progress/counters a completed (or interrupted) transfer reports. #[derive(Debug, Default, Clone)] @@ -45,7 +56,8 @@ pub async fn serve( mut recv: RecvStream, dir: PathBuf, ) -> anyhow::Result<()> { - match read_frame::<_, SyncMsg>(&mut recv).await? { + let first = read_timed::<_, SyncMsg>(&mut recv).await?; + match first { SyncMsg::Offer { rel_path, size, @@ -59,6 +71,19 @@ pub async fn serve( bail!("offer refused: {e}"); } }; + // The journal creates the root on demand; the resolve below + // needs it to exist. + if let Err(e) = std::fs::create_dir_all(&dir) { + refuse(&mut send, &e.to_string()).await?; + bail!("sync root not writable: {e}"); + } + // Fail fast when the resolved destination would escape the + // root through a symlinked component — assemble re-checks at + // write time, but refusing here saves moving the chunks. + if let Err(e) = resolve_under(&dir, &rel) { + refuse(&mut send, &e.to_string()).await?; + bail!("offer refused: {e}"); + } let manifest = match read_manifest(&mut recv, size, root, chunk_count).await { Ok(m) => m, Err(e) => { @@ -73,7 +98,8 @@ pub async fn serve( chunks = manifest.chunks.len(), "sync push accepted" ); - let stats = receive(&conn, &mut send, &dir, &rel.to_string_lossy(), &manifest).await?; + let (_dest, stats) = + receive(&conn, &mut send, &dir, &rel.to_string_lossy(), &manifest).await?; tracing::info!(?stats, "push receive complete"); Ok(()) } @@ -85,12 +111,17 @@ pub async fn serve( bail!("request refused: {e}"); } }; - let path = dir.join(&rel); - if !path.is_file() { - refuse(&mut send, "no such file").await?; - bail!("requested file absent: {}", rel.display()); - } - let manifest = manifest_of(&std::fs::read(&path)?); + // Lexical check passed — now prove the resolved path stays + // inside the sync root (a symlinked component can't be used + // to read outside it). + let path = match resolve_under(&dir, &rel) { + Ok(p) if p.is_file() => p, + _ => { + refuse(&mut send, "no such file").await?; + bail!("requested file absent or outside root: {}", rel.display()); + } + }; + let manifest = manifest_from_disk(&path).await?; tracing::info!( peer = %conn.remote_id(), rel = %rel.display(), @@ -99,12 +130,12 @@ pub async fn serve( "sync pull serving" ); send_manifest(&mut send, &rel.to_string_lossy(), &manifest).await?; - let SyncMsg::Need { bits } = read_frame::<_, SyncMsg>(&mut recv).await? else { + let SyncMsg::Need { bits } = read_timed::<_, SyncMsg>(&mut recv).await? else { bail!("expected Need"); }; let indices = bits_to_indices(&bits, manifest.chunks.len()); push_chunks(&conn, &path, &manifest, &indices).await?; - match read_frame::<_, SyncMsg>(&mut recv).await? { + match read_timed::<_, SyncMsg>(&mut recv).await? { SyncMsg::Done { .. } => { tracing::info!(sent = indices.len(), "sync pull complete"); Ok(()) @@ -128,7 +159,7 @@ pub async fn send_file( .file_name() .map(|n| n.to_string_lossy().to_string()) .ok_or_else(|| anyhow::anyhow!("{path:?} has no file name"))?; - let manifest = manifest_of(&std::fs::read(path)?); + let manifest = manifest_from_disk(path).await?; tracing::info!( peer = %conn.remote_id(), rel = %rel, @@ -137,13 +168,13 @@ pub async fn send_file( "sync push start" ); send_manifest(&mut send, &rel, &manifest).await?; - let indices = match read_frame::<_, SyncMsg>(&mut recv).await? { + let indices = match read_timed::<_, SyncMsg>(&mut recv).await? { SyncMsg::Need { bits } => bits_to_indices(&bits, manifest.chunks.len()), SyncMsg::Refuse { reason } => bail!("offer refused: {reason}"), other => bail!("expected Need, got {other:?}"), }; push_chunks(conn, path, &manifest, &indices).await?; - match read_frame::<_, SyncMsg>(&mut recv).await? { + match read_timed::<_, SyncMsg>(&mut recv).await? { SyncMsg::Done { root } if root == manifest.root => {} SyncMsg::Refuse { reason } => bail!("receiver refused: {reason}"), other => bail!("expected Done, got {other:?}"), @@ -175,7 +206,7 @@ pub async fn recv_file( }, ) .await?; - let (rel, size, root, chunk_count) = match read_frame::<_, SyncMsg>(&mut recv).await? { + let (rel, size, root, chunk_count) = match read_timed::<_, SyncMsg>(&mut recv).await? { SyncMsg::Offer { rel_path, size, @@ -191,42 +222,179 @@ pub async fn recv_file( other => bail!("expected Offer, got {other:?}"), }; let manifest = read_manifest(&mut recv, size, root, chunk_count).await?; - let stats = receive(conn, &mut send, dest_dir, &rel.to_string_lossy(), &manifest).await?; + let (dest, stats) = + receive(conn, &mut send, dest_dir, &rel.to_string_lossy(), &manifest).await?; tracing::info!(rel = %rel.display(), ?stats, "sync pull complete"); - Ok((dest_dir.join(&rel), stats)) + Ok((dest, stats)) +} + +/// Frame read bounded by the read-stall bound — a peer that stops talking +/// mid-transfer aborts instead of parking the session forever. +async fn read_timed(stream: &mut S) -> anyhow::Result +where + S: tokio::io::AsyncRead + Unpin, + T: serde::de::DeserializeOwned, +{ + match tokio::time::timeout(READ_STALL, read_frame(stream)).await { + Ok(r) => r.map_err(Into::into), + Err(_) => bail!("peer stalled mid-transfer"), + } +} + +/// Manifest of `path` on the blocking pool — chunking + hashing a +/// large file must not park an async worker. +async fn manifest_from_disk(path: &Path) -> anyhow::Result { + let path = path.to_path_buf(); + tokio::task::spawn_blocking(move || manifest_of_path(&path)) + .await + .context("manifest task")? + .context("build manifest") +} + +/// Filesystem-bound chunk store: writes run on a dedicated blocking +/// task so disk latency never parks an async worker mid-transfer, and +/// the wire read pipeline never waits on a flush. Owns the journal; +/// closing `jobs` ends it and hands the journal back for assembly. +struct JournalSink { + jobs: mpsc::Sender<(u32, Vec)>, + /// Verified chunks on disk — the receive loop's completion signal. + present: Arc, + /// First store failure, for error reporting across the task split. + error: Arc>>, + task: tokio::task::JoinHandle>, +} + +impl JournalSink { + fn start(mut journal: Journal) -> Self { + let (jobs, mut job_rx) = mpsc::channel::<(u32, Vec)>(FETCH_STREAMS * 4); + let present = Arc::new(AtomicU64::new(journal.have_set().len() as u64)); + let error = Arc::new(std::sync::Mutex::new(None)); + let (present_w, error_w) = (present.clone(), error.clone()); + let task = tokio::task::spawn_blocking(move || { + while let Some((index, data)) = job_rx.blocking_recv() { + match journal.store(index, &data) { + Ok(true) => { + present_w.fetch_add(1, Ordering::Relaxed); + } + Ok(false) => {} + Err(e) => { + *error_w.lock().unwrap() = Some(e.to_string()); + return Err(e); + } + } + } + Ok(journal) + }); + Self { + jobs, + present, + error, + task, + } + } + + fn present(&self) -> u64 { + self.present.load(Ordering::Relaxed) + } + + /// Queue one verified chunk for storage; backpressures when the + /// disk side falls behind. + async fn put(&self, index: u32, data: Vec) -> anyhow::Result<()> { + self.jobs.send((index, data)).await.map_err(|_| { + let why = self.error.lock().unwrap().take().unwrap_or_default(); + anyhow::anyhow!("journal writer died {why}") + }) + } + + /// Drain queued stores and take the journal back. + async fn finish(self) -> anyhow::Result { + drop(self.jobs); + match self.task.await { + Ok(Ok(j)) => Ok(j), + Ok(Err(e)) => Err(anyhow::anyhow!("chunk store failed: {e}")), + Err(e) => Err(anyhow::anyhow!("journal task join: {e}")), + } + } } /// Receiver half, shared by push and pull: journal the offer, answer /// `Need`, collect chunk streams until complete, assemble, `Done`. +/// Returns the assembled destination path (resolved under the root). async fn receive( conn: &Connection, send: &mut SendStream, dir: &Path, rel: &str, manifest: &Manifest, -) -> anyhow::Result { - let mut journal = Journal::open(dir, rel, manifest)?; +) -> anyhow::Result<(PathBuf, Stats)> { + // Journal open walks and re-verifies every stored part — disk-bound + // work belongs on the blocking pool, not an async worker. + let journal = { + let (dir, rel, manifest) = (dir.to_path_buf(), rel.to_string(), manifest.clone()); + tokio::task::spawn_blocking(move || Journal::open(&dir, &rel, &manifest)) + .await + .context("journal open task")? + .map_err(|e| anyhow::anyhow!("{e}"))? + }; + let total = journal.total() as u64; let bits = need_bits(journal.total(), journal.have_set()); write_frame(send, &SyncMsg::Need { bits }).await?; + // Chunk streams arrive tagged `UniHello::Sync` — routed by the + // connection's demux so a concurrent desktop session on the same + // connection can't consume them. + let mut uni = conn + .uni_streams(rds_core::UniHello::Sync) + .context("claim sync uni streams")?; + let sink = JournalSink::start(journal); + // The peer sends exactly the chunks `Need` asked for — count them + // on the wire, not via `sink.present()`, which the writer task + // advances asynchronously and would lag the final chunk (the loop + // would park in `recv` waiting for streams that never come). + let mut remaining = total - sink.present(); let mut fetched_bytes = 0u64; - while !journal.complete() { - let mut stream = conn.accept_uni().await.context("accept chunk stream")?; + while remaining > 0 { + // `recv` only parks once every routed stream is consumed — a + // stall here means the holder under-delivered. + let mut stream = match tokio::time::timeout(READ_STALL, uni.recv()).await { + Ok(Some(s)) => s, + Ok(None) => { + let _ = sink.finish().await; + bail!("chunk streams ended before transfer completed") + } + Err(_) => { + let _ = sink.finish().await; + bail!("chunk streams stalled") + } + }; loop { - match read_frame::<_, SyncMsg>(&mut stream).await? { + match read_timed::<_, SyncMsg>(&mut stream).await? { SyncMsg::ChunkSet { indices } => { for index in indices { let SyncMsg::ChunkHdr { index: i, len, .. } = - read_frame::<_, SyncMsg>(&mut stream).await? + read_timed::<_, SyncMsg>(&mut stream).await? else { bail!("expected ChunkHdr"); }; if i != index { bail!("chunk stream out of order: {i} != {index}"); } + // The wire len is untrusted: validate it against + // the manifest before it sizes the receive + // buffer (a forged u32 len would otherwise force + // a multi-GiB allocation). + match manifest.chunks.get(i as usize) { + Some(c) if c.len == len => {} + _ => bail!("chunk {i} header len {len} != manifest"), + } let mut buf = vec![0u8; len as usize]; - stream.read_exact(&mut buf).await?; - journal.store(i, &buf).map_err(|e| anyhow::anyhow!("{e}"))?; + match tokio::time::timeout(READ_STALL, stream.read_exact(&mut buf)).await { + Ok(Ok(())) => {} + Ok(Err(e)) => bail!("chunk body read: {e}"), + Err(_) => bail!("chunk body stalled"), + } + sink.put(i, buf).await?; + remaining = remaining.saturating_sub(1); fetched_bytes += u64::from(len); } } @@ -235,7 +403,16 @@ async fn receive( } } } - let dest = journal.assemble(dir).map_err(|e| anyhow::anyhow!("{e}"))?; + let journal = sink.finish().await?; + let fetched = journal.fetched(); + // Assembly concatenates and rehashes every part — blocking pool. + let dest = { + let dir = dir.to_path_buf(); + tokio::task::spawn_blocking(move || journal.assemble(&dir)) + .await + .context("assemble task")? + .map_err(|e| anyhow::anyhow!("{e}"))? + }; write_frame( send, &SyncMsg::Done { @@ -244,11 +421,14 @@ async fn receive( ) .await?; tracing::debug!(?dest, "sync file assembled"); - Ok(Stats { - fetched: journal.fetched(), - total: journal.total() as u64, - bytes: fetched_bytes, - }) + Ok(( + dest, + Stats { + fetched, + total, + bytes: fetched_bytes, + }, + )) } /// Holder half: open [`FETCH_STREAMS`] uni streams, each walking an @@ -275,7 +455,12 @@ async fn push_chunks( return Ok::<(), anyhow::Error>(()); } let mut stream = conn.open_uni().await?; - let mut file = std::fs::File::open(&path)?; + // First frame on every uni stream is its UniHello tag — + // the receiver's demux routes on it. + write_frame(&mut stream, &rds_core::UniHello::Sync).await?; + // tokio's fs file runs every op on the blocking pool — the + // chunk reads below never park an async worker. + let mut file = tokio::fs::File::open(&path).await?; for batch in mine.chunks(CHUNKSET_BATCH) { write_frame( &mut stream, @@ -287,8 +472,8 @@ async fn push_chunks( for &index in batch { let c = manifest.chunks[index as usize]; let mut buf = vec![0u8; c.len as usize]; - file.seek(SeekFrom::Start(c.offset))?; - file.read_exact(&mut buf)?; + file.seek(SeekFrom::Start(c.offset)).await?; + file.read_exact(&mut buf).await?; write_frame( &mut stream, &SyncMsg::ChunkHdr { @@ -353,7 +538,7 @@ async fn read_manifest( } let mut chunks = Vec::with_capacity(chunk_count as usize); while chunks.len() < chunk_count as usize { - match read_frame::<_, SyncMsg>(recv).await? { + match read_timed::<_, SyncMsg>(recv).await? { SyncMsg::ManifestPart { chunks: part } => chunks.extend(part), SyncMsg::Refuse { reason } => bail!("refused: {reason}"), other => bail!("expected ManifestPart, got {other:?}"), diff --git a/crates/rds-sync/src/journal.rs b/crates/rds-sync/src/journal.rs index 7bff0c0..a2466b1 100644 --- a/crates/rds-sync/src/journal.rs +++ b/crates/rds-sync/src/journal.rs @@ -13,7 +13,10 @@ use std::path::{Path, PathBuf}; use serde::{Deserialize, Serialize}; -use crate::{ChunkHash, Manifest, SyncError, proto::check_manifest}; +use crate::{ + ChunkHash, Manifest, SyncError, + proto::{check_manifest, resolve_under}, +}; /// Directory name (under the sync root) holding in-flight state. pub const STATE_DIR: &str = ".rds-sync"; @@ -43,7 +46,10 @@ impl Journal { /// rebuilt from the offer. pub fn open(dest_dir: &Path, rel_path: &str, manifest: &Manifest) -> Result { check_manifest(manifest)?; - let dir = dest_dir.join(STATE_DIR).join(hex(&manifest.root)); + std::fs::create_dir_all(dest_dir)?; + // The state dir is resolved under the canonical root: a + // symlinked `.rds-sync` cannot redirect journal writes outside. + let dir = resolve_under(dest_dir, &Path::new(STATE_DIR).join(hex(&manifest.root)))?; std::fs::create_dir_all(dir.join("parts"))?; // Meta is advisory: pin the destination but never trust it for // chunk truth. A torn/absent meta just gets rewritten. @@ -69,20 +75,28 @@ impl Journal { /// `dest_dir/rel_path` exists, chunk it and count every matching /// hash as present — identical content needs zero wire chunks. /// Chunk boundaries are content-defined, so the same bytes cut - /// identically. + /// identically. Streamed: memory stays at one max-size chunk + /// regardless of file size. fn seed_from_destination(&mut self, dest_dir: &Path) { - let dest = dest_dir.join(&self.meta.rel_path); - let Ok(existing) = std::fs::read(&dest) else { + // No seeding through a symlink that escapes the root. + let Ok(dest) = resolve_under(dest_dir, Path::new(&self.meta.rel_path)) else { return; }; - if existing.len() as u64 != self.manifest.size { + // Cheap gate first: only a regular file of identical size can + // contribute chunks — a manifest scan of anything else is waste. + let Ok(meta) = std::fs::metadata(&dest) else { + return; + }; + if !meta.is_file() || meta.len() != self.manifest.size { return; } - let present: HashSet = crate::manifest_of(&existing) - .chunks - .iter() - .map(|c| c.hash) - .collect(); + let Ok(file) = std::fs::File::open(&dest) else { + return; + }; + let Ok(existing) = crate::manifest_of_reader(file) else { + return; + }; + let present: HashSet = existing.chunks.iter().map(|c| c.hash).collect(); for (i, c) in self.manifest.chunks.iter().enumerate() { if present.contains(&c.hash) { self.have.insert(i as u32); @@ -116,8 +130,10 @@ impl Journal { /// Store one received chunk. The payload is verified against the /// manifest hash BEFORE it touches the state directory — a corrupt - /// or forged chunk is an error, never a part. - pub fn store(&mut self, index: u32, data: &[u8]) -> Result<(), SyncError> { + /// or forged chunk is an error, never a part. Returns `true` when + /// the chunk was newly present (a resent chunk verifies but is not + /// rewritten). + pub fn store(&mut self, index: u32, data: &[u8]) -> Result { let c = self .manifest .chunks @@ -128,16 +144,18 @@ impl Journal { "chunk {index} failed BLAKE3 verification" ))); } + if self.have.contains(&index) { + return Ok(false); + } let part = self.part_path(&c.hash); // Write tmp + rename: a crash mid-write leaves a *.tmp, which // rescan ignores — parts are only ever complete files. let tmp = part.with_extension("tmp"); std::fs::write(&tmp, data)?; std::fs::rename(&tmp, &part)?; - if self.have.insert(index) { - self.fetched += 1; - } - Ok(()) + self.have.insert(index); + self.fetched += 1; + Ok(true) } /// True once every manifest chunk is present and verified. @@ -167,11 +185,14 @@ impl Journal { if !self.complete() { return Err(SyncError::Manifest("assemble before complete".into())); } - let dest = dest_dir.join(&self.meta.rel_path); + // Resolved under the canonical root — a symlinked intermediate + // component is refused rather than followed outside. + let dest = resolve_under(dest_dir, Path::new(&self.meta.rel_path))?; // Dedup fast path: the destination may already hold the exact // content (identical resend) — verify its root and finish. - if let Ok(existing) = std::fs::read(&dest) - && blake3::hash(&existing).as_bytes() == &self.manifest.root + // Streamed hash: no whole-file read. + if let Ok(root) = hash_file(&dest) + && root == self.manifest.root { let _ = std::fs::remove_dir_all(&self.dir); return Ok(dest); @@ -179,11 +200,21 @@ impl Journal { if let Some(parent) = dest.parent() { std::fs::create_dir_all(parent)?; } - let tmp = dest.with_extension("rds-part"); + // Suffix is appended, not substituted, so a peer's literal + // `x.rds-part` name cannot alias the temp file to dest. + let tmp = dest.with_added_extension("rds-part"); + // A pre-existing tmp may be a stale artifact — or a planted + // symlink. Remove it (remove_file unlinks the link itself, not + // its target) and create exclusively so the assembly write can + // never follow a link. + let _ = std::fs::remove_file(&tmp); let mut root = blake3::Hasher::new(); { use std::io::Write; - let mut f = std::fs::File::create(&tmp)?; + let mut f = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&tmp)?; for c in &self.manifest.chunks { let data = std::fs::read(self.part_path(&c.hash))?; root.update(&data); @@ -229,6 +260,28 @@ fn read_meta(dir: &Path) -> Option { postcard::from_bytes(body).ok() } +/// Stream-hash a file — bounded memory regardless of size. +fn hash_file(path: &Path) -> std::io::Result { + let mut f = std::fs::File::open(path)?; + let mut h = blake3::Hasher::new(); + std::io::copy(&mut f, &mut HasherWriter(&mut h))?; + Ok(*h.finalize().as_bytes()) +} + +/// `std::io::Write` adapter that feeds a BLAKE3 hasher — lets +/// `io::copy` stream the file through the digest without buffering. +struct HasherWriter<'a>(&'a mut blake3::Hasher); + +impl std::io::Write for HasherWriter<'_> { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.update(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + fn hex(hash: &ChunkHash) -> String { hash.iter().map(|b| format!("{b:02x}")).collect() } diff --git a/crates/rds-sync/src/lib.rs b/crates/rds-sync/src/lib.rs index bdf5d7d..23ae3e3 100644 --- a/crates/rds-sync/src/lib.rs +++ b/crates/rds-sync/src/lib.rs @@ -58,6 +58,8 @@ pub enum SyncError { } /// Build the manifest for `data` using FastCDC 2020 chunking. +/// Loads nothing extra but requires the whole input in memory — engine +/// paths over files use [`manifest_of_reader`] instead. pub fn manifest_of(data: &[u8]) -> Manifest { manifest_with(data, MIN_CHUNK, AVG_CHUNK, MAX_CHUNK) } @@ -78,6 +80,54 @@ pub fn manifest_with(data: &[u8], min: u32, avg: u32, max: u32) -> Manifest { } } +/// Build the manifest by streaming `r` — the same FastCDC cut points +/// as [`manifest_of`] but with memory bounded at one max-size chunk, +/// which is what the engine needs for files too large to load whole. +pub fn manifest_of_reader(r: impl std::io::Read) -> std::io::Result { + manifest_reader_with(r, MIN_CHUNK, AVG_CHUNK, MAX_CHUNK) +} + +/// Streaming manifest with explicit FastCDC size bounds. +pub fn manifest_reader_with( + r: impl std::io::Read, + min: u32, + avg: u32, + max: u32, +) -> std::io::Result { + let chunker = fastcdc::v2020::StreamCDC::new(r, min as usize, avg as usize, max as usize); + let mut root = blake3::Hasher::new(); + let mut size = 0u64; + let mut chunks = Vec::new(); + for c in chunker { + let c = c.map_err(std::io::Error::from)?; + root.update(&c.data); + size += c.data.len() as u64; + chunks.push(Chunk { + hash: *blake3::hash(&c.data).as_bytes(), + offset: c.offset, + len: c.length as u32, + }); + if chunks.len() > proto::MAX_CHUNKS { + return Err(std::io::Error::other(format!( + "file exceeds {} chunks", + proto::MAX_CHUNKS + ))); + } + } + Ok(Manifest { + size, + root: *root.finalize().as_bytes(), + chunks, + }) +} + +/// Manifest of a file on disk, streamed — O(max chunk) memory, the +/// engine's path for large files. Blocking fs work: call from +/// `spawn_blocking`, not an async worker. +pub fn manifest_of_path(path: &std::path::Path) -> std::io::Result { + manifest_of_reader(std::fs::File::open(path)?) +} + /// Chunks in `b`'s manifest that `a`'s manifest lacks — the wire delta. pub fn missing_chunks<'a>(have: &Manifest, want: &'a Manifest) -> Vec<&'a Chunk> { use std::collections::HashSet; @@ -130,4 +180,27 @@ mod tests { assert_eq!(m.size as usize, data.len()); assert_eq!(m.chunks.iter().map(|c| c.len as u64).sum::(), m.size); } + + #[test] + fn streaming_manifest_matches_slice() { + // The streamed and in-memory chunkers must cut identical + // boundaries — resume/dedup correctness depends on it. + let mut data = vec![0u8; 3_000_000]; + let mut state = 0x9E3779B97F4A7C15u64; + for b in &mut data { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + *b = state as u8; + } + let a = manifest_of(&data); + let b = manifest_of_reader(std::io::Cursor::new(&data)).unwrap(); + assert_eq!(a.size, b.size); + assert_eq!(a.root, b.root); + assert_eq!(a.chunks, b.chunks); + // Edge: empty input streams to an empty manifest. + let e = manifest_of_reader(std::io::Cursor::new(Vec::::new())).unwrap(); + assert_eq!(e.size, 0); + assert!(e.chunks.is_empty()); + } } diff --git a/crates/rds-sync/src/proto.rs b/crates/rds-sync/src/proto.rs index a275517..730753b 100644 --- a/crates/rds-sync/src/proto.rs +++ b/crates/rds-sync/src/proto.rs @@ -16,7 +16,7 @@ //! receiver → holder Done { root } //! ``` -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use serde::{Deserialize, Serialize}; @@ -80,7 +80,9 @@ pub enum SyncMsg { /// Validate a peer-supplied relative path. Returns the safe /// normalized form. Rejects absolute paths, `..` escapes, empty /// components, NULs and overlong input — anything that could land a -/// write outside the sync root. +/// write outside the sync root. This is the *lexical* half of +/// confinement: [`resolve_under`] is the other half, proving the +/// result can't escape through a symlinked component either. pub fn check_rel_path(rel: &str) -> Result { if rel.is_empty() || rel.len() > MAX_REL_PATH { return Err(SyncError::Manifest(format!("bad rel_path {rel:?}"))); @@ -95,6 +97,7 @@ pub fn check_rel_path(rel: &str) -> Result { return Err(SyncError::Manifest(format!("absolute rel_path {rel:?}"))); } let mut out = PathBuf::new(); + let mut first = true; for part in rel.split(['/', '\\']) { match part { "" | "." => {} @@ -103,7 +106,17 @@ pub fn check_rel_path(rel: &str) -> Result { "traversal in rel_path {rel:?}" ))); } - p => out.push(p), + // `.rds-sync` at the root is the journal namespace — a peer + // may not read or plant state files in it. + p if first && p == crate::journal::STATE_DIR => { + return Err(SyncError::Manifest(format!( + "rel_path {rel:?} enters the sync journal" + ))); + } + p => { + first = false; + out.push(p); + } } } if out.as_os_str().is_empty() { @@ -112,6 +125,56 @@ pub fn check_rel_path(rel: &str) -> Result { Ok(out) } +/// Resolve `rel` beneath `root`, refusing any path that would escape +/// the root through a symlinked component. `root` must already exist; +/// the returned path is the canonical deepest-existing ancestor plus +/// the not-yet-created tail of `rel`, so every filesystem operation on +/// it lands inside `root`. +/// +/// `check_rel_path` proves `rel` is lexical; a sync root that itself +/// contains symlinks (managed dotfiles, nix-style dirs, …) could still +/// redirect a read or a `create_dir_all`/`rename` outside — this +/// resolves the existing prefix and requires it to stay under the +/// canonicalized root. Racy relinking between resolve and write is +/// only possible for someone who already has write access to the root. +pub fn resolve_under(root: &Path, rel: &Path) -> Result { + let canon_root = root + .canonicalize() + .map_err(|e| SyncError::Manifest(format!("sync root {}: {e}", root.display())))?; + let mut anc = canon_root.join(rel); + let mut tail: Vec = Vec::new(); + loop { + // symlink_metadata (lstat): a dangling symlink counts as + // existing — canonicalize then fails on it, which is the right + // refusal. + if anc.symlink_metadata().is_ok() { + let canon = anc + .canonicalize() + .map_err(|e| SyncError::Manifest(format!("resolve {}: {e}", anc.display())))?; + if !canon.starts_with(&canon_root) { + return Err(SyncError::Manifest(format!( + "{} escapes the sync root", + rel.display() + ))); + } + let mut out = canon; + for c in tail.iter().rev() { + out.push(c); + } + return Ok(out); + } + let name = anc + .file_name() + .ok_or_else(|| SyncError::Manifest(format!("bad path {}", anc.display())))?; + tail.push(name.to_os_string()); + // canon_root exists, so the walk terminates there at the latest. + anc = anc + .parent() + .ok_or_else(|| SyncError::Manifest("no existing ancestor".into()))? + .to_path_buf(); + } +} + /// Structural validation of a reassembled manifest: chunks sorted, /// non-overlapping, exactly covering `size`, individually bounded, /// list length capped. Content correctness is proven per-chunk by diff --git a/crates/rds-sync/tests/sync_e2e.rs b/crates/rds-sync/tests/sync_e2e.rs index 593d672..e9b4992 100644 --- a/crates/rds-sync/tests/sync_e2e.rs +++ b/crates/rds-sync/tests/sync_e2e.rs @@ -311,7 +311,13 @@ async fn path_traversal_rejected() { for bad in ["../x", "a/../../b", "/abs/path", "..\\win", "a\0b", ""] { assert!(check_rel_path(bad).is_err(), "accepted {bad:?}"); } + // The journal namespace is reserved: no reading or planting state. + for bad in [".rds-sync/meta", ".rds-sync/x/parts/aa", "./.rds-sync/meta"] { + assert!(check_rel_path(bad).is_err(), "accepted {bad:?}"); + } assert!(check_rel_path("dir/sub/file.bin").is_ok()); + // `.rds-sync` deeper in the tree is just a filename — allowed. + assert!(check_rel_path("a/.rds-sync/notes").is_ok()); // And over the wire: a hostile Offer gets Refuse. let (_s, c_ep, target, _task, _server_dir) = pair().await; @@ -337,6 +343,193 @@ async fn path_traversal_rejected() { } } +/// Symlink confinement: a sync root containing links to outside must +/// not serve or write through them — `check_rel_path` is lexical, so +/// `resolve_under` proves the resolved path stays inside. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn symlink_escape_refused() { + let (_s, c_ep, target, _task, server_dir) = pair().await; + let outside = scratch("outside"); + std::fs::write(outside.join("secret.txt"), b"not for sync").unwrap(); + + // `link` inside the sync root points outside it. + #[cfg(unix)] + std::os::unix::fs::symlink(&outside, server_dir.join("link")).unwrap(); + #[cfg(windows)] + std::os::windows::fs::symlink_dir(&outside, server_dir.join("link")).unwrap(); + + // Pull through the link: refused, nothing served. + let conn = client_conn(&c_ep, target.clone()).await; + let (send, recv) = conn.open_bi().await.unwrap(); + let err = recv_file(&conn, "link/secret.txt", &scratch("dest"), send, recv).await; + assert!(err.is_err(), "pull through symlinked dir was served"); + + // Pull of a file that IS a link pointing outside: also refused. + #[cfg(unix)] + std::os::unix::fs::symlink(outside.join("secret.txt"), server_dir.join("alias.txt")).unwrap(); + let conn = client_conn(&c_ep, target.clone()).await; + let (send, recv) = conn.open_bi().await.unwrap(); + let err = recv_file(&conn, "alias.txt", &scratch("dest2"), send, recv).await; + assert!(err.is_err(), "pull of symlink-to-outside was served"); + + // Push into the linked dir: the offer passes the lexical check but + // resolve_under refuses it — the transfer aborts before a chunk + // moves and nothing lands outside. + let data = b"payload-bytes".to_vec(); + let manifest = manifest_of(&data); + let conn = client_conn(&c_ep, target.clone()).await; + let (mut send, mut recv) = conn.open_bi().await.unwrap(); + rds_core::write_frame( + &mut send, + &rds_sync::proto::SyncMsg::Offer { + rel_path: "link/victim.bin".into(), + size: manifest.size, + root: manifest.root, + chunk_count: manifest.chunks.len() as u32, + }, + ) + .await + .unwrap(); + let verdict = tokio::time::timeout( + Duration::from_secs(10), + rds_core::read_frame::<_, rds_sync::proto::SyncMsg>(&mut recv), + ) + .await; + match verdict { + Err(_) | Ok(Err(_)) => panic!("escape attempt killed the stream, expected Refuse"), + Ok(Ok(rds_sync::proto::SyncMsg::Refuse { .. })) => {} + Ok(Ok(other)) => panic!("escape attempt got {other:?}"), + } + assert!( + !outside.join("victim.bin").exists(), + "push wrote through the symlink outside the root" + ); + + // And a symlinked `.rds-sync` can't redirect the journal either: + // plant one, then run a normal push — it must be refused rather + // than journal state landing outside. + let jail = scratch("jailed-server"); + let journal_out = scratch("journal-out"); + #[cfg(unix)] + std::os::unix::fs::symlink(&journal_out, jail.join(".rds-sync")).unwrap(); + let jail_ep = bind_endpoint(EndpointConfig::default()).await.unwrap(); + let jail_task = spawn_server(jail_ep.clone(), jail.clone()); + let jail_target = jail_ep.addr(); + let conn = client_conn(&c_ep, jail_target).await; + let (mut send, mut recv) = conn.open_bi().await.unwrap(); + rds_core::write_frame( + &mut send, + &rds_sync::proto::SyncMsg::Offer { + rel_path: "ok.bin".into(), + size: manifest.size, + root: manifest.root, + chunk_count: manifest.chunks.len() as u32, + }, + ) + .await + .unwrap(); + rds_core::write_frame( + &mut send, + &rds_sync::proto::SyncMsg::ManifestPart { + chunks: manifest.chunks.clone(), + }, + ) + .await + .unwrap(); + let verdict = tokio::time::timeout( + Duration::from_secs(10), + rds_core::read_frame::<_, rds_sync::proto::SyncMsg>(&mut recv), + ) + .await; + match verdict { + // Journal::open fails → the stream dies without Need. + Err(_) | Ok(Err(_)) => {} + Ok(Ok(other)) => panic!("journal-through-symlink got {other:?}"), + } + assert!( + std::fs::read_dir(&journal_out).unwrap().next().is_none(), + "journal state escaped through .rds-sync symlink" + ); + jail_task.abort(); +} + +/// A forged `ChunkHdr` length must not size the receive buffer: the +/// header is checked against the manifest before allocation. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn forged_chunk_len_rejected() { + let (_s, c_ep, target, _task, _server_dir) = pair().await; + let conn = client_conn(&c_ep, target).await; + let (mut send, mut recv) = conn.open_bi().await.unwrap(); + + // Offer a real one-chunk file so the manifest validates. + let data = random_bytes(64 * 1024, 0xBAD); + let manifest = manifest_of(&data); + rds_core::write_frame( + &mut send, + &rds_sync::proto::SyncMsg::Offer { + rel_path: "victim.bin".into(), + size: manifest.size, + root: manifest.root, + chunk_count: manifest.chunks.len() as u32, + }, + ) + .await + .unwrap(); + rds_core::write_frame( + &mut send, + &rds_sync::proto::SyncMsg::ManifestPart { + chunks: manifest.chunks.clone(), + }, + ) + .await + .unwrap(); + // Server answers Need. + match rds_core::read_frame::<_, rds_sync::proto::SyncMsg>(&mut recv) + .await + .unwrap() + { + rds_sync::proto::SyncMsg::Need { .. } => {} + other => panic!("expected Need, got {other:?}"), + } + + // Chunk stream: tag, set, then a ChunkHdr claiming 4 GiB. + let mut stream = conn.open_uni().await.unwrap(); + rds_core::write_frame(&mut stream, &rds_core::UniHello::Sync) + .await + .unwrap(); + rds_core::write_frame( + &mut stream, + &rds_sync::proto::SyncMsg::ChunkSet { indices: vec![0] }, + ) + .await + .unwrap(); + rds_core::write_frame( + &mut stream, + &rds_sync::proto::SyncMsg::ChunkHdr { + index: 0, + hash: manifest.chunks[0].hash, + len: u32::MAX, + }, + ) + .await + .unwrap(); + stream.write_all(b"short").await.unwrap(); + stream.finish().unwrap(); + + // The receiver must abort the transfer on the len mismatch — the + // control stream ends without Done. + let verdict = tokio::time::timeout( + Duration::from_secs(10), + rds_core::read_frame::<_, rds_sync::proto::SyncMsg>(&mut recv), + ) + .await; + match verdict { + Err(_) | Ok(Err(_)) => {} + Ok(Ok(rds_sync::proto::SyncMsg::Refuse { .. })) => {} + Ok(Ok(other)) => panic!("forged chunk len accepted: {other:?}"), + } +} + /// Impaired lane: 5% loss + delay + jitter under both sockets, plus a /// mid-transfer connection drop — the file still lands byte-identical /// and resume re-fetches only what was lost. diff --git a/deny.toml b/deny.toml index 02fcd13..66b2d21 100644 --- a/deny.toml +++ b/deny.toml @@ -1,5 +1,6 @@ # cargo-deny policy for remote-device-sync. -# Run: `cargo deny check`. CI runs `cargo deny check licenses bans`. +# Run: `cargo deny check`. CI runs +# `cargo deny check bans licenses advisories sources` (rust-supply-chain). [licenses] allow = [ @@ -10,10 +11,15 @@ allow = [ "BSD-3-Clause", "BSL-1.0", "CC0-1.0", + # CDLA-Permissive-2.0: webpki-roots/webpki-root-certs (Mozilla root + # store data) via rustls-platform-verifier/iroh-relay. + "CDLA-Permissive-2.0", "ISC", "MIT", "MPL-2.0", "OpenSSL", + # Unlicense: ws_stream_wasm/async_io_stream/pharos via iroh-relay. + "Unlicense", "Unicode-3.0", "Unicode-DFS-2016", "Zlib", @@ -27,3 +33,11 @@ wildcards = "warn" [advisories] # Cargo.lock is committed; advisories gate on it. db-path = "~/.cargo/advisory-db" +ignore = [ + # atomic-polyfill (unmaintained) via heapless <- postcard: no safe + # upgrade exists; severity is informational, not a vulnerability. + "RUSTSEC-2023-0089", + # paste (unmaintained) via netlink-packet-core <- netwatch <- iroh: + # no safe upgrade exists; informational only. + "RUSTSEC-2024-0436", +] diff --git a/deploy/systemd/rds-agent.service b/deploy/systemd/rds-agent.service index 4962cfb..3249ce5 100644 --- a/deploy/systemd/rds-agent.service +++ b/deploy/systemd/rds-agent.service @@ -37,6 +37,11 @@ LimitNOFILE=65536 NoNewPrivileges=yes ProtectSystem=strict ProtectHome=yes +# NOTE on desktop capture: PrivateTmp gives the unit its own /tmp, which +# hides /tmp/.X11-unix — X11 capture/input then can't reach the display. +# For desktop serving run the agent as the session user (see +# docs/deployment.md), or drop PrivateTmp on dedicated X11 rigs. Abstract +# X11 sockets are unaffected; tcp/6000 is not used by XTEST paths. PrivateTmp=yes PrivateDevices=yes ProtectKernelTunables=yes diff --git a/docs/architecture.md b/docs/architecture.md index d57aab1..ab899ca 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -171,24 +171,51 @@ Every stream opens with a length-prefixed postcard `StreamHello`: | `Desktop` | bi + uni | hello/capabilities; input events client→server; one uni stream per video frame server→client | | `Sync` | bi + uni | offer/request → manifest parts → `Need` bitmap → chunk pull on 4 dedicated uni streams → `Done` | +Every uni stream leads with a `UniHello` tag frame (protocol v3). The +accepting side runs one per-connection demux (`Connection::uni_streams`) +that routes each stream to the consumer registered for its tag — a +desktop session and a sync pull can share a connection without either +stealing the other's streams off `accept_uni`. The demux accepts, then +hands each stream its own tag-read task under a 10s bound: a peer that +opens a stream and never writes its tag cannot stall the routing of the +streams queued behind it, and a claimed inbox whose consumer dropped is +reclaimable by the next `uni_streams` call. + Desktop media: capture → BGRA→I420 → H.264 (OpenH264 baseline, no B-frames; hw encoders behind a trait) → per-frame uni stream with a `FrameHeader` -`{seq, keyframe, capture_ts_ms, send_ts_ms}`. Freshness is enforced -twice rather than by stream reset: the producer→writer channel is a -bounded collapse (a queued keyframe always survives; otherwise newest -wins), sends are serialized and token-bucket-paced to the controller's -bitrate so QUIC's own buffer never fills with stale-on-arrival frames, -and the receiver drops anything below a "next expected seq" watermark — -a delivered-seq gap auto-requests an IDR, and a backpressured keyframe -re-arms the producer's IDR flag. The control stream (`DesktopControl`: -input events, `RequestIdr`, `SetBitrate`, heartbeats; `DesktopEvent`: -input acks, heartbeat echoes) runs at max stream priority; per-frame -priorities were tried and removed — under load they starve in-flight -streams. This yields decode-what-survives behavior without a custom -UDP stack. +`{seq, keyframe, capture_ts_ms, send_ts_ms}`. Freshness is enforced at +every stage: the producer→writer channel collapses to the newest queued +frame (a queued keyframe always survives — deltas behind it cannot decode +without it), sends are serialized and token-bucket-paced to the +controller's bitrate so QUIC's own buffer never fills with +stale-on-arrival frames, and a frame that goes stale *while its stream +is still sending* is reset mid-write (MoQ-style) — a stale delta's tail +only consumes path capacity the fresher frame needs. The reset breaks +the client's delta chain, so the producer's IDR flag is re-armed and +queued deltas are drained. The encoder's `keyframe` flag is read off the +emitted NAL units (forced IDRs, periodic IDRs at the configured +~8s `intra_frame_period`, and encoder rebuilds on a >15% bitrate change +all mark real IDRs) — never assumed from a schedule. The receiver drops +anything below a "next expected seq" watermark, decodes with a +session-local decoder (a shared one would cross-contaminate reference +chains), and auto-requests an IDR on a delivered-seq gap or a decode +failure, rate-limited so a corrupt stretch cannot storm. Frame stream +headers and bodies are bounded (32 MiB cap, 30s stall); every stream +leads with its `UniHello` tag under a 10s bound. The control stream +(`DesktopControl`: input events, `RequestIdr`, `SetBitrate`, heartbeats; +`DesktopEvent`: input acks, heartbeat echoes) runs at max stream +priority; frame streams sit at a constant midpoint — above QUIC's +default, strictly below control. Per-frame *escalating* priorities were +tried and removed — under load they starve in-flight sends — but a +constant rank keeps frames ahead of background traffic without frames +fighting each other. This yields decode-what-survives behavior without +a custom UDP stack. File sync (`rds send`/`rds recv`, agent `--sync-dir`): the file is cut -by FastCDC into BLAKE3-addressed chunks. The control stream carries +by FastCDC into BLAKE3-addressed chunks — streamed (`StreamCDC`), so +manifest memory stays at one max-size chunk regardless of file size and +disk-bound work (manifest scan, journal open/verify, assembly) runs on +the blocking pool, never an async worker. The control stream carries `Offer`/`Request` then the manifest in ≤512-entry `ManifestPart` batches (a 1 GiB manifest exceeds the 64 KiB frame cap). The receiver opens a journal under `/.rds-sync//`, re-verifies every @@ -199,9 +226,17 @@ indices (≤4096/batch) then chunk payloads across 4 dedicated uni streams; `SetDone`/`Done` close the session. Assembly concatenates verified parts, checks the BLAKE3 root, and renames atomically — a torn or corrupt part is refetched, a killed transfer resumes from the -journal, and `rel_path` is validated against traversal, absolute and -NUL paths. One sync session per connection (chunk streams share the -connection's `accept_uni` queue). +journal, and `rel_path` is validated twice: lexically (traversal, +absolute, NUL, the `.rds-sync` journal namespace) and by resolution — +the canonicalized destination must stay inside the canonicalized sync +root, so a symlinked component can't redirect reads, journal state or +assembly outside it. Verified chunks are written by a dedicated +blocking-pool sink behind a bounded queue, so disk latency never parks +the wire pipeline; completion is counted on the wire (the peer sends +exactly the `Need` set), not on the sink's lagging counter. Every +protocol read and chunk body is bounded by a 300s stall — a peer alive +but silent aborts rather than parking the session. One sync session per +connection. ### Stability measures @@ -209,9 +244,17 @@ connection's `accept_uni` queue). hole-punch upgrade — both handled by iroh. - QUIC connection migration survives NAT rebinding/Wi-Fi↔LTE moves. - Agent reconnects to relay with backoff; CLI can pin `--relay`. -- Serialized frame sends + collapse bound worst-case latency under - loss: queues stay near-empty and the residual tail is retransmit - physics, not queueing. +- Serialized frame sends + collapse + mid-send stale reset bound + worst-case latency under loss: queues stay near-empty and the + residual tail is retransmit physics, not queueing. +- Every handshake and stream stage is stall-bounded: agent stream hello + 15s, desktop session ack / frame stream header+body 30s, relay + register 15s, uni-stream tag 10s, CLI acks 15s and connect 30s, + sync protocol reads 300s. A peer that opens a stream and goes silent + costs seconds, not a parked task for the connection's lifetime. +- Agent state mutexes recover from poisoning (`into_inner`) — one + panicked holder cannot deny service forever, and request paths refuse + explicitly instead of `unreachable!`. ### Observability diff --git a/docs/deployment.md b/docs/deployment.md index 249ba26..643788e 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -81,6 +81,9 @@ Caveats: session display and `/dev/uinput` — run the agent as a user service in the graphical session for desktop, or relax `PrivateDevices` and add `DeviceAllow=/dev/uinput rw` plus `SupplementaryGroups=input`. + `PrivateTmp` also hides `/tmp/.X11-unix` from the unit — the session + X socket won't be reachable unless the display listens on the + abstract socket (default on Linux) or `PrivateTmp` is dropped. - **Agent `--sync-dir`**: add its path to `ReadWritePaths` or keep it under `/var/lib/rds-agent`. - **Key material**: `endpoint.key` must stay `0600` and owned by the diff --git a/docs/reports/checkpoint-c5.md b/docs/reports/checkpoint-c5.md index 065586f..00b89b7 100644 --- a/docs/reports/checkpoint-c5.md +++ b/docs/reports/checkpoint-c5.md @@ -20,7 +20,13 @@ Verdict: **pending review** - FrameHeader decoder + stream demux fuzz (proptest): PASS ## Manual checklist (fill before merge) -- [ ] All "not run" items above explained -- [ ] Reports committed: bench-*.json, bench-*.md, this file -- [ ] docs/ updated for anything this wave changed -- [ ] security/unsafe review done for new code paths +- [x] All "not run" items above explained — the 30min soak is the only + conditional row and is noted honestly (20s smoke ran in-gate) +- [x] Reports committed: this file; C5 evidence is the session_v2 e2e + suite (bench artifacts belong to the C1/C7 waves) +- [x] docs/ updated: architecture.md stream/media protocol sections +- [x] security/unsafe review done: review found a parked-`recv` + lost-wakeup in `mailbox` (fixed: `Sender::drop` notifies; regression + test `parked_recv_wakes_when_last_sender_drops`), an unbounded X11 + scroll loop (clamped at 32 clicks/event), and per-event display_id + now checked against the session display — all fixed in this wave diff --git a/docs/reports/checkpoint-c6.md b/docs/reports/checkpoint-c6.md index 623e308..9d6aa1e 100644 --- a/docs/reports/checkpoint-c6.md +++ b/docs/reports/checkpoint-c6.md @@ -18,7 +18,14 @@ Verdict: **pending review** ≤256K chunks, every wire frame ≤64KB MAX_MESSAGE_LEN ## Manual checklist (fill before merge) -- [ ] All "not run" items above explained -- [ ] Reports committed: bench-*.json, bench-*.md, this file -- [ ] docs/ updated for anything this wave changed -- [ ] security/unsafe review done for new code paths +- [x] All "not run" items above explained — every check ran +- [x] Reports committed: this file; C6 evidence is the sync_e2e suite + (now 12 tests — adds symlink-escape refusal and forged-chunk-len + rejection) +- [x] docs/ updated: architecture.md documents resolved-path + confinement and the v3 uni demux +- [x] security/unsafe review done: review found sync confinement was + lexical-only (symlinked components could escape the root — fixed by + `resolve_under` on pull/journal/assemble plus `.rds-sync` namespace + refusal), and `ChunkHdr.len` sized the receive buffer unchecked + (now verified against the manifest first) — fixed in this wave