Compare commits

..
Author SHA1 Message Date
renovate Bot 29f59b0c70 chore(deps): update github actions
Rust CI / cargo fmt (pull_request) Successful in 16s
Rust CI / cargo clippy (pull_request) Successful in 38s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (pull_request) Successful in 31s
Rust CI / cargo check (x86_64-unknown-linux-musl) (pull_request) Successful in 35s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (pull_request) Successful in 1m20s
Rust CI / cargo doc (pull_request) Successful in 29s
Rust CI / cargo test (x86_64-unknown-linux-musl) (pull_request) Successful in 1m33s
Proto Fuzz Smoke / cargo fuzz smoke (pull_request) Successful in 1m28s
Netem Interop / netem interop (pull_request) Successful in 1m14s
Supply chain / cargo-deny advisories (pull_request) Successful in 11s
Supply chain / cargo-deny bans licenses sources (pull_request) Successful in 9s
Supply chain / actionlint (pull_request) Successful in 3s
2026-09-14 05:07:41 +00:00
Nils Andreas Svee b9c2d024d9 chore(release): irtt-client 0.5.1, irtt-rs 0.7.0
Rust CI / cargo fmt (push) Successful in 40s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 1m3s
Rust CI / cargo clippy (push) Successful in 1m3s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m48s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Successful in 1m2s
Rust CI / cargo doc (push) Successful in 38s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Successful in 1m51s
Proto Fuzz Smoke / cargo fuzz smoke (push) Successful in 1m45s
Supply chain / cargo-deny advisories (push) Successful in 14s
Supply chain / cargo-deny bans licenses sources (push) Successful in 11s
Netem Interop / netem interop (push) Successful in 1m36s
Supply chain / actionlint (push) Successful in 7s
Semver / cargo-semver-checks (push) Successful in 1m31s
Publish irtt-server image / publish (push) Successful in 3m9s
irtt-client exposes a synchronous wait for managed target-set updates and keeps its packaged test dependency resolvable from crates.io. irtt-rs adds stdin-managed desired target sets and per-target HMAC configuration.
2026-09-01 12:55:44 +07:00
Nils Andreas Svee 7ee7474605 fix(ci): use GNU target for fuzzing
Rust CI / cargo fmt (push) Successful in 41s
Rust CI / cargo clippy (push) Successful in 1m1s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 1m1s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m44s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Successful in 59s
Rust CI / cargo doc (push) Successful in 47s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Successful in 1m50s
Proto Fuzz Smoke / cargo fuzz smoke (push) Successful in 1m41s
Supply chain / cargo-deny advisories (push) Successful in 15s
Netem Interop / netem interop (push) Successful in 1m31s
Supply chain / cargo-deny bans licenses sources (push) Successful in 11s
Supply chain / actionlint (push) Successful in 7s
2026-09-01 12:48:01 +07:00
Nils Andreas Svee c18903b097 ci: download Gitea tool binaries
Rust CI / cargo fmt (push) Successful in 42s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 1m3s
Rust CI / cargo clippy (push) Successful in 1m4s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m47s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Successful in 51s
Rust CI / cargo doc (push) Successful in 44s
Proto Fuzz Smoke / cargo fuzz smoke (push) Failing after 41s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Successful in 1m43s
Supply chain / cargo-deny advisories (push) Successful in 14s
Supply chain / cargo-deny bans licenses sources (push) Successful in 11s
Netem Interop / netem interop (push) Successful in 1m29s
Supply chain / actionlint (push) Successful in 7s
2026-09-01 12:27:45 +07:00
Nils Andreas Svee b397916ec6 fix(ci): run cargo-deny directly on Gitea
Rust CI / cargo fmt (push) Successful in 34s
Rust CI / cargo clippy (push) Successful in 54s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 53s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m42s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Successful in 51s
Rust CI / cargo doc (push) Successful in 43s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Successful in 1m55s
Netem Interop / netem interop (push) Successful in 1m32s
Proto Fuzz Smoke / cargo fuzz smoke (push) Successful in 2m5s
Supply chain / actionlint (push) Failing after 14s
Supply chain / cargo-deny advisories (push) Successful in 2m34s
Supply chain / cargo-deny bans licenses sources (push) Successful in 2m25s
2026-09-01 12:06:59 +07:00
Nils Andreas Svee 9f6ef340f9 docs(managed): fix broken intra-doc link for BlockingManagedClient
Rust CI / cargo fmt (push) Successful in 35s
Rust CI / cargo clippy (push) Successful in 55s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 50s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m43s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Successful in 59s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Successful in 1m49s
Rust CI / cargo doc (push) Successful in 47s
Proto Fuzz Smoke / cargo fuzz smoke (push) Successful in 2m5s
Supply chain / cargo-deny advisories (push) Canceled after 0s
Supply chain / cargo-deny bans licenses sources (push) Canceled after 0s
Supply chain / actionlint (push) Canceled after 0s
Netem Interop / netem interop (push) Successful in 1m33s
Update the link path to use a crate-relative reference, ensuring the documentation resolves correctly for users.
2026-09-01 12:02:01 +07:00
Nils Andreas Svee 3a29b4efb2 ci: make Gitea primary Linux forge
Rust CI / cargo fmt (push) Successful in 45s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 1m4s
Rust CI / cargo clippy (push) Successful in 1m6s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m48s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Successful in 50s
Rust CI / cargo doc (push) Failing after 48s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Successful in 1m49s
Proto Fuzz Smoke / cargo fuzz smoke (push) Successful in 2m3s
Supply chain / cargo-deny advisories (push) Failing after 29s
Netem Interop / netem interop (push) Successful in 1m33s
Supply chain / cargo-deny bans licenses sources (push) Failing after 27s
Supply chain / actionlint (push) Failing after 14s
2026-09-01 11:18:51 +07:00
Lochnair a2f542bb97 Merge pull request 'feat(app): add stdin-managed targets and per-target HMAC' (#136) from feature/targets-stdin into main
Reviewed-on: #136
2026-09-01 04:06:19 +00:00
Nils Andreas Svee 5dd478e81a fix(app): bound dynamic stream accounting 2026-09-01 10:36:05 +07:00
Nils Andreas Svee 1ba4f61072 feat(app): add stdin-managed client target sets 2026-09-01 01:50:56 +07:00
Nils Andreas Svee 31fd76f78d feat(client): await managed updates from blocking callers 2026-09-01 01:49:59 +07:00
Nils Andreas Svee 344533f8b6 feat(app): parse target-level HMAC authentication 2026-09-01 01:49:48 +07:00
renovate Bot e8b3f64292 Merge pull request 'chore(deps): update github actions to v0.6.3' (#135) from renovate/github-actions into main 2026-08-30 22:07:41 +00:00
renovate Bot c98452aa2c chore(deps): update github actions to v0.6.3 2026-08-30 22:07:36 +00:00
renovate Bot 807d8f2846 chore(deps): maintain cargo lockfile 2026-08-29 14:40:37 +00:00
renovate Bot 0cc6b1ea80 chore(deps): update github actions 2026-08-29 14:40:09 +00:00
renovate Bot 0ccca9b0ab chore(deps): update rust docker tag to v1.98 2026-08-29 14:39:18 +00:00
renovate Bot df0d69e35d fix(deps): update rust crate libfuzzer-sys to 0.4.13 2026-08-29 14:38:56 +00:00
renovate Bot 9ab1b78637 chore(deps): pin dependencies 2026-08-29 14:36:38 +00:00
31 changed files with 1732 additions and 409 deletions

No files matched your search

@@ -7,7 +7,6 @@ on:
pull_request: pull_request:
branches: branches:
- main - main
merge_group:
workflow_dispatch: workflow_dispatch:
permissions: permissions:
@@ -32,7 +31,7 @@ jobs:
uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1 uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
with: with:
components: rustfmt components: rustfmt
cache-key: fmt cache: false
- name: Check formatting - name: Check formatting
uses: actions-rust-lang/rustfmt@4066006ec54a31931b9b1fddfd38f2fdf2d27143 # v1 uses: actions-rust-lang/rustfmt@4066006ec54a31931b9b1fddfd38f2fdf2d27143 # v1
@@ -52,7 +51,7 @@ jobs:
uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1 uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
with: with:
components: clippy components: clippy
cache-key: clippy cache: false
- name: Run clippy - name: Run clippy
run: | run: |
@@ -79,46 +78,22 @@ jobs:
matrix: matrix:
include: include:
- task: check - task: check
runner: ubuntu-24.04
target: x86_64-unknown-linux-gnu target: x86_64-unknown-linux-gnu
musl: false musl: false
- task: test - task: test
runner: ubuntu-24.04
target: x86_64-unknown-linux-gnu target: x86_64-unknown-linux-gnu
musl: false musl: false
- task: check - task: check
runner: ubuntu-24.04
target: x86_64-unknown-linux-musl target: x86_64-unknown-linux-musl
musl: true musl: true
- task: test - task: test
runner: ubuntu-24.04
target: x86_64-unknown-linux-musl target: x86_64-unknown-linux-musl
musl: true musl: true
- task: check runs-on: ubuntu-latest
runner: ubuntu-24.04-arm
target: aarch64-unknown-linux-gnu
musl: false
- task: test
runner: ubuntu-24.04-arm
target: aarch64-unknown-linux-gnu
musl: false
- task: check
runner: ubuntu-24.04-arm
target: aarch64-unknown-linux-musl
musl: true
- task: test
runner: ubuntu-24.04-arm
target: aarch64-unknown-linux-musl
musl: true
runs-on: ${{ matrix.runner }}
timeout-minutes: 25 timeout-minutes: 25
steps: steps:
@@ -137,7 +112,7 @@ jobs:
uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1 uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
with: with:
target: ${{ matrix.target }} target: ${{ matrix.target }}
cache-key: ${{ matrix.task }}-${{ matrix.target }} cache: false
- name: Run ${{ matrix.task }} with all features - name: Run ${{ matrix.task }} with all features
run: | run: |
@@ -174,7 +149,7 @@ jobs:
- name: Install Rust toolchain - name: Install Rust toolchain
uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1 uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
with: with:
cache-key: docs cache: false
- name: Check docs - name: Check docs
run: | run: |
+5 -5
View File
@@ -21,7 +21,7 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- name: Checkout - name: Checkout
uses: actions/checkout@v4 uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
- name: Determine image tags - name: Determine image tags
id: meta id: meta
@@ -42,10 +42,10 @@ jobs:
# (native builds, no emulation), but buildx still needs a driver that # (native builds, no emulation), but buildx still needs a driver that
# can assemble a multi-platform manifest list. # can assemble a multi-platform manifest list.
- name: Set up Docker Buildx - name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3 uses: docker/setup-buildx-action@37fe631027851001ddb9b187196cc803df7f5f0e # v4
- name: Log in to Gitea Container Registry - name: Log in to Gitea Container Registry
uses: docker/login-action@v3 uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4
with: with:
registry: ${{ env.GITEA_REGISTRY }} registry: ${{ env.GITEA_REGISTRY }}
# The automatic per-job token comes back unauthorized for package # The automatic per-job token comes back unauthorized for package
@@ -55,7 +55,7 @@ jobs:
password: ${{ secrets.REGISTRY_TOKEN }} password: ${{ secrets.REGISTRY_TOKEN }}
- name: Log in to GHCR - name: Log in to GHCR
uses: docker/login-action@v3 uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4
with: with:
registry: ${{ env.GHCR_REGISTRY }} registry: ${{ env.GHCR_REGISTRY }}
username: ${{ secrets.GHCR_USERNAME }} username: ${{ secrets.GHCR_USERNAME }}
@@ -76,7 +76,7 @@ jobs:
# below), never a separate build, so the hash matches everywhere. # below), never a separate build, so the hash matches everywhere.
- name: Build and push to Gitea - name: Build and push to Gitea
id: build id: build
uses: docker/build-push-action@v6 uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7
with: with:
context: . context: .
platforms: linux/amd64,linux/arm64 platforms: linux/amd64,linux/arm64
@@ -16,6 +16,9 @@ env:
CARGO_TERM_COLOR: always CARGO_TERM_COLOR: always
# Pinned for reproducibility; bump deliberately alongside `fuzz/Cargo.lock`. # Pinned for reproducibility; bump deliberately alongside `fuzz/Cargo.lock`.
CARGO_FUZZ_VERSION: 0.13.1 CARGO_FUZZ_VERSION: 0.13.1
CARGO_FUZZ_SHA256: 86edc9b8f20c29ff07c3e5fcd67bc3f0784b3b960a5428635060f20d50555461
FUZZ_TOOLCHAIN: nightly-x86_64-unknown-linux-gnu
FUZZ_TARGET: x86_64-unknown-linux-gnu
jobs: jobs:
fuzz-smoke: fuzz-smoke:
@@ -32,22 +35,33 @@ jobs:
- name: Install stable Rust toolchain - name: Install stable Rust toolchain
uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1 uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
with: with:
cache-key: fuzz cache: false
- name: Install nightly Rust toolchain - name: Install nightly Rust toolchain
# Installed alongside, but deliberately not made the directory # Installed alongside, but deliberately not made the directory
# default: cargo-fuzz itself is built with the default stable # default: cargo-fuzz itself is built with the default stable
# toolchain below, and only `fuzz build`/`fuzz run` are run with # toolchain below, and only `fuzz build`/`fuzz run` are run with
# `+nightly`. Building cargo-fuzz's own dependency tree under a # the GNU-host nightly below. Building cargo-fuzz's own dependency
# nightly override can trip build scripts (e.g. rustix) that gate # tree under a nightly override can trip build scripts (e.g. rustix) that gate
# unstable internal attributes on detecting a nightly compiler. # unstable internal attributes on detecting a nightly compiler.
run: rustup toolchain install nightly --profile minimal --no-self-update run: rustup toolchain install "${FUZZ_TOOLCHAIN}" --profile minimal --no-self-update
- name: Install cargo-fuzz - name: Install cargo-fuzz
run: cargo install cargo-fuzz --locked --version "${CARGO_FUZZ_VERSION}" run: |
set -euo pipefail
archive="cargo-fuzz-${CARGO_FUZZ_VERSION}-x86_64-unknown-linux-musl.tar.gz"
curl --fail --show-error --location --retry 3 \
"https://github.com/rust-fuzz/cargo-fuzz/releases/download/${CARGO_FUZZ_VERSION}/${archive}" \
--output "$archive"
echo "${CARGO_FUZZ_SHA256} ${archive}" | sha256sum --check --status -
tar -xzf "$archive" cargo-fuzz
install -Dm755 cargo-fuzz "${RUNNER_TEMP}/bin/cargo-fuzz"
echo "${RUNNER_TEMP}/bin" >> "$GITHUB_PATH"
- name: Build all fuzz targets - name: Build all fuzz targets
run: cargo +nightly fuzz build # libFuzzer sanitizers require dynamically linked GNU binaries; the
# runner's default Rust host can be static musl.
run: RUSTUP_TOOLCHAIN="${FUZZ_TOOLCHAIN}" cargo fuzz build --target "${FUZZ_TARGET}"
working-directory: fuzz working-directory: fuzz
- name: Run bounded fuzz smoke for every target - name: Run bounded fuzz smoke for every target
@@ -59,7 +73,7 @@ jobs:
set -euo pipefail set -euo pipefail
for target in decode_request decode_open_reply decode_echo_reply decode_params decode_varint; do for target in decode_request decode_open_reply decode_echo_reply decode_params decode_varint; do
echo "::group::fuzz smoke: ${target}" echo "::group::fuzz smoke: ${target}"
cargo +nightly fuzz run "${target}" -- \ RUSTUP_TOOLCHAIN="${FUZZ_TOOLCHAIN}" cargo fuzz run --target "${FUZZ_TARGET}" "${target}" -- \
-max_total_time=10 \ -max_total_time=10 \
-timeout=5 -timeout=5
echo "::endgroup::" echo "::endgroup::"
@@ -32,13 +32,14 @@ jobs:
- name: Install Rust toolchain - name: Install Rust toolchain
uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1 uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
with: with:
cache-key: interop-netem cache: false
- name: Check host tools - name: Check host tools and network administration access
run: | run: |
ip -V ip -V
tc -V tc -V
go version go version
sudo -n true
- name: Install upstream irtt - name: Install upstream irtt
run: | run: |
@@ -62,14 +63,14 @@ jobs:
run: | run: |
latest_summary="$(find target/interop-netem -mindepth 2 -maxdepth 2 -name github-step-summary.md -type f -print 2>/dev/null | sort | tail -n 1)" latest_summary="$(find target/interop-netem -mindepth 2 -maxdepth 2 -name github-step-summary.md -type f -print 2>/dev/null | sort | tail -n 1)"
if [ -n "$latest_summary" ]; then if [ -n "$latest_summary" ]; then
cat "$latest_summary" >> "$GITHUB_STEP_SUMMARY" cat "$latest_summary" >> "$GITEA_STEP_SUMMARY"
else else
echo "No netem interop github-step-summary.md found." >> "$GITHUB_STEP_SUMMARY" echo "No netem interop github-step-summary.md found." >> "$GITEA_STEP_SUMMARY"
fi fi
- name: Upload netem interop artifacts - name: Upload netem interop artifacts
if: always() if: always()
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 uses: https://gitea.com/actions/gitea-upload-artifact@v7
with: with:
name: interop-netem name: interop-netem
path: target/interop-netem path: target/interop-netem
+52
View File
@@ -0,0 +1,52 @@
name: Semver
on:
workflow_dispatch:
push:
tags:
- "irtt-proto/v*.*.*"
- "irtt-client/v*.*.*"
- "irtt-server/v*.*.*"
- "irtt-stats/v*.*.*"
- "irtt-rs/v*.*.*"
permissions:
contents: read
env:
CARGO_SEMVER_CHECKS_VERSION: 0.50.0
CARGO_SEMVER_CHECKS_SHA256: 52a65dc88dc53fa8b57d6087954eb52cda149ca03bcfca78ce3fdecd23f893c4
jobs:
semver:
name: cargo-semver-checks
runs-on: ubuntu-latest
timeout-minutes: 20
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with:
persist-credentials: false
- name: Install Rust toolchain
uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
with:
cache: false
# The CLI preserves the previous wrapper's `feature-group: all-features`
# behavior without its GitHub API/cache dependency.
- name: Install cargo-semver-checks
run: |
set -euo pipefail
archive="cargo-semver-checks-x86_64-unknown-linux-gnu.tar.gz"
curl --fail --show-error --location --retry 3 \
"https://github.com/obi1kenobi/cargo-semver-checks/releases/download/v${CARGO_SEMVER_CHECKS_VERSION}/${archive}" \
--output "$archive"
echo "${CARGO_SEMVER_CHECKS_SHA256} ${archive}" | sha256sum --check --status -
tar -xzf "$archive" cargo-semver-checks
install -Dm755 cargo-semver-checks "${RUNNER_TEMP}/bin/cargo-semver-checks"
echo "${RUNNER_TEMP}/bin" >> "$GITHUB_PATH"
- name: Check semver compatibility
run: cargo semver-checks check-release --all-features
+109
View File
@@ -0,0 +1,109 @@
name: Supply chain
on:
push:
branches:
- main
pull_request:
branches:
- main
schedule:
- cron: "41 2 * * 1"
workflow_dispatch:
permissions:
contents: read
env:
ACTIONLINT_VERSION: 1.7.7
CARGO_DENY_VERSION: 0.20.2
CARGO_DENY_SHA256: 9f12ed4c49936e09b48bf862b595cde2fe64fcbd9d74dfacac6131ca824c8d5f
ACTIONLINT_SHA256: 023070a287cd8cccd71515fedc843f1985bf96c436b7effaecce67290e7e0757
jobs:
cargo-deny:
name: cargo-deny ${{ matrix.checks }}
runs-on: ubuntu-latest
timeout-minutes: 15
strategy:
fail-fast: false
matrix:
include:
# Advisories can appear after code was merged. Keep this visible,
# but don't let a surprise upstream advisory nuke every unrelated PR.
- checks: advisories
continue-on-error: true
# These are policy choices in this repo. They should block PRs.
- checks: bans licenses sources
continue-on-error: false
continue-on-error: ${{ matrix.continue-on-error }}
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with:
persist-credentials: false
- name: Install cargo-deny
run: |
set -euo pipefail
archive="cargo-deny-${CARGO_DENY_VERSION}-x86_64-unknown-linux-musl.tar.gz"
curl --fail --show-error --location --retry 3 \
"https://github.com/EmbarkStudios/cargo-deny/releases/download/${CARGO_DENY_VERSION}/${archive}" \
--output "$archive"
echo "${CARGO_DENY_SHA256} ${archive}" | sha256sum --check --status -
tar -xzf "$archive" --strip-components=1 \
"cargo-deny-${CARGO_DENY_VERSION}-x86_64-unknown-linux-musl/cargo-deny"
install -Dm755 cargo-deny "${RUNNER_TEMP}/bin/cargo-deny"
echo "${RUNNER_TEMP}/bin" >> "$GITHUB_PATH"
- name: Run cargo-deny
run: >-
cargo deny --manifest-path Cargo.toml --all-features check
${{ matrix.checks }}
actionlint:
name: actionlint
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with:
persist-credentials: false
- name: Install actionlint
run: |
set -euo pipefail
archive="actionlint_${ACTIONLINT_VERSION}_linux_amd64.tar.gz"
curl --fail --show-error --location --retry 3 \
"https://github.com/rhysd/actionlint/releases/download/v${ACTIONLINT_VERSION}/${archive}" \
--output "$archive"
echo "${ACTIONLINT_SHA256} ${archive}" | sha256sum --check --status -
tar -xzf "$archive" actionlint
install -Dm755 actionlint "${RUNNER_TEMP}/bin/actionlint"
echo "${RUNNER_TEMP}/bin" >> "$GITHUB_PATH"
- name: actionlint handwritten workflows
# actionlint validates GitHub Actions syntax. It cannot parse Gitea's
# required absolute action URL in interop-netem.yml, so that workflow
# is checked by the Gitea runner instead.
run: |
actionlint \
.gitea/workflows/ci.yml \
.gitea/workflows/docker-publish.yml \
.gitea/workflows/fuzz.yml \
.gitea/workflows/semver.yml \
.gitea/workflows/supply-chain.yml \
.github/workflows/actions-security.yml \
.github/workflows/ci-arm.yml \
.github/workflows/osv-scanner.yml \
.github/workflows/rust-clippy.yml
- name: actionlint generated release workflow
run: >-
actionlint -shellcheck= .github/workflows/irtt-rs-release.yml
+7 -29
View File
@@ -16,34 +16,6 @@ permissions:
contents: read contents: read
jobs: jobs:
actionlint:
name: actionlint
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with:
persist-credentials: false
- name: actionlint handwritten workflows
uses: raven-actions/actionlint@205b530c5d9fa8f44ae9ed59f341a0db994aa6f8
with:
files: >-
.github/workflows/actions-security.yml,
.github/workflows/ci.yml,
.github/workflows/interop-netem.yml,
.github/workflows/osv-scanner.yml,
.github/workflows/rust-clippy.yml,
.github/workflows/semver.yml,
.github/workflows/supply-chain.yml
- name: actionlint generated release workflow
uses: raven-actions/actionlint@205b530c5d9fa8f44ae9ed59f341a0db994aa6f8
with:
files: .github/workflows/irtt-rs-release.yml
shellcheck: false
zizmor: zizmor:
name: zizmor name: zizmor
runs-on: ubuntu-latest runs-on: ubuntu-latest
@@ -60,7 +32,13 @@ jobs:
with: with:
persist-credentials: false persist-credentials: false
# Zizmor's SARIF integration publishes GitHub Code Scanning results.
# It remains here while portable actionlint runs with Gitea CI, but scans
# both workflow directories through its GitHub Actions rule set.
- name: Run zizmor - name: Run zizmor
uses: zizmorcore/zizmor-action@3dc1ecc9bcb9e94e9b2c709687979e1298497054 # v0.6.2 uses: zizmorcore/zizmor-action@cc914d7f3750a2d13d75c7f184a1060aa0e9d482 # v0.6.4
with: with:
persona: regular persona: regular
inputs: |
.github/workflows
.gitea/workflows
+80
View File
@@ -0,0 +1,80 @@
name: Rust ARM CI
on:
push:
branches:
- main
pull_request:
branches:
- main
merge_group:
workflow_dispatch:
permissions:
contents: read
env:
CARGO_TERM_COLOR: always
jobs:
verify:
name: cargo ${{ matrix.task }} (${{ matrix.target }})
# Gitea currently has only amd64 Linux runners; retain ARM coverage here.
strategy:
fail-fast: false
matrix:
include:
- task: check
target: aarch64-unknown-linux-gnu
musl: false
- task: test
target: aarch64-unknown-linux-gnu
musl: false
- task: check
target: aarch64-unknown-linux-musl
musl: true
- task: test
target: aarch64-unknown-linux-musl
musl: true
runs-on: ubuntu-24.04-arm
timeout-minutes: 25
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with:
persist-credentials: false
- name: Install musl toolchain
if: matrix.musl
run: |
sudo apt-get update
sudo apt-get install --no-install-recommends musl-tools
- name: Install Rust toolchain
uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
with:
target: ${{ matrix.target }}
cache-key: ${{ matrix.task }}-${{ matrix.target }}
- name: Run ${{ matrix.task }} with all features
run: |
cargo ${{ matrix.task }} \
--workspace \
--all-targets \
--all-features \
--target ${{ matrix.target }} \
--locked
- name: Run ${{ matrix.task }} without default features
run: |
cargo ${{ matrix.task }} \
--workspace \
--all-targets \
--no-default-features \
--target ${{ matrix.target }} \
--locked
+2 -2
View File
@@ -31,7 +31,7 @@ permissions:
jobs: jobs:
scan-scheduled: scan-scheduled:
if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }} if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }}
uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@8deb546fdb875b9996d27d4950be7312dac076a1" # v2.5.0 uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@a345acffa64b0eaede81a3d9aae6141214d9c8fc" # v2.6.0
with: with:
# Example of specifying custom arguments # Example of specifying custom arguments
scan-args: |- scan-args: |-
@@ -40,7 +40,7 @@ jobs:
./ ./
scan-pr: scan-pr:
if: ${{ github.event_name == 'pull_request' || github.event_name == 'merge_group' }} if: ${{ github.event_name == 'pull_request' || github.event_name == 'merge_group' }}
uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@8deb546fdb875b9996d27d4950be7312dac076a1" # v2.5.0 uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@a345acffa64b0eaede81a3d9aae6141214d9c8fc" # v2.6.0
with: with:
# Example of specifying custom arguments # Example of specifying custom arguments
scan-args: |- scan-args: |-
+1 -1
View File
@@ -60,7 +60,7 @@ jobs:
exit 0 exit 0
- name: Upload Clippy SARIF - name: Upload Clippy SARIF
uses: github/codeql-action/upload-sarif@ff2f1c621b7f889edc0d3c761ac2e6a3f8cdb0dd # v4 uses: github/codeql-action/upload-sarif@b96794f015dfd88f77b49b1c93e0fa7110f94c63 # v4
if: always() if: always()
with: with:
sarif_file: rust-clippy-results.sarif sarif_file: rust-clippy-results.sarif
-31
View File
@@ -1,31 +0,0 @@
name: Semver
on:
workflow_dispatch:
push:
tags:
- "irtt-proto/v*.*.*"
- "irtt-client/v*.*.*"
- "irtt-server/v*.*.*"
- "irtt-stats/v*.*.*"
- "irtt-rs/v*.*.*"
permissions:
contents: read
jobs:
semver:
name: cargo-semver-checks
runs-on: ubuntu-latest
timeout-minutes: 20
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with:
persist-credentials: false
- name: Check semver compatibility
uses: obi1kenobi/cargo-semver-checks-action@6b69fcf40e9b5fb17adeb57e4b6ecd020649a239 # v2
with:
feature-group: all-features
-79
View File
@@ -1,79 +0,0 @@
name: Supply chain
on:
push:
branches:
- main
pull_request:
branches:
- main
merge_group:
schedule:
- cron: "41 2 * * 1"
workflow_dispatch:
permissions:
contents: read
jobs:
cargo-deny:
name: cargo-deny ${{ matrix.checks }}
runs-on: ubuntu-latest
timeout-minutes: 15
strategy:
fail-fast: false
matrix:
include:
# Advisories can appear after code was merged. Keep this visible,
# but don't let a surprise upstream advisory nuke every unrelated PR.
- checks: advisories
continue-on-error: true
# These are policy choices in this repo. They should block PRs.
- checks: bans licenses sources
continue-on-error: false
continue-on-error: ${{ matrix.continue-on-error }}
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with:
persist-credentials: false
- name: Run cargo-deny
uses: EmbarkStudios/cargo-deny-action@3c6349835b2b7b196a839186cb8b78e02f7b5f25 # v2
with:
command: check
arguments: --all-features
command-arguments: ${{ matrix.checks }}
actionlint:
name: actionlint
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- name: Checkout
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with:
persist-credentials: false
- name: actionlint handwritten workflows
uses: raven-actions/actionlint@205b530c5d9fa8f44ae9ed59f341a0db994aa6f8
with:
files: >-
.github/workflows/actions-security.yml,
.github/workflows/ci.yml,
.github/workflows/interop-netem.yml,
.github/workflows/osv-scanner.yml,
.github/workflows/rust-clippy.yml,
.github/workflows/semver.yml,
.github/workflows/supply-chain.yml
- name: actionlint generated release workflow
uses: raven-actions/actionlint@205b530c5d9fa8f44ae9ed59f341a0db994aa6f8
with:
files: .github/workflows/irtt-rs-release.yml
shellcheck: false
+17
View File
@@ -28,6 +28,12 @@ release. The format is loosely based on
## irtt-client ## irtt-client
### 0.5.1
#### Added
- `ManagedCommandReceipt::blocking_wait()` lets synchronous callers wait for a managed target-set update to be applied or rejected without entering an async runtime.
### 0.5.0 ### 0.5.0
#### Added #### Added
@@ -102,6 +108,17 @@ release. The format is loosely based on
## irtt-rs ## irtt-rs
### 0.7.0
#### Added
- Client target arguments now support per-target HMAC configuration: append `@hmac=KEY` to override the global `--hmac` value for that target, or use `@hmac=` to explicitly disable HMAC for it. The target syntax supports context-aware escaping for literal delimiters.
- Continuous client runs can now take complete desired target sets from standard input with `--targets-stdin`. Each non-empty record atomically replaces the current set, `[]` selects an empty set, and EOF requests a graceful stop. The controller retains only the newest unapplied record under transient replacement backpressure, with explicit bounds on record size, desired targets, and live target generations.
#### Changed
- A peer-initiated close is now target-local in stdin-controlled continuous runs: the controller and other desired targets keep running, and a later desired set can start that target again.
### 0.6.2 ### 0.6.2
#### Added #### Added
Generated
+28 -28
View File
@@ -242,7 +242,7 @@ dependencies = [
"heck", "heck",
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 3.0.3", "syn 3.0.4",
] ]
[[package]] [[package]]
@@ -385,9 +385,9 @@ dependencies = [
[[package]] [[package]]
name = "darling" name = "darling"
version = "0.24.0" version = "0.24.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "88490bf1b990d87eaaa7ac8aa887f629a08e7359765b4911faf63c3763347d23" checksum = "ed17f5901b6630b993ca003def43f2f8ef4014fc13b047b57aad617ff32bc2ec"
dependencies = [ dependencies = [
"darling_core", "darling_core",
"darling_macro", "darling_macro",
@@ -395,26 +395,26 @@ dependencies = [
[[package]] [[package]]
name = "darling_core" name = "darling_core"
version = "0.24.0" version = "0.24.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "084e274f91c482280130e1e34e0b8d6e66776a060d7b6de7b84289ca778868c4" checksum = "6837e2cf7485aaae18f86181d2f0e9a7ed297a025e220aeabf63fdebd3a2ddff"
dependencies = [ dependencies = [
"ident_case", "ident_case",
"proc-macro2", "proc-macro2",
"quote", "quote",
"strsim", "strsim",
"syn 3.0.3", "syn 3.0.4",
] ]
[[package]] [[package]]
name = "darling_macro" name = "darling_macro"
version = "0.24.0" version = "0.24.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "68f5792fa0d41cd2325ce0ffa64f0a340eaebd4971a3a0c5e1ffd2cc488a355e" checksum = "2ac7135c3ef02b2f7833bbeb1be5ba7f966dcde8a87c6b87f65a778d71a02785"
dependencies = [ dependencies = [
"darling_core", "darling_core",
"quote", "quote",
"syn 3.0.3", "syn 3.0.4",
] ]
[[package]] [[package]]
@@ -496,9 +496,9 @@ dependencies = [
[[package]] [[package]]
name = "either" name = "either"
version = "1.17.0" version = "1.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34"
[[package]] [[package]]
name = "equivalent" name = "equivalent"
@@ -710,12 +710,12 @@ dependencies = [
"indoc", "indoc",
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 3.0.3", "syn 3.0.4",
] ]
[[package]] [[package]]
name = "irtt-client" name = "irtt-client"
version = "0.5.0" version = "0.5.1"
dependencies = [ dependencies = [
"irtt-proto", "irtt-proto",
"irtt-server", "irtt-server",
@@ -749,7 +749,7 @@ dependencies = [
[[package]] [[package]]
name = "irtt-rs" name = "irtt-rs"
version = "0.6.2" version = "0.7.0"
dependencies = [ dependencies = [
"clap", "clap",
"crossterm", "crossterm",
@@ -881,15 +881,15 @@ dependencies = [
[[package]] [[package]]
name = "log" name = "log"
version = "0.4.33" version = "0.4.34"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
[[package]] [[package]]
name = "lru" name = "lru"
version = "0.18.2" version = "0.18.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d2f2f9b4ba7e6b24d95e7e899329d35be83bcded72c8540cdd5368932d1d90a" checksum = "0d317b4b9eb398e6acce275758ec6125535505e7a146fb1a9b8bda2451b0ff4c"
dependencies = [ dependencies = [
"hashbrown 0.17.1", "hashbrown 0.17.1",
] ]
@@ -1185,7 +1185,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d" checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d"
dependencies = [ dependencies = [
"phf_shared", "phf_shared",
"rand 0.8.7", "rand 0.8.8",
] ]
[[package]] [[package]]
@@ -1294,9 +1294,9 @@ checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
[[package]] [[package]]
name = "rand" name = "rand"
version = "0.8.7" version = "0.8.8"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" checksum = "e058c7de0b26af77780c769414d6257830bb240f3c38477dbc2c16e5f54d6d4c"
dependencies = [ dependencies = [
"rand_core 0.6.4", "rand_core 0.6.4",
] ]
@@ -1569,7 +1569,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 3.0.3", "syn 3.0.4",
] ]
[[package]] [[package]]
@@ -1705,9 +1705,9 @@ dependencies = [
[[package]] [[package]]
name = "syn" name = "syn"
version = "3.0.3" version = "3.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" checksum = "e6275cddf4610d1775e6d1fe9469b2e77d0f39fd98fb7450901b821e0c53649f"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@@ -1840,7 +1840,7 @@ checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 3.0.3", "syn 3.0.4",
] ]
[[package]] [[package]]
@@ -1886,7 +1886,7 @@ checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 3.0.3", "syn 3.0.4",
] ]
[[package]] [[package]]
@@ -1944,9 +1944,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
[[package]] [[package]]
name = "uuid" name = "uuid"
version = "1.24.1" version = "1.26.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2cefc03fd367c0c6d4305de1b312cf00248c4114f4a0418ce6a6af769e3b0bd9" checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812"
dependencies = [ dependencies = [
"atomic", "atomic",
"getrandom 0.4.3", "getrandom 0.4.3",
+1 -1
View File
@@ -18,7 +18,7 @@ homepage = "https://github.com/Lochnair/irtt-rs"
[workspace.dependencies] [workspace.dependencies]
irtt-proto = { version = "0.5.0", path = "crates/irtt-proto" } irtt-proto = { version = "0.5.0", path = "crates/irtt-proto" }
irtt-client = { version = "0.5.0", path = "crates/irtt-client" } irtt-client = { version = "0.5.1", path = "crates/irtt-client" }
irtt-server = { version = "0.5.2", path = "crates/irtt-server" } irtt-server = { version = "0.5.2", path = "crates/irtt-server" }
irtt-stats = { version = "0.5.0", path = "crates/irtt-stats" } irtt-stats = { version = "0.5.0", path = "crates/irtt-stats" }
+1 -1
View File
@@ -10,7 +10,7 @@
FROM --platform=$BUILDPLATFORM tonistiigi/xx AS xx FROM --platform=$BUILDPLATFORM tonistiigi/xx AS xx
FROM --platform=$BUILDPLATFORM rust:1.88-alpine AS builder FROM --platform=$BUILDPLATFORM rust:1.98-alpine AS builder
COPY --from=xx / / COPY --from=xx / /
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "irtt-rs" name = "irtt-rs"
version = "0.6.2" version = "0.7.0"
description = "IRTT-compatible client, TUI, and server applications" description = "IRTT-compatible client, TUI, and server applications"
edition.workspace = true edition.workspace = true
license.workspace = true license.workspace = true
+33 -7
View File
@@ -11,15 +11,22 @@ pub const DEFAULT_CLIENT_DURATION: Duration = Duration::from_secs(10);
#[derive(Debug, Clone, Parser)] #[derive(Debug, Clone, Parser)]
#[command(name = "irtt-client", about = "Minimal IRTT-compatible stream client")] #[command(name = "irtt-client", about = "Minimal IRTT-compatible stream client")]
pub struct ClientArgs { pub struct ClientArgs {
/// Server address/host, optionally prefixed with LABEL=. Repeat for multi-target mode. /// Server address/host, optionally prefixed with LABEL= and suffixed with @hmac=KEY. Repeat for multi-target mode.
#[arg( #[arg(
value_name = "TARGET", value_name = "TARGET",
num_args = 0.., num_args = 0..,
value_parser = parse_target, value_parser = parse_target,
long_help = "Server address/host, optionally prefixed with LABEL=. Repeat for multi-target mode.\n\nExamples:\n irtt-client host-a:2112 host-b:2112\n irtt-client eu=host-a:2112 us=host-b:2112\n irtt-client host-a:2112 eu=host-b:2112" long_help = "Server address/host, optionally prefixed with LABEL= and suffixed with @hmac=KEY. Repeat for multi-target mode.\n\nExamples:\n irtt-client host-a:2112 host-b:2112\n irtt-client eu=host-a:2112 us=host-b:2112\n irtt-client ams=host-a:2112@hmac=secret public=host-b:2112@hmac=\n\nA target without @hmac= inherits --hmac. @hmac= explicitly disables it for that target."
)] )]
pub targets: Vec<TargetArg>, pub targets: Vec<TargetArg>,
/// Read complete desired target sets from standard input in continuous mode.
#[arg(
long,
long_help = "Read one complete desired target set per stdin line in continuous mode. Commas separate targets and backslash escapes commas within a target. [] selects an empty desired set. Records are limited to 64 KiB. EOF stops the client gracefully and may discard a revision that has not yet been acknowledged as applied."
)]
pub targets_stdin: bool,
/// Managed group pacing for multi-target mode. /// Managed group pacing for multi-target mode.
#[arg( #[arg(
long, long,
@@ -34,7 +41,7 @@ pub struct ClientArgs {
default_value = "10s", default_value = "10s",
value_parser = parse_test_duration, value_parser = parse_test_duration,
help = "Test duration; use 0 for continuous mode", help = "Test duration; use 0 for continuous mode",
long_help = "Test duration; use 0 for continuous mode.\n\nFinite runs retain exact statistics for final summaries. Continuous mode uses bounded-memory running statistics and prints a final summary only when interrupted." long_help = "Test duration; use 0 for continuous mode.\n\nFinite runs retain exact statistics for final summaries. Continuous mode uses bounded-memory running statistics and prints a final summary when interrupted; --targets-stdin also prints the final retained summary on EOF."
)] )]
pub duration: Duration, pub duration: Duration,
@@ -86,12 +93,16 @@ impl ClientArgs {
/// Targets are only required here. `--list-columns` returns before this is /// Targets are only required here. `--list-columns` returns before this is
/// called, so listing columns still needs no target. /// called, so listing columns still needs no target.
pub fn prepare(&self) -> Result<ManagedRunSetup, String> { pub fn prepare(&self) -> Result<ManagedRunSetup, String> {
if self.targets_stdin && !self.is_continuous() {
return Err("--targets-stdin requires --duration 0".to_owned());
}
prepare_managed_run( prepare_managed_run(
&self.common, &self.common,
self.duration, self.duration,
TargetSelection { TargetSelection {
targets: &self.targets, targets: &self.targets,
pacing: self.pacing, pacing: self.pacing,
stdin_controlled: self.targets_stdin,
}, },
) )
} }
@@ -159,14 +170,14 @@ mod tests {
let args = parse(&["127.0.0.1:2112"]).unwrap(); let args = parse(&["127.0.0.1:2112"]).unwrap();
assert_eq!(args.targets.len(), 1); assert_eq!(args.targets.len(), 1);
assert_eq!(args.targets[0].label, None); assert_eq!(args.prepare().unwrap().targets[0].label, "127.0.0.1:2112");
assert_eq!(args.targets[0].addr, "127.0.0.1:2112");
assert_eq!(args.pacing, GroupPacingArg::Staggered); assert_eq!(args.pacing, GroupPacingArg::Staggered);
assert_eq!(args.duration, DEFAULT_CLIENT_DURATION); assert_eq!(args.duration, DEFAULT_CLIENT_DURATION);
assert_eq!(args.interval, Duration::from_secs(1)); assert_eq!(args.interval, Duration::from_secs(1));
assert_eq!(args.format, OutputFormat::Table); assert_eq!(args.format, OutputFormat::Table);
assert_eq!(args.columns, None); assert_eq!(args.columns, None);
assert_eq!(args.header, HeaderMode::Auto); assert_eq!(args.header, HeaderMode::Auto);
assert!(!args.targets_stdin);
assert!(!args.is_continuous()); assert!(!args.is_continuous());
assert_eq!(args.timestamp_mode(), TimestampArg::Both); assert_eq!(args.timestamp_mode(), TimestampArg::Both);
@@ -254,8 +265,8 @@ mod tests {
#[test] #[test]
fn invalid_labelled_target_syntax_is_rejected() { fn invalid_labelled_target_syntax_is_rejected() {
assert!(parse(&["=127.0.0.1:2112"]).is_err()); assert!(parse(&["=127.0.0.1:2112"]).unwrap().prepare().is_err());
assert!(parse(&["label="]).is_err()); assert!(parse(&["label="]).unwrap().prepare().is_err());
} }
#[test] #[test]
@@ -309,6 +320,21 @@ mod tests {
assert_eq!(config.hmac_key, Some(b"secret".to_vec())); assert_eq!(config.hmac_key, Some(b"secret".to_vec()));
} }
#[test]
fn targets_stdin_requires_continuous_mode_and_allows_empty_initial_set() {
assert!(parse(&["--targets-stdin"]).unwrap().prepare().is_err());
let args = parse(&["--duration", "0", "--targets-stdin"]).unwrap();
let setup = args.prepare().unwrap();
assert!(setup.targets.is_empty());
assert!(setup.stdin_controlled);
let config = setup.managed_config();
assert_eq!(
config.completion,
irtt_client::managed::ManagedCompletionPolicy::ExplicitStop
);
assert_eq!(config.max_live_target_generations, 256);
}
#[test] #[test]
fn output_format_summary_policy_is_table_only() { fn output_format_summary_policy_is_table_only() {
assert!(OutputFormat::Table.prints_summary()); assert!(OutputFormat::Table.prints_summary());
+778 -10
View File
@@ -1,16 +1,16 @@
use std::collections::BTreeMap;
use std::{ use std::{
collections::HashSet, collections::{BTreeMap, BTreeSet, HashSet, VecDeque},
io::{self, Write}, io::{self, BufRead, Write},
sync::atomic::AtomicBool, sync::{atomic::AtomicBool, Arc, Mutex},
thread, thread,
time::Duration, time::Duration,
}; };
use irtt_client::{ use irtt_client::{
managed::{ managed::{
BlockingManagedClient, ManagedEndReason, ManagedEvent, ManagedEventSubscription, BlockingManagedClient, ManagedClientHandle, ManagedCommandApplyError, ManagedEndReason,
ManagedEventTryRecvError, ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance, ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError, ManagedStatus,
ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance,
}, },
ClientEvent, ClientEvent,
}; };
@@ -20,13 +20,13 @@ use super::{
output::{EventRenderStats, OutputConfig}, output::{EventRenderStats, OutputConfig},
}; };
use crate::shared::client::expected_probe_count;
use crate::shared::client::{ use crate::shared::client::{
is_shutdown_requested, expected_probe_count, is_shutdown_requested, parse_stdin_target_set,
session::{ session::{
drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close, drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close,
request_managed_stop_once, should_print_final_summary, ManagedDrainState, request_managed_stop_once, should_print_final_summary, ManagedDrainState,
}, },
ManagedRunSetup, STDIN_MAX_DESIRED_TARGETS, STDIN_OUTCOME_HISTORY_LIMIT,
}; };
use irtt_stats::{StatsCollector, StatsConfig}; use irtt_stats::{StatsCollector, StatsConfig};
@@ -41,6 +41,7 @@ const FINITE_STATS_MEMORY_STRONG_WARNING_BYTES: u64 = 512 * MIB;
const FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES: u64 = GIB; const FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES: u64 = GIB;
const MANAGED_EVENT_WAIT_SLICE: Duration = Duration::from_millis(20); const MANAGED_EVENT_WAIT_SLICE: Duration = Duration::from_millis(20);
const MAX_STDIN_RECORD_BYTES: usize = 64 * 1024;
pub fn run_stream( pub fn run_stream(
args: ClientArgs, args: ClientArgs,
@@ -61,6 +62,9 @@ pub fn run_stream(
args.verbose, args.verbose,
) )
.map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
if setup.stdin_controlled {
return run_stdin_stream(setup, output_config, shutdown_requested);
}
let continuous = args.is_continuous(); let continuous = args.is_continuous();
let target_count = setup.target_count(); let target_count = setup.target_count();
@@ -196,6 +200,561 @@ pub fn run_stream(
Ok(()) Ok(())
} }
#[derive(Clone)]
struct StdinTargetSet {
revision: u64,
targets: Vec<ManagedTargetConfig>,
}
#[derive(Default)]
struct StdinMailboxState {
next_revision: u64,
latest: Option<StdinTargetSet>,
eof: bool,
fatal: Option<String>,
}
#[derive(Default)]
struct StdinMailbox {
state: Mutex<StdinMailboxState>,
}
impl StdinMailbox {
fn publish(&self, targets: Vec<ManagedTargetConfig>) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.next_revision = state.next_revision.saturating_add(1);
state.latest = Some(StdinTargetSet {
revision: state.next_revision,
targets,
});
}
fn finish(&self) {
self.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.eof = true;
}
fn fail(&self, error: String) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.fatal = Some(error);
state.latest = None;
}
fn latest(&self) -> Option<StdinTargetSet> {
self.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.latest
.clone()
}
fn clear_if_current(&self, revision: u64) {
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if state
.latest
.as_ref()
.is_some_and(|set| set.revision == revision)
{
state.latest = None;
}
}
fn is_current(&self, revision: u64) -> bool {
self.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.latest
.as_ref()
.is_some_and(|set| set.revision == revision)
}
fn fatal(&self) -> Option<String> {
self.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.fatal
.clone()
}
fn is_finished(&self) -> bool {
self.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.eof
}
}
fn read_stdin_record<R: BufRead>(
reader: &mut R,
buffer: &mut Vec<u8>,
) -> io::Result<Option<String>> {
buffer.clear();
loop {
let available = reader.fill_buf()?;
if available.is_empty() {
if buffer.is_empty() {
return Ok(None);
}
if buffer.len() > MAX_STDIN_RECORD_BYTES {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"stdin target record exceeds the maximum size",
));
}
return std::str::from_utf8(buffer)
.map(str::to_owned)
.map(Some)
.map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "stdin is not UTF-8"));
}
if buffer.len() == MAX_STDIN_RECORD_BYTES + 1 {
if available.first() == Some(&b'\n') && buffer.last() == Some(&b'\r') {
reader.consume(1);
buffer.pop();
return std::str::from_utf8(buffer)
.map(str::to_owned)
.map(Some)
.map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "stdin is not UTF-8"));
}
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"stdin target record exceeds the maximum size",
));
}
if let Some(newline) = available.iter().position(|byte| *byte == b'\n') {
let before_newline = &available[..newline];
if buffer.len().saturating_add(before_newline.len()) > MAX_STDIN_RECORD_BYTES + 1 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"stdin target record exceeds the maximum size",
));
}
buffer.extend_from_slice(before_newline);
reader.consume(newline + 1);
if buffer.last() == Some(&b'\r') {
buffer.pop();
}
if buffer.len() > MAX_STDIN_RECORD_BYTES {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"stdin target record exceeds the maximum size",
));
}
return std::str::from_utf8(buffer)
.map(str::to_owned)
.map(Some)
.map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "stdin is not UTF-8"));
}
let remaining = MAX_STDIN_RECORD_BYTES + 1 - buffer.len();
let take = available.len().min(remaining);
buffer.extend_from_slice(&available[..take]);
reader.consume(take);
if buffer.len() > MAX_STDIN_RECORD_BYTES && buffer.last() != Some(&b'\r') {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"stdin target record exceeds the maximum size",
));
}
}
}
fn read_stdin_target_sets<R: BufRead>(reader: &mut R, mailbox: &StdinMailbox) {
let mut line = 0_u64;
let mut record = Vec::with_capacity(MAX_STDIN_RECORD_BYTES + 1);
loop {
match read_stdin_record(reader, &mut record) {
Ok(Some(record)) => {
line = line.saturating_add(1);
if record.is_empty() {
continue;
}
match parse_stdin_target_set(&record, STDIN_MAX_DESIRED_TARGETS) {
Ok(targets) => {
mailbox.publish(targets.into_iter().map(|target| target.managed).collect())
}
Err(error) => {
mailbox.fail(format!("invalid --targets-stdin line {line}: {error}"));
return;
}
}
}
Ok(None) => {
mailbox.finish();
return;
}
Err(_) => {
mailbox.fail(format!("failed to read --targets-stdin line {}", line + 1));
return;
}
}
}
}
fn spawn_stdin_target_reader(mailbox: Arc<StdinMailbox>) -> io::Result<()> {
thread::Builder::new()
.name("irtt-targets-stdin".to_owned())
.spawn(move || {
let stdin = io::stdin();
read_stdin_target_sets(&mut stdin.lock(), &mailbox);
})
.map(|_| ())
}
#[derive(Clone)]
struct RetryAfterStatus {
revision: u64,
status: Arc<ManagedStatus>,
}
fn retry_waits_for_status_change(
retry_after: &mut Option<RetryAfterStatus>,
target_set: &StdinTargetSet,
status: &Arc<ManagedStatus>,
) -> bool {
let Some(retry) = retry_after else {
return false;
};
if retry.revision != target_set.revision {
*retry_after = None;
return false;
}
Arc::ptr_eq(&retry.status, status)
}
fn retry_after_capacity_rejection(
mailbox: &StdinMailbox,
target_set: &StdinTargetSet,
status: Arc<ManagedStatus>,
) -> Option<RetryAfterStatus> {
mailbox
.is_current(target_set.revision)
.then_some(RetryAfterStatus {
revision: target_set.revision,
status,
})
}
fn apply_latest_stdin_target_set(
handle: &ManagedClientHandle,
mailbox: &StdinMailbox,
retry_after: &mut Option<RetryAfterStatus>,
) -> Result<(), String> {
let Some(target_set) = mailbox.latest() else {
*retry_after = None;
return Ok(());
};
let status = handle.status();
if retry_waits_for_status_change(retry_after, &target_set, &status) {
return Ok(());
}
let receipt = handle
.update_targets(target_set.targets.clone())
.map_err(|_| "failed to submit --targets-stdin update".to_owned())?;
match receipt.blocking_wait() {
Ok(_) => {
mailbox.clear_if_current(target_set.revision);
*retry_after = None;
Ok(())
}
Err(ManagedCommandApplyError::LiveGenerationLimitExceeded { .. }) => {
*retry_after = retry_after_capacity_rejection(mailbox, &target_set, status);
Ok(())
}
Err(_) => Err("--targets-stdin update was rejected".to_owned()),
}
}
struct BoundedTargetSet {
limit: usize,
order: VecDeque<TargetInstance>,
members: HashSet<TargetInstance>,
}
impl BoundedTargetSet {
fn new(limit: usize) -> Self {
Self {
limit,
order: VecDeque::with_capacity(limit),
members: HashSet::with_capacity(limit),
}
}
fn insert(&mut self, target: TargetInstance) -> bool {
if !self.members.insert(target.clone()) {
return false;
}
if self.order.len() == self.limit {
let evicted = self
.order
.pop_front()
.expect("bounded target order is nonempty");
self.members.remove(&evicted);
}
self.order.push_back(target);
true
}
}
fn process_stdin_event<W: Write>(
event: ManagedEvent,
stream_output: &mut StreamOutput<'_, W>,
stats: &mut BTreeMap<TargetInstance, StatsCollector>,
terminal_targets: &mut BoundedTargetSet,
) -> io::Result<()> {
match event {
ManagedEvent::TargetStateChanged { target, .. } => {
stats
.entry(target)
.or_insert_with(|| StatsCollector::new(stats_config(true)));
}
ManagedEvent::Client { target, event } => {
let collector = stats
.entry(target.clone())
.or_insert_with(|| StatsCollector::new(stats_config(true)));
print_events_with_stats(
stream_output,
std::slice::from_ref(&event),
Some(target.id.as_str()),
collector,
)?;
}
ManagedEvent::TargetFinished { outcome }
if terminal_targets.insert(outcome.target.clone()) =>
{
report_target_failure(&outcome);
}
_ => {}
}
Ok(())
}
fn drain_stdin_events<W: Write>(
events: &mut ManagedEventSubscription,
stream_output: &mut StreamOutput<'_, W>,
stats: &mut BTreeMap<TargetInstance, StatsCollector>,
terminal_targets: &mut BoundedTargetSet,
dropped_events: &mut u64,
) -> io::Result<ManagedDrainState> {
drain_managed_events(events, dropped_events, |event| {
process_stdin_event(event, stream_output, stats, terminal_targets)
})
}
fn drain_final_stdin_events<W: Write>(
events: &mut ManagedEventSubscription,
stream_output: &mut StreamOutput<'_, W>,
stats: &mut BTreeMap<TargetInstance, StatsCollector>,
terminal_targets: &mut BoundedTargetSet,
dropped_events: &mut u64,
) -> io::Result<()> {
loop {
match events.try_recv() {
Ok(event) => process_stdin_event(event, stream_output, stats, terminal_targets)?,
Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break,
Err(ManagedEventTryRecvError::Lagged(count)) => {
*dropped_events = dropped_events.saturating_add(count);
}
}
}
Ok(())
}
fn reconcile_stdin_stats(
stats: &mut BTreeMap<TargetInstance, StatsCollector>,
status: &ManagedStatus,
final_summary_targets: Option<&BTreeSet<TargetInstance>>,
) {
let live = status
.targets
.iter()
.map(|target| target.target.clone())
.collect::<HashSet<_>>();
stats.retain(|target, _| {
live.contains(target)
|| final_summary_targets.is_some_and(|targets| targets.contains(target))
});
}
fn snapshot_stdin_summary_targets(
stats: &mut BTreeMap<TargetInstance, StatsCollector>,
status: &ManagedStatus,
) -> BTreeSet<TargetInstance> {
let targets = status
.targets
.iter()
.map(|target| target.target.clone())
.collect::<BTreeSet<_>>();
for target in &targets {
stats
.entry(target.clone())
.or_insert_with(|| StatsCollector::new(stats_config(true)));
}
targets
}
#[derive(Clone)]
enum StdinStop {
Interrupted,
Eof,
Fatal(String),
}
fn stdin_stop_request(
shutdown_requested: &AtomicBool,
mailbox: &StdinMailbox,
) -> Option<StdinStop> {
if is_shutdown_requested(shutdown_requested) {
Some(StdinStop::Interrupted)
} else if let Some(error) = mailbox.fatal() {
Some(StdinStop::Fatal(error))
} else if mailbox.is_finished() {
Some(StdinStop::Eof)
} else {
None
}
}
fn run_stdin_stream(
setup: ManagedRunSetup,
output_config: OutputConfig,
shutdown_requested: &AtomicBool,
) -> Result<(), Box<dyn std::error::Error>> {
if is_shutdown_requested(shutdown_requested) {
return Ok(());
}
let (owner, mut events) = BlockingManagedClient::start_with_subscription(
setup.managed_config(),
setup.managed_targets(),
)?;
let handle = owner.handle();
let mailbox = Arc::new(StdinMailbox::default());
spawn_stdin_target_reader(Arc::clone(&mailbox))?;
let mut stdout = io::LineWriter::new(io::stdout().lock());
let mut stream_output = StreamOutput {
config: output_config,
header_printed: false,
print_final_summary: false,
show_running_only_summary_note: false,
out: &mut stdout,
};
let mut stats = BTreeMap::new();
let mut terminal_targets = BoundedTargetSet::new(STDIN_OUTCOME_HISTORY_LIMIT);
let mut retry_after = None;
let mut dropped_events = 0_u64;
let mut stop_requested = false;
let mut stop = None;
let mut summary_targets = None;
let mut subscription_closed = false;
loop {
let status = handle.status();
if stop.is_none() {
if let Some(requested_stop) = stdin_stop_request(shutdown_requested, &mailbox) {
summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status));
stop = Some(requested_stop);
if request_managed_stop_once(&mut stop_requested) {
drop(handle.stop());
}
} else if let Err(error) =
apply_latest_stdin_target_set(&handle, &mailbox, &mut retry_after)
{
summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status));
stop = Some(StdinStop::Fatal(error));
if request_managed_stop_once(&mut stop_requested) {
drop(handle.stop());
}
}
}
let drain_state = drain_stdin_events(
&mut events,
&mut stream_output,
&mut stats,
&mut terminal_targets,
&mut dropped_events,
)?;
reconcile_stdin_stats(&mut stats, &handle.status(), summary_targets.as_ref());
match drain_state {
ManagedDrainState::Empty => thread::sleep(MANAGED_EVENT_WAIT_SLICE),
ManagedDrainState::Closed => subscription_closed = true,
ManagedDrainState::BudgetExhausted => {}
}
if handle.status().final_outcome.is_some() || subscription_closed {
break;
}
}
let outcome = owner.join()?;
drain_final_stdin_events(
&mut events,
&mut stream_output,
&mut stats,
&mut terminal_targets,
&mut dropped_events,
)?;
if let Some(warning) = dropped_event_warning(dropped_events) {
eprintln!("{warning}");
}
for target in outcome.recent_target_outcomes.iter() {
if terminal_targets.insert(target.target.clone()) {
report_target_failure(target);
}
}
if outcome.discarded_target_outcomes != 0 {
eprintln!(
"irtt-rs: warning: {} final target outcomes were discarded",
outcome.discarded_target_outcomes
);
}
let print_summary = matches!(stop, Some(StdinStop::Interrupted | StdinStop::Eof));
stream_output.print_final_summary = print_summary;
stream_output.show_running_only_summary_note = print_summary;
if let Some(summary_targets) = summary_targets {
let multi_target = summary_targets.len() > 1;
for target in summary_targets {
let stats = stats
.get(&target)
.expect("summary snapshot retains every selected target collector");
if multi_target && stream_output.config.prints_summary() {
writeln!(stream_output.out)?;
writeln!(
stream_output.out,
"target: {} (generation {})",
target.id, target.generation
)?;
}
stream_output.print_summary(stats)?;
}
}
stream_output.out.flush()?;
match stop {
Some(StdinStop::Fatal(error)) => Err(error.into()),
_ => match outcome.end_reason {
ManagedEndReason::DriverFailed(failure) => {
Err(format!("managed driver failed: {failure}").into())
}
_ => Ok(()),
},
}
}
fn process_event<W: Write>( fn process_event<W: Write>(
event: ManagedEvent, event: ManagedEvent,
stream_output: &mut StreamOutput<'_, W>, stream_output: &mut StreamOutput<'_, W>,
@@ -384,9 +943,218 @@ fn finite_stats_memory_warning(args: &ClientArgs, target_count: usize) -> Option
mod tests { mod tests {
use super::*; use super::*;
use irtt_client::managed::{ use irtt_client::managed::{
ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetFailurePhase, TargetId, ManagedLifecycle, ManagedTargetFailure, ManagedTargetFailureKind,
ManagedTargetFailurePhase, ManagedTargetLifecycle, ManagedTargetStatus, TargetId,
}; };
use std::sync::Arc; use std::{io::BufReader, io::Cursor, sync::Arc};
#[test]
fn stdin_record_reader_removes_only_line_framing() {
let mut reader = BufReader::with_capacity(2, Cursor::new(b" first \r\n\nthird\r"));
let mut buffer = Vec::with_capacity(MAX_STDIN_RECORD_BYTES + 1);
assert_eq!(
read_stdin_record(&mut reader, &mut buffer).unwrap(),
Some(" first ".to_owned())
);
assert_eq!(
read_stdin_record(&mut reader, &mut buffer).unwrap(),
Some(String::new())
);
assert_eq!(
read_stdin_record(&mut reader, &mut buffer).unwrap(),
Some("third\r".to_owned())
);
assert_eq!(read_stdin_record(&mut reader, &mut buffer).unwrap(), None);
}
#[test]
fn stdin_record_reader_rejects_over_limit_unterminated_record() {
let input = vec![b'x'; MAX_STDIN_RECORD_BYTES + 1];
let mut reader = BufReader::with_capacity(17, Cursor::new(input));
let mut buffer = Vec::with_capacity(MAX_STDIN_RECORD_BYTES + 1);
let error = read_stdin_record(&mut reader, &mut buffer).unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::InvalidData);
assert!(error.to_string().contains("maximum size"));
assert!(buffer.len() <= MAX_STDIN_RECORD_BYTES + 1);
}
#[test]
fn newer_mailbox_value_survives_an_earlier_submitted_acknowledgement() {
let mailbox = StdinMailbox::default();
mailbox.publish(vec![ManagedTargetConfig::new("a", "127.0.0.1:2112")]);
let submitted = mailbox.latest().unwrap();
mailbox.publish(vec![ManagedTargetConfig::new("b", "127.0.0.1:2113")]);
let pending = mailbox.latest().unwrap();
mailbox.clear_if_current(submitted.revision);
assert_eq!(mailbox.latest().unwrap().revision, pending.revision);
}
#[test]
fn rejected_update_retries_after_status_publishes_during_acknowledgement() {
let mailbox = StdinMailbox::default();
mailbox.publish(Vec::new());
let target_set = mailbox.latest().unwrap();
let saturated = Arc::new(status_with_targets(&[TargetInstance {
id: TargetId::from("old"),
generation: 1,
}]));
let capacity_freed = Arc::new(status_with_targets(&[]));
// Capacity becomes available after the transaction rejected but before
// its acknowledgement reaches the controller. The rejection path must
// retain the pre-submission status, not this later snapshot.
let mut retry_after =
retry_after_capacity_rejection(&mailbox, &target_set, Arc::clone(&saturated));
assert!(retry_waits_for_status_change(
&mut retry_after,
&target_set,
&saturated
));
assert!(!retry_waits_for_status_change(
&mut retry_after,
&target_set,
&capacity_freed
));
}
#[test]
fn later_desired_set_supersedes_a_capacity_rejected_candidate_before_retry() {
let mailbox = StdinMailbox::default();
mailbox.publish(vec![ManagedTargetConfig::new("a", "127.0.0.1:2112")]);
let applied = mailbox.latest().unwrap();
mailbox.clear_if_current(applied.revision);
mailbox.publish(vec![ManagedTargetConfig::new("b", "127.0.0.1:2113")]);
let rejected = mailbox.latest().unwrap();
let saturated = Arc::new(status_with_targets(&[TargetInstance {
id: TargetId::from("old"),
generation: 1,
}]));
let mut retry_after = retry_after_capacity_rejection(&mailbox, &rejected, saturated);
mailbox.publish(vec![ManagedTargetConfig::new("c", "127.0.0.1:2114")]);
let replacement = mailbox.latest().unwrap();
assert!(!retry_waits_for_status_change(
&mut retry_after,
&replacement,
&Arc::new(status_with_targets(&[TargetInstance {
id: TargetId::from("old"),
generation: 1,
}]))
));
assert!(retry_after.is_none());
assert_eq!(replacement.targets[0].id.as_str(), "c");
mailbox.clear_if_current(replacement.revision);
assert!(mailbox.latest().is_none());
}
#[test]
fn eof_and_shutdown_stop_without_waiting_for_the_stdin_reader() {
let mailbox = StdinMailbox::default();
let shutdown_requested = AtomicBool::new(false);
assert!(stdin_stop_request(&shutdown_requested, &mailbox).is_none());
mailbox.finish();
assert!(matches!(
stdin_stop_request(&shutdown_requested, &mailbox),
Some(StdinStop::Eof)
));
let mailbox = StdinMailbox::default();
shutdown_requested.store(true, std::sync::atomic::Ordering::Relaxed);
assert!(matches!(
stdin_stop_request(&shutdown_requested, &mailbox),
Some(StdinStop::Interrupted)
));
}
fn status_with_targets(targets: &[TargetInstance]) -> ManagedStatus {
ManagedStatus {
lifecycle: ManagedLifecycle::Running,
stop_requested: false,
applied_command_sequence: 0,
desired_target_count: targets.len(),
connecting_target_count: 0,
opening_target_count: 0,
active_target_count: 0,
draining_target_count: 0,
closing_target_count: 0,
terminal_target_count: 0,
total_target_outcomes: 0,
successful_target_outcomes: 0,
failed_target_outcomes: 0,
peer_closed_target_outcomes: 0,
discarded_target_outcomes: 0,
targets: Arc::from(
targets
.iter()
.cloned()
.map(|target| ManagedTargetStatus {
target,
desired: true,
lifecycle: ManagedTargetLifecycle::Active,
server_addr: Arc::from("127.0.0.1:2112"),
remote: None,
})
.collect::<Vec<_>>()
.into_boxed_slice(),
),
recent_target_outcomes: Arc::from([]),
final_outcome: None,
}
}
#[test]
fn dynamic_accounting_prunes_retired_generations_and_retains_only_shutdown_snapshot() {
let retired = TargetInstance {
id: TargetId::from("edge"),
generation: 1,
};
let replacement = TargetInstance {
id: TargetId::from("edge"),
generation: 2,
};
let live_status = status_with_targets(std::slice::from_ref(&replacement));
let empty_status = status_with_targets(&[]);
let mut stats = BTreeMap::from([
(retired.clone(), StatsCollector::new(stats_config(true))),
(replacement.clone(), StatsCollector::new(stats_config(true))),
]);
reconcile_stdin_stats(&mut stats, &live_status, None);
assert_eq!(stats.keys().cloned().collect::<Vec<_>>(), [replacement]);
let summary_targets = snapshot_stdin_summary_targets(&mut stats, &live_status);
stats.insert(retired, StatsCollector::new(stats_config(true)));
reconcile_stdin_stats(&mut stats, &empty_status, Some(&summary_targets));
assert_eq!(
stats.keys().cloned().collect::<Vec<_>>(),
[TargetInstance {
id: TargetId::from("edge"),
generation: 2,
}]
);
}
#[test]
fn terminal_deduplication_is_bounded_under_target_churn() {
let mut terminal_targets = BoundedTargetSet::new(2);
let target = |generation| TargetInstance {
id: TargetId::from("edge"),
generation,
};
assert!(terminal_targets.insert(target(1)));
assert!(terminal_targets.insert(target(2)));
assert!(terminal_targets.insert(target(3)));
assert!(terminal_targets.insert(target(1)));
assert!(!terminal_targets.insert(target(3)));
}
fn long_finite_args() -> ClientArgs { fn long_finite_args() -> ClientArgs {
use clap::Parser; use clap::Parser;
+6 -6
View File
@@ -14,13 +14,13 @@ pub const DEFAULT_TUI_DURATION: Duration = Duration::ZERO;
#[derive(Debug, Clone, Parser)] #[derive(Debug, Clone, Parser)]
#[command(name = "irtt-tui", about = "Minimal IRTT-compatible TUI client")] #[command(name = "irtt-tui", about = "Minimal IRTT-compatible TUI client")]
pub struct TuiArgs { pub struct TuiArgs {
/// Server address/host, optionally prefixed with LABEL=. Repeat for multi-target mode. /// Server address/host, optionally prefixed with LABEL= and suffixed with @hmac=KEY. Repeat for multi-target mode.
#[arg( #[arg(
value_name = "TARGET", value_name = "TARGET",
num_args = 1.., num_args = 1..,
required = true, required = true,
value_parser = parse_target, value_parser = parse_target,
long_help = "Server address/host, optionally prefixed with LABEL=. Repeat for multi-target mode.\n\nExplicit labels are used in the legend and status table.\n\nExamples:\n irtt-tui host.example\n irtt-tui eu=host.example\n irtt-tui eu=host-a.example us=host-b.example" long_help = "Server address/host, optionally prefixed with LABEL= and suffixed with @hmac=KEY. Repeat for multi-target mode.\n\nExplicit labels are used in the legend and status table. A target without @hmac= inherits --hmac; @hmac= disables it for that target.\n\nExamples:\n irtt-tui host.example\n irtt-tui eu=host.example\n irtt-tui eu=host-a.example@hmac=secret us=host-b.example"
)] )]
pub targets: Vec<TargetArg>, pub targets: Vec<TargetArg>,
@@ -58,6 +58,7 @@ impl TuiArgs {
TargetSelection { TargetSelection {
targets: &self.targets, targets: &self.targets,
pacing: self.pacing, pacing: self.pacing,
stdin_controlled: false,
}, },
) )
} }
@@ -107,8 +108,7 @@ mod tests {
fn tui_parser_defaults_to_continuous_and_has_no_output_option() { fn tui_parser_defaults_to_continuous_and_has_no_output_option() {
let args = parse(&["127.0.0.1:2112"]).unwrap(); let args = parse(&["127.0.0.1:2112"]).unwrap();
assert_eq!(args.targets.len(), 1); assert_eq!(args.targets.len(), 1);
assert_eq!(args.targets[0].label, None); assert_eq!(args.prepare().unwrap().targets[0].label, "127.0.0.1:2112");
assert_eq!(args.targets[0].addr, "127.0.0.1:2112");
assert_eq!(args.pacing, GroupPacingArg::Staggered); assert_eq!(args.pacing, GroupPacingArg::Staggered);
assert_eq!(args.duration, DEFAULT_TUI_DURATION); assert_eq!(args.duration, DEFAULT_TUI_DURATION);
assert!(args.is_continuous()); assert!(args.is_continuous());
@@ -181,8 +181,8 @@ mod tests {
#[test] #[test]
fn invalid_labelled_target_syntax_is_rejected() { fn invalid_labelled_target_syntax_is_rejected() {
assert!(parse(&["=127.0.0.1:2112"]).is_err()); assert!(parse(&["=127.0.0.1:2112"]).unwrap().prepare().is_err());
assert!(parse(&["label="]).is_err()); assert!(parse(&["label="]).unwrap().prepare().is_err());
} }
#[test] #[test]
+6 -3
View File
@@ -9,9 +9,12 @@ pub use args::{
ClockArg, CommonClientArgs, ReceivedStatsArg, TimestampArg, ClockArg, CommonClientArgs, ReceivedStatsArg, TimestampArg,
}; };
pub use config::{expected_probe_count, DEFAULT_RECV_TIMEOUT}; pub use config::{expected_probe_count, DEFAULT_RECV_TIMEOUT};
pub use prepare::{prepare_managed_run, ManagedRunSetup, TargetSelection, MANAGED_EVENT_CAPACITY}; pub use prepare::{
prepare_managed_run, ManagedRunSetup, TargetSelection, MANAGED_EVENT_CAPACITY,
STDIN_MAX_DESIRED_TARGETS, STDIN_MAX_LIVE_TARGET_GENERATIONS, STDIN_OUTCOME_HISTORY_LIMIT,
};
pub use session::is_shutdown_requested; pub use session::is_shutdown_requested;
pub use targets::{ pub use targets::{
parse_target, prepare_managed_targets, target_specs, GroupPacingArg, PreparedTarget, TargetArg, parse_stdin_target_set, parse_target, prepare_managed_targets, target_specs, GroupPacingArg,
TargetSpec, PreparedTarget, TargetArg, TargetSpec,
}; };
+36 -11
View File
@@ -22,6 +22,9 @@ use super::{
/// Capacity of the lossy managed presentation event channel. /// Capacity of the lossy managed presentation event channel.
pub const MANAGED_EVENT_CAPACITY: usize = 16_384; pub const MANAGED_EVENT_CAPACITY: usize = 16_384;
pub const STDIN_MAX_DESIRED_TARGETS: usize = 128;
pub const STDIN_MAX_LIVE_TARGET_GENERATIONS: usize = 256;
pub const STDIN_OUTCOME_HISTORY_LIMIT: usize = 256;
/// Target and pacing arguments common to the client applets. /// Target and pacing arguments common to the client applets.
/// ///
@@ -31,6 +34,7 @@ pub const MANAGED_EVENT_CAPACITY: usize = 16_384;
pub struct TargetSelection<'a> { pub struct TargetSelection<'a> {
pub targets: &'a [TargetArg], pub targets: &'a [TargetArg],
pub pacing: GroupPacingArg, pub pacing: GroupPacingArg,
pub stdin_controlled: bool,
} }
/// Everything a managed run needs, prepared and validated together. /// Everything a managed run needs, prepared and validated together.
@@ -46,6 +50,8 @@ pub struct ManagedRunSetup {
pub client: ClientConfig, pub client: ClientConfig,
/// Send coordination across active targets. /// Send coordination across active targets.
pub pacing: ManagedPacing, pub pacing: ManagedPacing,
/// Whether this stream may replace its desired set from stdin.
pub stdin_controlled: bool,
} }
impl ManagedRunSetup { impl ManagedRunSetup {
@@ -70,7 +76,7 @@ impl ManagedRunSetup {
/// Managed driver configuration for this run. /// Managed driver configuration for this run.
pub fn managed_config(&self) -> ManagedClientConfig { pub fn managed_config(&self) -> ManagedClientConfig {
let target_count = self.target_count(); let target_count = self.target_count();
ManagedClientConfig { let mut config = ManagedClientConfig {
client: self.client.clone(), client: self.client.clone(),
pacing: self.pacing, pacing: self.pacing,
completion: ManagedCompletionPolicy::FinishWhenQuiescent, completion: ManagedCompletionPolicy::FinishWhenQuiescent,
@@ -78,7 +84,13 @@ impl ManagedRunSetup {
outcome_history_limit: target_count, outcome_history_limit: target_count,
max_live_target_generations: target_count, max_live_target_generations: target_count,
..ManagedClientConfig::default() ..ManagedClientConfig::default()
};
if self.stdin_controlled {
config.completion = ManagedCompletionPolicy::ExplicitStop;
config.outcome_history_limit = STDIN_OUTCOME_HISTORY_LIMIT;
config.max_live_target_generations = STDIN_MAX_LIVE_TARGET_GENERATIONS;
} }
config
} }
} }
@@ -91,12 +103,22 @@ pub fn prepare_managed_run(
duration: Duration, duration: Duration,
selection: TargetSelection<'_>, selection: TargetSelection<'_>,
) -> Result<ManagedRunSetup, String> { ) -> Result<ManagedRunSetup, String> {
let specs = target_specs(selection.targets)?; let specs = if selection.stdin_controlled {
if selection.targets.len() > STDIN_MAX_DESIRED_TARGETS {
return Err(format!(
"initial target set exceeds the {STDIN_MAX_DESIRED_TARGETS}-target limit"
));
}
super::targets::target_specs_with_empty(selection.targets, true)?
} else {
target_specs(selection.targets)?
};
let targets = prepare_managed_targets(specs)?; let targets = prepare_managed_targets(specs)?;
Ok(ManagedRunSetup { Ok(ManagedRunSetup {
targets, targets,
client: common.to_client_config(duration), client: common.to_client_config(duration),
pacing: selection.pacing.into(), pacing: selection.pacing.into(),
stdin_controlled: selection.stdin_controlled,
}) })
} }
@@ -119,17 +141,11 @@ mod tests {
} }
fn unlabeled(addr: &str) -> TargetArg { fn unlabeled(addr: &str) -> TargetArg {
TargetArg { TargetArg::new(addr)
label: None,
addr: addr.to_owned(),
}
} }
fn labeled(label: &str, addr: &str) -> TargetArg { fn labeled(label: &str, addr: &str) -> TargetArg {
TargetArg { TargetArg::new(format!("{label}={addr}"))
label: Some(label.to_owned()),
addr: addr.to_owned(),
}
} }
fn prepare( fn prepare(
@@ -137,7 +153,15 @@ mod tests {
pacing: GroupPacingArg, pacing: GroupPacingArg,
duration: Duration, duration: Duration,
) -> Result<ManagedRunSetup, String> { ) -> Result<ManagedRunSetup, String> {
prepare_managed_run(&common(&[]), duration, TargetSelection { targets, pacing }) prepare_managed_run(
&common(&[]),
duration,
TargetSelection {
targets,
pacing,
stdin_controlled: false,
},
)
} }
#[test] #[test]
@@ -212,6 +236,7 @@ mod tests {
TargetSelection { TargetSelection {
targets: &targets, targets: &targets,
pacing: GroupPacingArg::Staggered, pacing: GroupPacingArg::Staggered,
stdin_controlled: false,
}, },
) )
.unwrap(); .unwrap();
+394 -129
View File
@@ -1,26 +1,99 @@
use std::collections::HashSet; use std::{collections::HashSet, fmt};
use clap::ValueEnum; use clap::ValueEnum;
use irtt_client::managed::{ManagedPacing, ManagedTargetConfig, TargetId}; use irtt_client::{
managed::{ManagedPacing, ManagedTargetConfig, TargetId},
ClientAuthConfig,
};
#[derive(Debug, Clone, PartialEq, Eq)] /// One raw positional target captured by Clap.
///
/// Parsing happens during run preparation so invalid target syntax never makes
/// Clap include the original argument (which may carry an HMAC key) in an
/// error diagnostic.
#[derive(Clone, PartialEq, Eq)]
pub struct TargetArg { pub struct TargetArg {
pub label: Option<String>, input: String,
pub addr: String,
} }
#[derive(Debug, Clone, PartialEq, Eq)] impl TargetArg {
pub fn new(input: impl Into<String>) -> Self {
Self {
input: input.into(),
}
}
}
impl fmt::Debug for TargetArg {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("TargetArg(<redacted>)")
}
}
/// Capture one positional target without exposing it through Clap diagnostics.
pub fn parse_target(input: &str) -> Result<TargetArg, String> {
Ok(TargetArg::new(input))
}
#[derive(Clone, PartialEq, Eq)]
pub enum TargetAuth {
Inherit,
Override(Vec<u8>),
Disable,
}
impl fmt::Debug for TargetAuth {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Inherit => f.write_str("Inherit"),
Self::Override(_) => f.write_str("Override(<redacted>)"),
Self::Disable => f.write_str("Disable"),
}
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct TargetSpec { pub struct TargetSpec {
pub label: String, pub label: String,
pub addr: String, pub addr: String,
pub auth: TargetAuth,
} }
#[derive(Debug, Clone)] impl fmt::Debug for TargetSpec {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("TargetSpec")
.field("label", &self.label)
.field("addr", &self.addr)
.field("auth", &self.auth)
.finish()
}
}
#[derive(Clone)]
pub struct PreparedTarget { pub struct PreparedTarget {
pub label: String, pub label: String,
pub managed: ManagedTargetConfig, pub managed: ManagedTargetConfig,
} }
impl fmt::Debug for PreparedTarget {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("PreparedTarget")
.field("label", &self.label)
.field("server_addr", &self.managed.server_addr)
.field(
"auth",
&match &self.managed.auth {
None => TargetAuth::Inherit,
Some(ClientAuthConfig { hmac_key: Some(_) }) => {
TargetAuth::Override(Vec::new())
}
Some(ClientAuthConfig { hmac_key: None }) => TargetAuth::Disable,
},
)
.finish()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] #[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
pub enum GroupPacingArg { pub enum GroupPacingArg {
Staggered, Staggered,
@@ -36,191 +109,383 @@ impl From<GroupPacingArg> for ManagedPacing {
} }
} }
/// Parse one positional target argument, as `TARGET` or `LABEL=TARGET`. #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub fn parse_target(input: &str) -> Result<TargetArg, String> { enum TargetParseError {
match input.split_once('=') { EmptyLabel,
None => Ok(TargetArg { EmptyAddress,
label: None, InvalidEscape,
addr: input.to_owned(), EmptyTarget,
}),
Some((label, addr)) => {
if label.is_empty() {
return Err("target label must not be empty".to_owned());
}
if addr.is_empty() {
return Err("target address must not be empty".to_owned());
}
Ok(TargetArg {
label: Some(label.to_owned()),
addr: addr.to_owned(),
})
} }
impl fmt::Display for TargetParseError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let message = match self {
Self::EmptyLabel => "target label must not be empty",
Self::EmptyAddress => "target address must not be empty",
Self::InvalidEscape => "invalid target escape sequence",
Self::EmptyTarget => "target must not be empty",
};
f.write_str(message)
} }
} }
pub fn target_specs(targets: &[TargetArg]) -> Result<Vec<TargetSpec>, String> { fn split_unescaped(input: &str, delimiter: char) -> Result<Vec<&str>, TargetParseError> {
let mut parts = Vec::new();
let mut start = 0;
let mut escaped = false;
for (index, ch) in input.char_indices() {
if escaped {
escaped = false;
} else if ch == '\\' {
escaped = true;
} else if ch == delimiter {
parts.push(&input[start..index]);
start = index + ch.len_utf8();
}
}
if escaped {
return Err(TargetParseError::InvalidEscape);
}
parts.push(&input[start..]);
Ok(parts)
}
fn first_unescaped(input: &str, delimiter: char) -> Result<Option<usize>, TargetParseError> {
let mut escaped = false;
for (index, ch) in input.char_indices() {
if escaped {
escaped = false;
} else if ch == '\\' {
escaped = true;
} else if ch == delimiter {
return Ok(Some(index));
}
}
if escaped {
return Err(TargetParseError::InvalidEscape);
}
Ok(None)
}
fn unescape(input: &str) -> Result<String, TargetParseError> {
let mut output = String::with_capacity(input.len());
let mut escaped = false;
for ch in input.chars() {
if escaped {
if !matches!(ch, '\\' | '=' | ';' | ',' | '@') {
return Err(TargetParseError::InvalidEscape);
}
output.push(ch);
escaped = false;
} else if ch == '\\' {
escaped = true;
} else {
output.push(ch);
}
}
if escaped {
return Err(TargetParseError::InvalidEscape);
}
Ok(output)
}
fn hmac_modifier(input: &str) -> Result<Option<usize>, TargetParseError> {
let mut escaped = false;
for (index, ch) in input.char_indices() {
if escaped {
escaped = false;
} else if ch == '\\' {
escaped = true;
} else if input[index..].starts_with("@hmac=") {
return Ok(Some(index));
}
}
if escaped {
return Err(TargetParseError::InvalidEscape);
}
Ok(None)
}
fn parse_target_syntax(
input: &str,
) -> Result<(Option<String>, String, TargetAuth), TargetParseError> {
if input.is_empty() {
return Err(TargetParseError::EmptyTarget);
}
let (main, auth) = match hmac_modifier(input)? {
Some(index) => {
let value = unescape(&input[index + "@hmac=".len()..])?;
let auth = if value.is_empty() {
TargetAuth::Disable
} else {
TargetAuth::Override(value.into_bytes())
};
(&input[..index], auth)
}
None => (input, TargetAuth::Inherit),
};
let (label, address) = match first_unescaped(main, '=')? {
None => (None, unescape(main)?),
Some(index) => {
let label = unescape(&main[..index])?;
let address = unescape(&main[index + '='.len_utf8()..])?;
(Some(label), address)
}
};
if label.as_deref().is_some_and(str::is_empty) {
return Err(TargetParseError::EmptyLabel);
}
if address.is_empty() {
return Err(TargetParseError::EmptyAddress);
}
Ok((label, address, auth))
}
pub(crate) fn target_specs_with_empty(
targets: &[TargetArg],
allow_empty: bool,
) -> Result<Vec<TargetSpec>, String> {
let mut specs = Vec::with_capacity(targets.len()); let mut specs = Vec::with_capacity(targets.len());
let mut unlabeled_counts = std::collections::HashMap::<&str, usize>::new(); let mut unlabeled_counts = std::collections::HashMap::<String, usize>::new();
for target in targets { for (index, target) in targets.iter().enumerate() {
let label = match &target.label { let (explicit_label, addr, auth) = parse_target_syntax(&target.input)
Some(label) => label.clone(), .map_err(|error| format!("invalid target {}: {error}", index + 1))?;
let label = match explicit_label {
Some(label) => label,
None => { None => {
let count = unlabeled_counts.entry(target.addr.as_str()).or_default(); let count = unlabeled_counts.entry(addr.clone()).or_default();
*count += 1; *count += 1;
if *count == 1 { if *count == 1 {
target.addr.clone() addr.clone()
} else { } else {
format!("{}#{}", target.addr, *count) format!("{}#{}", addr, *count)
} }
} }
}; };
specs.push(TargetSpec { specs.push(TargetSpec { label, addr, auth });
label,
addr: target.addr.clone(),
});
} }
if specs.is_empty() { if specs.is_empty() && !allow_empty {
return Err("at least one target is required unless --list-columns is set".to_owned()); return Err("at least one target is required unless --list-columns is set".to_owned());
} }
let mut labels = HashSet::new(); let mut labels = HashSet::new();
for spec in &specs { for spec in &specs {
if !labels.insert(spec.label.clone()) { if !labels.insert(spec.label.clone()) {
return Err(format!("duplicate target label {:?}", spec.label)); return Err("duplicate target label".to_owned());
} }
} }
Ok(specs) Ok(specs)
} }
pub fn target_specs(targets: &[TargetArg]) -> Result<Vec<TargetSpec>, String> {
target_specs_with_empty(targets, false)
}
pub fn prepare_managed_targets(specs: Vec<TargetSpec>) -> Result<Vec<PreparedTarget>, String> { pub fn prepare_managed_targets(specs: Vec<TargetSpec>) -> Result<Vec<PreparedTarget>, String> {
let mut targets = Vec::with_capacity(specs.len()); let mut targets = Vec::with_capacity(specs.len());
for spec in specs { for spec in specs {
let mut managed = ManagedTargetConfig::new(TargetId::from(spec.label.clone()), spec.addr);
managed.auth = match spec.auth {
TargetAuth::Inherit => None,
TargetAuth::Override(hmac_key) => Some(ClientAuthConfig {
hmac_key: Some(hmac_key),
}),
TargetAuth::Disable => Some(ClientAuthConfig { hmac_key: None }),
};
targets.push(PreparedTarget { targets.push(PreparedTarget {
label: spec.label.clone(), label: spec.label,
managed: ManagedTargetConfig::new(TargetId::from(spec.label), spec.addr), managed,
}); });
} }
Ok(targets) Ok(targets)
} }
/// Parse one complete stdin target set after its line terminator was removed.
///
/// Commas frame stdin elements only; positional target arguments do not use
/// this framing and may therefore contain raw commas.
pub fn parse_stdin_target_set(
record: &str,
maximum_targets: usize,
) -> Result<Vec<PreparedTarget>, String> {
if record == "[]" {
return Ok(Vec::new());
}
let elements = split_unescaped(record, ',').map_err(|error| error.to_string())?;
if elements.len() > maximum_targets {
return Err(format!(
"target set exceeds the {maximum_targets}-target limit"
));
}
let args = elements.into_iter().map(TargetArg::new).collect::<Vec<_>>();
prepare_managed_targets(target_specs_with_empty(&args, true)?)
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
fn unlabeled(addr: &str) -> TargetArg { fn target(input: &str) -> TargetArg {
TargetArg { TargetArg::new(input)
label: None,
addr: addr.to_owned(),
}
} }
fn labeled(label: &str, addr: &str) -> TargetArg { fn specs(inputs: &[&str]) -> Result<Vec<TargetSpec>, String> {
TargetArg { target_specs(&inputs.iter().map(|input| target(input)).collect::<Vec<_>>())
label: Some(label.to_owned()),
addr: addr.to_owned(),
}
} }
#[test] #[test]
fn parse_target_without_equals_is_unlabeled() { fn legacy_target_syntax_and_ipv6_remain_supported() {
let target = parse_target("host.example").unwrap(); let parsed = specs(&[
assert_eq!(target.label, None); "host.example",
assert_eq!(target.addr, "host.example"); "host.example:2112",
"[::1]:2112",
let target = parse_target("host.example:2112").unwrap(); "eu=host.example",
assert_eq!(target.label, None); ])
assert_eq!(target.addr, "host.example:2112"); .unwrap();
assert_eq!(parsed[0].label, "host.example");
let target = parse_target("[::1]:2112").unwrap(); assert_eq!(parsed[1].label, "host.example:2112");
assert_eq!(target.label, None); assert_eq!(parsed[2].label, "[::1]:2112");
assert_eq!(target.addr, "[::1]:2112"); assert_eq!(parsed[3].label, "eu");
assert!(parsed
.iter()
.all(|target| matches!(target.auth, TargetAuth::Inherit)));
} }
#[test] #[test]
fn parse_target_with_equals_splits_on_first() { fn target_hmac_supports_inherit_override_disable_and_base64_padding() {
let target = parse_target("eu=host.example").unwrap(); let parsed = specs(&[
assert_eq!(target.label, Some("eu".to_owned())); "default=default.example",
assert_eq!(target.addr, "host.example"); "private=private.example@hmac=YWJjZA==",
"public=public.example@hmac=",
let target = parse_target("eu=host.example:2112").unwrap(); ])
assert_eq!(target.label, Some("eu".to_owned())); .unwrap();
assert_eq!(target.addr, "host.example:2112"); assert!(matches!(parsed[0].auth, TargetAuth::Inherit));
assert_eq!(parsed[1].auth, TargetAuth::Override(b"YWJjZA==".to_vec()));
assert!(matches!(parsed[2].auth, TargetAuth::Disable));
} }
#[test] #[test]
fn parse_target_rejects_empty_label_or_address() { fn target_delimiters_are_escaped_contextually() {
assert!(parse_target("=host.example").is_err()); let parsed = specs(&["eu\\=west=host\\@hmac=literal@hmac=a\\@b\\,c\\=d\\\\e"]).unwrap();
assert!(parse_target("eu=").is_err()); assert_eq!(parsed[0].label, "eu=west");
assert_eq!(parsed[0].addr, "host@hmac=literal");
assert_eq!(parsed[0].auth, TargetAuth::Override(b"a@b,c=d\\e".to_vec()));
}
#[test]
fn positional_commas_are_not_reserved() {
let parsed = specs(&["edge=host,region@hmac=key,part"]).unwrap();
assert_eq!(parsed[0].addr, "host,region");
assert_eq!(parsed[0].auth, TargetAuth::Override(b"key,part".to_vec()));
}
#[test]
fn parser_diagnostics_and_debug_do_not_reveal_hmac_values() {
let secret = "very-secret-value";
let error = specs(&[&format!("target=host@hmac={secret}\\q")]).unwrap_err();
assert!(!error.contains(secret));
assert!(!format!("{:?}", target(&format!("target=host@hmac={secret}"))).contains(secret));
assert!(
!format!("{:?}", TargetAuth::Override(secret.as_bytes().to_vec())).contains(secret)
);
} }
#[test] #[test]
fn target_specs_suffix_repeated_unlabeled_and_reject_duplicate_labels() { fn target_specs_suffix_repeated_unlabeled_and_reject_duplicate_labels() {
let targets = vec![unlabeled("host-a:2112"), unlabeled("host-a:2112")]; let parsed = specs(&["host-a:2112", "host-a:2112"]).unwrap();
let specs = target_specs(&targets).unwrap(); assert_eq!(parsed[0].label, "host-a:2112");
assert_eq!(parsed[1].label, "host-a:2112#2");
assert_eq!(specs[0].label, "host-a:2112"); assert_eq!(
assert_eq!(specs[1].label, "host-a:2112#2"); specs(&["host-a:2112", "host-a:2112=host-b:2112"]).unwrap_err(),
"duplicate target label"
let targets = vec![ );
unlabeled("host-a:2112"),
labeled("host-a:2112", "host-b:2112"),
];
let err = target_specs(&targets).unwrap_err();
assert!(err.contains("duplicate target label"));
} }
#[test] #[test]
fn target_specs_preserve_argument_order_with_mixed_labels() { fn target_specs_preserve_argument_order_and_unlabeled_counts() {
let targets = vec![ let parsed = specs(&["local", "eu=eu.example", "foo", "us=us.example", "foo"]).unwrap();
unlabeled("local"), let labels: Vec<_> = parsed.iter().map(|target| target.label.as_str()).collect();
labeled("eu", "eu.example"), assert_eq!(labels, ["local", "eu", "foo", "us", "foo#2"]);
unlabeled("backup"), let addresses: Vec<_> = parsed.iter().map(|target| target.addr.as_str()).collect();
labeled("us", "us.example"), assert_eq!(
]; addresses,
let specs = target_specs(&targets).unwrap(); ["local", "eu.example", "foo", "us.example", "foo"]
);
let labels: Vec<_> = specs.iter().map(|spec| spec.label.as_str()).collect();
assert_eq!(labels, ["local", "eu", "backup", "us"]);
let addrs: Vec<_> = specs.iter().map(|spec| spec.addr.as_str()).collect();
assert_eq!(addrs, ["local", "eu.example", "backup", "us.example"]);
}
#[test]
fn target_specs_count_unlabeled_occurrences_independently_of_explicit_labels() {
let targets = vec![unlabeled("foo"), labeled("eu", "foo"), unlabeled("foo")];
let specs = target_specs(&targets).unwrap();
let labels: Vec<_> = specs.iter().map(|spec| spec.label.as_str()).collect();
assert_eq!(labels, ["foo", "eu", "foo#2"]);
} }
#[test] #[test]
fn target_specs_reject_generated_suffix_collisions() { fn target_specs_reject_generated_suffix_collisions() {
let targets = vec![ assert_eq!(
unlabeled("host.example"), specs(&[
unlabeled("host.example"), "host.example",
labeled("host.example#2", "other.example"), "host.example",
]; "host.example#2=other.example"
let err = target_specs(&targets).unwrap_err(); ])
assert!(err.contains("duplicate target label")); .unwrap_err(),
"duplicate target label"
);
} }
#[test] #[test]
fn target_preparation_preserves_hostname_and_allows_duplicate_endpoints() { fn malformed_target_syntax_is_rejected_without_echoing_input() {
let specs = vec![ for input in ["=host", "label=", "host@hmac=value\\q", "host\\x"] {
TargetSpec { assert!(specs(&[input]).is_err(), "{input}");
label: "a".to_owned(), }
addr: "example.test".to_owned(), }
},
TargetSpec { #[test]
label: "b".to_owned(), fn target_preparation_maps_auth_to_managed_config() {
addr: "example.test".to_owned(), let prepared =
}, prepare_managed_targets(specs(&["a=one@hmac=key", "b=two@hmac="]).unwrap()).unwrap();
]; assert_eq!(
let targets = prepare_managed_targets(specs).unwrap(); prepared[0]
assert_eq!(targets[0].managed.server_addr, "example.test"); .managed
assert_eq!(targets[1].managed.server_addr, "example.test"); .auth
.as_ref()
.unwrap()
.hmac_key
.as_deref(),
Some(b"key".as_slice())
);
assert_eq!(prepared[1].managed.auth.as_ref().unwrap().hmac_key, None);
}
#[test]
fn target_preparation_preserves_duplicate_endpoints() {
let prepared =
prepare_managed_targets(specs(&["a=example.test", "b=example.test"]).unwrap()).unwrap();
assert_eq!(prepared[0].managed.server_addr, "example.test");
assert_eq!(prepared[1].managed.server_addr, "example.test");
}
#[test]
fn stdin_target_sets_frame_commas_without_trimming_payload() {
let parsed = parse_stdin_target_set("a=one@hmac=x\\,y,b=two", 128).unwrap();
assert_eq!(
parsed[0].managed.auth.as_ref().unwrap().hmac_key.as_deref(),
Some(b"x,y".as_slice())
);
assert_eq!(parsed[1].label, "b");
assert_eq!(
parse_stdin_target_set(" []", 128).unwrap()[0]
.managed
.server_addr,
" []"
);
assert_eq!(
parse_stdin_target_set("[] ", 128).unwrap()[0]
.managed
.server_addr,
"[] "
);
assert!(parse_stdin_target_set("", 128).is_err());
assert!(parse_stdin_target_set("a=one,b=two", 1).is_err());
}
#[test]
fn stdin_empty_set_is_exactly_bracket_pair() {
assert!(parse_stdin_target_set("[]", 128).unwrap().is_empty());
} }
} }
+4 -2
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "irtt-client" name = "irtt-client"
version = "0.5.0" version = "0.5.1"
description = "IRTT-compatible Rust client library" description = "IRTT-compatible Rust client library"
edition.workspace = true edition.workspace = true
license.workspace = true license.workspace = true
@@ -44,6 +44,8 @@ required-features = ["tokio"]
doc-scrape-examples = true doc-scrape-examples = true
[dev-dependencies] [dev-dependencies]
irtt-server.workspace = true # Keep the packaged client testable against the latest published server while
# workspace tests continue to exercise the local source through this path.
irtt-server = { version = "0.5.1", path = "../irtt-server" }
proptest = "1.11.0" proptest = "1.11.0"
tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] } tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] }
@@ -250,15 +250,11 @@ fn dynamic_update_reaches_background_worker() {
.unwrap(); .unwrap();
let handle = owner.handle(); let handle = owner.handle();
let acknowledgement = test_runtime().block_on(async { let acknowledgement = handle
timeout( .update_targets(vec![target(&server)])
Duration::from_secs(2),
handle.update_targets(vec![target(&server)]).unwrap(),
)
.await
.unwrap() .unwrap()
.unwrap() .blocking_wait()
}); .unwrap();
assert_eq!(acknowledgement.sequence, 1); assert_eq!(acknowledgement.sequence, 1);
server.wait_probe(); server.wait_probe();
test_runtime().block_on(async { test_runtime().block_on(async {
+13
View File
@@ -301,6 +301,19 @@ pub struct ManagedCommandReceipt {
receiver: oneshot::Receiver<Result<ManagedCommandAcknowledgement, ManagedCommandApplyError>>, receiver: oneshot::Receiver<Result<ManagedCommandAcknowledgement, ManagedCommandApplyError>>,
} }
impl ManagedCommandReceipt {
/// Block until the target-set transaction is applied or rejected.
///
/// This is for synchronous callers such as [`crate::managed::BlockingManagedClient`]. It
/// must not be called from an asynchronous runtime; use the [`Future`]
/// implementation there instead.
pub fn blocking_wait(self) -> Result<ManagedCommandAcknowledgement, ManagedCommandApplyError> {
self.receiver
.blocking_recv()
.unwrap_or(Err(ManagedCommandApplyError::AcknowledgementDisconnected))
}
}
impl Future for ManagedCommandReceipt { impl Future for ManagedCommandReceipt {
type Output = Result<ManagedCommandAcknowledgement, ManagedCommandApplyError>; type Output = Result<ManagedCommandAcknowledgement, ManagedCommandApplyError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+31
View File
@@ -1576,6 +1576,37 @@ fn authenticated_peer_close_outcome() {
assert!(!has_close(&records)); assert!(!has_close(&records));
} }
#[test]
fn peer_close_leaves_other_explicitly_managed_targets_running() {
let closing = start_server(ServerBehavior::PeerClose, None);
let continuing = start_server(ServerBehavior::Echo, None);
let mut managed = config(ManagedPacing::Staggered);
managed.completion = ManagedCompletionPolicy::ExplicitStop;
managed.client.duration = None;
let (task, handle) = ManagedClient::task(
managed,
vec![
target("closing", closing.addr),
target("continuing", continuing.addr),
],
)
.unwrap();
let runtime = runtime();
let mut task = Box::pin(task);
drive_task_until(&runtime, &mut task, |task| {
handle.status().peer_closed_target_outcomes == 1
&& task.targets[1].counters.packets_sent != 0
});
assert_eq!(handle.status().lifecycle, ManagedLifecycle::Running);
let stop = handle.stop();
runtime.block_on(task);
runtime.block_on(stop);
closing.finish();
assert!(!probes(&continuing.finish()).is_empty());
}
#[test] #[test]
fn peer_close_during_stop_drain_remains_authoritative() { fn peer_close_during_stop_drain_remains_authoritative() {
let key = b"managed-drain-peer-close".to_vec(); let key = b"managed-drain-peer-close".to_vec();
+1 -1
View File
@@ -139,5 +139,5 @@ scenario contains the existing comparison harness output under
`netem-summary.json` files. Server stdout/stderr are captured at the run root as `netem-summary.json` files. Server stdout/stderr are captured at the run root as
`server.stdout` and `server.stderr`. `server.stdout` and `server.stderr`.
On macOS, `ip netns` and `tc netem` are unavailable; use the GitHub Actions On macOS, `ip netns` and `tc netem` are unavailable; use the Gitea Actions
workflow for netem coverage. workflow for netem coverage.
+85 -7
View File
@@ -6,7 +6,7 @@ irtt-client - IRTT-compatible stream client
## SYNOPSIS ## SYNOPSIS
`irtt-client` [*OPTIONS*] [*[LABEL=]TARGET*]... `irtt-client` [*OPTIONS*] [*[LABEL=]TARGET[@hmac=KEY]*]...
`irtt-client` `--list-columns` `irtt-client` `--list-columns`
@@ -18,17 +18,39 @@ eligible runs.
## TARGETS ## TARGETS
Every target argument accepts `[LABEL=]TARGET`. An optional `LABEL=` prefix Every target argument accepts `[LABEL=]TARGET[@hmac=KEY]`. An optional
assigns the logical target name used in output; without it, the target `LABEL=` prefix assigns the logical target name used in output; without it,
string itself is used as the label. Labeled and unlabeled targets can be the target string itself is used as the label. Labeled and unlabeled targets
freely mixed in one argument list. can be freely mixed in one argument list. Prefer explicit labels for
long-running dynamic target sets: the label is the stable target identity
across replacements.
```sh ```sh
irtt-client host-a:2112 irtt-client host-a:2112
irtt-client eu=host-a:2112 us=host-b:2112 irtt-client eu=host-a:2112 us=host-b:2112
irtt-client host-a:2112 eu=host-b:2112 irtt-client host-a:2112 eu=host-b:2112
irtt-client --hmac default-secret eu=host-a:2112@hmac=eu-secret public=host-b:2112@hmac=
``` ```
Targets without `@hmac=` inherit the global `--hmac` value. `@hmac=KEY`
overrides it for that target, and `@hmac=` explicitly disables HMAC for that
target. The modifier uses `@`, not a shell command separator; quote an argument
if its key needs shell quoting.
Target syntax supports these backslash escapes in labels, addresses, and HMAC
values: `\\` for a literal backslash, `\@` for a literal `@`, `\=` for a
literal `=`, and `\;` for a literal semicolon. Escaping `@` is necessary in a
label or address when it would otherwise begin `@hmac=`; escaping the first
`=` is how an unlabeled target or a label can contain that character. An `=` in
an HMAC value, including Base64 padding, does not need escaping. Positional
target arguments may contain a raw comma; stdin target sets have the additional
`\,` escape described below.
Shells process unquoted backslashes and characters such as semicolons before
the client sees them. Single-quote a target when it contains parser escapes,
for example `'west\=edge=host\@hmac=literal@hmac=a\;b\\c'`, so the
backslashes reach `irtt-client` unchanged.
Multiple targets are probed concurrently. By default, a multi-target session Multiple targets are probed concurrently. By default, a multi-target session
uses **staggered** pacing, spacing active targets across the probe interval; uses **staggered** pacing, spacing active targets across the probe interval;
pass `--pacing burst` to send one probe to every active target back-to-back pass `--pacing burst` to send one probe to every active target back-to-back
@@ -36,6 +58,57 @@ once per interval instead. Final per-target summaries print in the order
targets were supplied on the command line, not the order in which they targets were supplied on the command line, not the order in which they
finish or their labels sort alphabetically. finish or their labels sort alphabetically.
## DYNAMIC TARGET SETS FROM STANDARD INPUT
`--targets-stdin` is available only with continuous mode (`--duration 0`).
Positional targets remain an initial desired set, and may be omitted to
start empty:
~~~sh
irtt-client --duration 0 --targets-stdin
irtt-client --duration 0 --targets-stdin ams=ams.example:2112
~~~
Each non-empty stdin record declares the complete desired target set, rather
than a delta. A later record replaces an earlier set: targets absent from it
are retired, unchanged target configurations retain their generation, and a
changed address or HMAC setting creates a fresh generation. Under transient
live-generation backpressure, only the latest unapplied desired set is kept.
Commas frame targets in stdin records; escape a literal comma within one
target as `\,`. Target parsing inside each element is otherwise the same as
for positional arguments, including the other escapes above, `@hmac=KEY`, and
`@hmac=`. Use a persistent producer or keep a pipe/FIFO writer open while the
client is being controlled. For example, in one terminal:
~~~sh
mkfifo /tmp/irtt-targets
irtt-client --duration 0 --targets-stdin --hmac default-secret \
< /tmp/irtt-targets
~~~
Then open the FIFO once from another terminal and keep that descriptor open
while publishing complete desired sets:
~~~sh
exec 3> /tmp/irtt-targets
printf '%s\n' 'ams=ams.example:2112,sg=sg.example:2112@hmac=sg-secret' >&3
printf '%s\n' 'sg=sg.example:2112@hmac=sg-secret,nyc=nyc.example:2112' >&3
# Publish more sets as needed, then send EOF and stop the client:
exec 3>&-
~~~
The exact record `[]` selects an empty desired set. Empty records are
ignored; whitespace is otherwise payload, not trimming. EOF is immediate loss
or closure of the control plane and requests a graceful stop. A desired-state
revision that was read but has not yet been acknowledged as applied may be
discarded when EOF is observed; EOF does not wait for a capacity-rejected
revision to become applicable.
The maximum stdin record payload is 64 KiB. Stdin-controlled mode bounds each
desired set to 128 targets and retains at most 256 live target generations
while replacements drain.
## FINITE VERSUS CONTINUOUS OPERATION ## FINITE VERSUS CONTINUOUS OPERATION
`--duration` controls the run length: `--duration` controls the run length:
@@ -45,8 +118,9 @@ finish or their labels sort alphabetically.
(see [OUTPUT](#output) below — CSV, TSV, and JSON Lines never print one). (see [OUTPUT](#output) below — CSV, TSV, and JSON Lines never print one).
- `--duration 0` is **continuous**: the client runs until interrupted - `--duration 0` is **continuous**: the client runs until interrupted
(`Ctrl-C`). For table output, a final summary is printed only if the run (`Ctrl-C`). For table output, a final summary is printed only if the run
was interrupted; an uninterrupted continuous run (one that exits for was interrupted. Stdin-controlled mode is the exception: EOF also requests
another reason) does not get one. a graceful stop and prints the final retained summary. An uninterrupted
continuous run that exits for another reason does not get one.
See [MEASUREMENTS AND MEMORY](#measurements-and-memory) below for why this See [MEASUREMENTS AND MEMORY](#measurements-and-memory) below for why this
distinction also determines what statistics can be computed and how much distinction also determines what statistics can be computed and how much
@@ -192,6 +266,10 @@ the client exits nonzero — so a process supervisor can restart it; an
interrupted run (`Ctrl-C`) never reports peer closure as an error, whichever interrupted run (`Ctrl-C`) never reports peer closure as an error, whichever
mode it was in. mode it was in.
With `--targets-stdin`, peer closure is target-local: other desired targets
and the stdin controller continue running. Including that target in a later
complete set starts a fresh generation.
## EXIT STATUS ## EXIT STATUS
`0` on success (including a finite run's ordinary completion, and an `0` on success (including a finite run's ordinary completion, and an
+4 -4
View File
@@ -19,9 +19,9 @@ dependencies = [
[[package]] [[package]]
name = "cc" name = "cc"
version = "1.4.3" version = "1.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d" checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
dependencies = [ dependencies = [
"find-msvc-tools", "find-msvc-tools",
"jobserver", "jobserver",
@@ -204,9 +204,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
[[package]] [[package]]
name = "syn" name = "syn"
version = "3.0.3" version = "3.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" checksum = "e6275cddf4610d1775e6d1fe9469b2e77d0f39fd98fb7450901b821e0c53649f"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
+1 -1
View File
@@ -8,7 +8,7 @@ edition = "2021"
cargo-fuzz = true cargo-fuzz = true
[dependencies] [dependencies]
libfuzzer-sys = "0.4" libfuzzer-sys = "0.4.13"
[dependencies.irtt-proto] [dependencies.irtt-proto]
path = "../crates/irtt-proto" path = "../crates/irtt-proto"