feat(reflectors): refactor reflector configuration

Ensure IcmpEngine initializes within a Tokio runtime, improve timestamp validation, and refine probe management.

.gitea/workflows/build.yml

@@ -0,0 +1,114 @@
+name: Build binaries
+
+on:
+  push:
+  pull_request:
+    branches: [master]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+env:
+  CARGO_TERM_COLOR: always
+  # Measure before stripping; the packaged copy is stripped below.
+  CARGO_PROFILE_RELEASE_STRIP: none
+
+jobs:
+  build:
+    name: Build (${{ matrix.target }})
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    env:
+      TARGET: ${{ matrix.target }}
+    defaults:
+      run:
+        shell: bash
+    strategy:
+      fail-fast: false
+      # Supported by Gitea 28; bound concurrent cross-build memory use.
+      max-parallel: 2
+      matrix:
+        target:
+          - x86_64-musl
+          - i686-musl
+          - arm-musleabi
+          - arm-musleabihf
+          - armv7-musleabi
+          - armv7-musleabihf
+          - aarch64-musl
+          - mips-musl
+          - mipsel-musl
+          - mips64-muslabi64
+          - mips64el-muslabi64
+          - riscv64gc-musl
+    steps:
+      - name: Checkout
+        uses: builtin:checkout
+
+      - name: Configure Rust directories
+        # Cross remote's symlink traversal skips hidden directories under /root.
+        run: |
+          echo "CARGO_HOME=$RUNNER_TEMP/cargo" >> "$GITHUB_ENV"
+          echo "RUSTUP_HOME=$RUNNER_TEMP/rustup" >> "$GITHUB_ENV"
+          echo "$RUNNER_TEMP/cargo/bin" >> "$GITHUB_PATH"
+
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@ecabd13d1c56bd1345c230e542e9144811ad706f # v2
+        with:
+          toolchain: ${{ startsWith(matrix.target, 'mips') && 'nightly' || 'stable' }}
+          cache: false
+          build-warnings: warn
+
+      - name: Install cross and binary tools
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends binutils llvm
+          cargo install --git https://github.com/cross-rs/cross --locked cross
+          echo "RUST_TARGET=${TARGET%-*}-unknown-linux-${TARGET#*-}" >> "$GITHUB_ENV"
+
+      - name: Build release binary
+        env:
+          # MIPS build-std lacks Rust's bundled CRT objects; use the cross GCC copies.
+          RUSTFLAGS: ${{ startsWith(matrix.target, 'mips') && '-C target-feature=+crt-static -C link-self-contained=no' || '-C target-feature=+crt-static' }}
+          # Copy toolchains and sources instead of translating nested bind mounts.
+          CROSS_REMOTE: "true"
+          # Cross's per-artifact JSON capture hides compiler diagnostics.
+          CROSS_REMOTE_COPY_FULL_TARGET_DIR: ${{ startsWith(matrix.target, 'mips') && 'true' || 'false' }}
+        run: |
+          cross build --release --locked --target "$RUST_TARGET"
+
+      - name: Verify, package, and summarize binary
+        run: |
+          binary="target/${RUST_TARGET}/release/sqm-autorate-rust"
+          readelf --program-headers "$binary" > /tmp/program-headers.txt
+          readelf --dynamic "$binary" > /tmp/dynamic-section.txt
+          if grep -q INTERP /tmp/program-headers.txt || grep -q NEEDED /tmp/dynamic-section.txt; then
+            echo "Binary is not statically linked" >&2
+            exit 1
+          fi
+          mkdir -p dist
+          stripped="dist/sqm-autorate-rust-${TARGET}"
+          artifact="${stripped}.tar.gz"
+          cp "$binary" "$stripped"
+          llvm-strip --strip-all "$stripped"
+          # Tar preserves executable permissions when downloading the artifact.
+          tar -czf "$artifact" -C dist "sqm-autorate-rust-${TARGET}"
+
+          size() {
+            stat -c %s "$1" | numfmt --to=iec-i --suffix=B --format='%.1f'
+          }
+          {
+            echo '| Target | Unstripped | Stripped | Compressed |'
+            echo '|:--|--:|--:|--:|'
+            printf '| %s | %s | %s | %s |\n' \
+              "$TARGET" "$(size "$binary")" "$(size "$stripped")" "$(size "$artifact")"
+          } >> "$GITHUB_STEP_SUMMARY"
+
+      - name: Upload binary
+        # Use the v3 artifact protocol supported by Gitea.
+        uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2
+        with:
+          name: sqm-autorate-rust-${{ matrix.target }}
+          path: dist/sqm-autorate-rust-${{ matrix.target }}.tar.gz
+          if-no-files-found: error

.gitea/workflows/ci.yml

@@ -0,0 +1,110 @@
+name: Rust CI
+
+on:
+  push:
+    branches: [master]
+  pull_request:
+    branches: [master]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  fmt:
+    name: cargo fmt
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
+        with:
+          persist-credentials: false
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@ecabd13d1c56bd1345c230e542e9144811ad706f # v2
+        with:
+          components: rustfmt
+          cache: false
+      - name: Check formatting
+        run: cargo fmt --all -- --check
+
+  clippy:
+    name: cargo clippy
+    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@ecabd13d1c56bd1345c230e542e9144811ad706f # v2
+        with:
+          components: clippy
+          cache: false
+      - name: Run clippy with all features
+        run: cargo clippy --workspace --all-targets --all-features --locked -- -D warnings
+      - name: Run clippy without default features
+        run: cargo clippy --workspace --all-targets --no-default-features --features toml --locked -- -D warnings
+
+  verify:
+    name: cargo ${{ matrix.task }} (${{ matrix.target }})
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - task: check
+            target: x86_64-unknown-linux-gnu
+            musl: false
+          - task: test
+            target: x86_64-unknown-linux-gnu
+            musl: false
+          - task: check
+            target: x86_64-unknown-linux-musl
+            musl: true
+          - task: test
+            target: x86_64-unknown-linux-musl
+            musl: true
+    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@ecabd13d1c56bd1345c230e542e9144811ad706f # v2
+        with:
+          target: ${{ matrix.target }}
+          cache: false
+      - 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 --features toml --target ${{ matrix.target }} --locked
+
+  docs:
+    name: cargo doc
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    env:
+      RUSTDOCFLAGS: -D warnings
+    steps:
+      - name: Checkout
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
+        with:
+          persist-credentials: false
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@ecabd13d1c56bd1345c230e542e9144811ad706f # v2
+        with:
+          cache: false
+      - name: Check documentation
+        run: cargo doc --workspace --all-features --no-deps --locked
+      - name: Run doctests
+        run: cargo test --workspace --doc --all-features --locked

.gitea/workflows/supply-chain.yml

@@ -0,0 +1,66 @@
+name: Supply chain
+
+on:
+  push:
+    branches: [master]
+  pull_request:
+    branches: [master]
+  schedule:
+    - cron: "41 2 * * 1"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+env:
+  ACTIONLINT_VERSION: 1.7.7
+  CARGO_DENY_VERSION: 0.20.2
+
+jobs:
+  cargo-deny:
+    name: cargo-deny ${{ matrix.checks }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - checks: advisories
+            continue-on-error: true
+          - 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: cargo install cargo-deny --version "${CARGO_DENY_VERSION}" --locked
+      - 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: go install "github.com/rhysd/actionlint/cmd/actionlint@v${ACTIONLINT_VERSION}"
+      - name: Check workflow syntax
+        # actionlint does not recognize Gitea's built-in checkout action.
+        run: |
+          "$(go env GOPATH)/bin/actionlint" \
+            -ignore 'specifying action "builtin:checkout" in invalid format' \
+            .gitea/workflows/build.yml \
+            .gitea/workflows/ci.yml \
+            .gitea/workflows/supply-chain.yml \
+            .github/workflows/actions-security.yml \
+            .github/workflows/ci-arm.yml \
+            .github/workflows/devskim.yml \
+            .github/workflows/osv-scanner.yml \
+            .github/workflows/rust-clippy.yml

.github/workflows/actions-security.yml

@@ -0,0 +1,35 @@
+name: GitHub Actions Security
+
+on:
+  push:
+    branches: [master]
+  pull_request:
+    branches: [master]
+  schedule:
+    - cron: "11 3 * * 1"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  zizmor:
+    name: zizmor
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      actions: read
+      security-events: write
+    steps:
+      - name: Checkout
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
+        with:
+          persist-credentials: false
+      - name: Run zizmor
+        uses: zizmorcore/zizmor-action@cc914d7f3750a2d13d75c7f184a1060aa0e9d482 # v0.6.4
+        with:
+          persona: regular
+          inputs: |
+            .github/workflows
+            .gitea/workflows

.github/workflows/build.yml

@@ -1,22 +0,0 @@
-name: Build
-
-on:
-  push:
-    branches: [ "master" ]
-  pull_request:
-    branches: [ "master" ]
-
-env:
-  CARGO_TERM_COLOR: always
-
-jobs:
-  build:
-
-    runs-on: ubuntu-latest
-
-    steps:
-    - uses: actions/checkout@v7
-    - name: Install Mold
-      run: sudo apt install mold
-    - name: Build
-      run: cargo build --verbose

.github/workflows/ci-arm.yml

@@ -0,0 +1,55 @@
+name: Rust ARM CI
+
+on:
+  push:
+    branches: [master]
+  pull_request:
+    branches: [master]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  verify:
+    name: cargo ${{ matrix.task }} (${{ matrix.target }})
+    runs-on: ubuntu-24.04-arm
+    timeout-minutes: 30
+    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
+    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@ecabd13d1c56bd1345c230e542e9144811ad706f # v2
+        with:
+          target: ${{ matrix.target }}
+          cache: false
+      - 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 --features toml --target ${{ matrix.target }} --locked

.github/workflows/devskim.yml

@@ -1,33 +1,29 @@
-# This workflow uses actions that are not certified by GitHub.
-# They are provided by a third-party and are governed by
-# separate terms of service, privacy policy, and support
-# documentation.
-
+# This workflow uploads SARIF results to GitHub Code Scanning.
 name: DevSkim
 
 on:
   push:
-    branches: [ "master" ]
+    branches: [master]
   pull_request:
-    branches: [ "master" ]
+    branches: [master]
   schedule:
     - cron: '18 0 * * 4'
 
+permissions:
+  contents: read
+
 jobs:
   lint:
     name: DevSkim
     runs-on: ubuntu-latest
     permissions:
-      actions: read
       contents: read
       security-events: write
     steps:
       - name: Checkout code
         uses: actions/checkout@v7
-
       - name: Run DevSkim scanner
         uses: microsoft/DevSkim-Action@v1
-
       - name: Upload DevSkim scan results to GitHub Security tab
         uses: github/codeql-action/upload-sarif@v4
         with:

.github/workflows/osv-scanner.yml

@@ -1,55 +1,34 @@
 # Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
+# Licensed under the Apache License, Version 2.0.
 
-name: OSV-Scanner Scheduled Scan
+name: OSV-Scanner
 
 on:
   pull_request:
-    branches: ["master", "v1"]
+    branches: [master, v1]
   schedule:
     - cron: "12 12 * * 1"
   push:
-    branches: ["master", "v1"]
+    branches: [master, v1]
 
-# Restrict jobs in this workflow to have no permissions by default; permissions
-# should be granted per job as needed using a dedicated `permissions` block
 permissions: {}
 
 jobs:
   scan-scheduled:
     permissions:
-      contents: read # to fetch code (actions/checkout)
-      security-events: write # for uploading SARIF files
+      contents: read
+      security-events: write
       actions: read
     if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }}
-    # If you want to copy this config, highly suggest pinning this version to a release rather than tracking the main branch
-    uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@v2.5.0"
+    uses: google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@v2.6.0
     with:
-      # Just scan the root directory and docs, since everything else is fixtures
-      scan-args: |-
-        ./
-        ./docs/
+      scan-args: ./
   scan-pr:
     permissions:
-      contents: read # to fetch code (actions/checkout)
-      security-events: write # for uploading SARIF files
+      contents: read
+      security-events: write
       actions: read
     if: ${{ github.event_name == 'pull_request' || github.event_name == 'merge_group' }}
-    # If you want to copy this config, highly suggest pinning this version to a release rather than tracking the main branch
-    uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@v2.5.0"
+    uses: google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@v2.6.0
     with:
-      # Just scan the root directory and docs, since everything else is fixtures
-      scan-args: |-
-        ./
-        ./docs/
+      scan-args: ./

.github/workflows/rust-clippy.yml

@@ -1,55 +1,61 @@
-# This workflow uses actions that are not certified by GitHub.
-# They are provided by a third-party and are governed by
-# separate terms of service, privacy policy, and support
-# documentation.
-# rust-clippy is a tool that runs a bunch of lints to catch common
-# mistakes in your Rust code and help improve your Rust code.
-# More details at https://github.com/rust-lang/rust-clippy
-# and https://rust-lang.github.io/rust-clippy/
-
-name: rust-clippy analyze
+name: Clippy SARIF
 
 on:
   push:
-    branches: [ "master" ]
-  pull_request:
-    # The branches below must be a subset of the branches above
-    branches: [ "master" ]
+    branches: [master]
   schedule:
-    - cron: '28 20 * * 3'
+    - cron: "17 12 * * 4"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+  security-events: write
+  actions: read
+
+env:
+  CARGO_TERM_COLOR: always
 
 jobs:
-  rust-clippy-analyze:
-    name: Run rust-clippy analyzing
+  clippy-sarif:
+    name: Clippy SARIF
     runs-on: ubuntu-latest
-    permissions:
-      contents: read
-      security-events: write
-      actions: read # only required for a private repository by github/codeql-action/upload-sarif to get the Action run status
+    timeout-minutes: 25
     steps:
-      - name: Checkout code
-        uses: actions/checkout@v7
-
+      - name: Checkout
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
+        with:
+          persist-credentials: false
       - name: Install Rust toolchain
-        uses: actions-rs/toolchain@16499b5e05bf2e26879000db0c1d13f7e13fa3af #@v1
+        uses: actions-rust-lang/setup-rust-toolchain@ecabd13d1c56bd1345c230e542e9144811ad706f # v2
         with:
-          profile: minimal
-          toolchain: stable
           components: clippy
-          override: true
-
-      - name: Install required cargo
-        run: cargo install clippy-sarif sarif-fmt
-
-      - name: Run rust-clippy
-        run:
-          cargo clippy
-          --all-features
-          --message-format=json | clippy-sarif | tee rust-clippy-results.sarif | sarif-fmt
-        continue-on-error: true
-
-      - name: Upload analysis results to GitHub
-        uses: github/codeql-action/upload-sarif@v4
+          cache: false
+      - name: Install SARIF tools
+        run: cargo install --locked clippy-sarif sarif-fmt
+      - name: Run Clippy and generate SARIF
+        id: clippy
+        shell: bash
+        run: |
+          set +e
+          cargo clippy \
+            --workspace \
+            --all-targets \
+            --all-features \
+            --locked \
+            --message-format=json \
+            | clippy-sarif \
+            | tee rust-clippy-results.sarif \
+            | sarif-fmt
+          clippy_status=${PIPESTATUS[0]}
+          echo "exit_code=${clippy_status}" >> "$GITHUB_OUTPUT"
+          exit 0
+      - name: Upload Clippy SARIF
+        uses: github/codeql-action/upload-sarif@24c54180a607b1449ed407dd24f251e4e9147c8d # v4
+        if: always()
         with:
           sarif_file: rust-clippy-results.sarif
+          category: clippy
           wait-for-processing: true
+      - name: Fail if Clippy failed
+        if: steps.clippy.outputs.exit_code != '0'
+        run: exit 1

Cargo.lock

@@ -4,9 +4,9 @@ version = 4
 
 [[package]]
 name = "aho-corasick"
-version = "1.1.4"
+version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
+checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba"
 dependencies = [
  "memchr",
 ]
@@ -17,6 +17,23 @@ version = "1.0.104"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470"
 
+[[package]]
+name = "async-trait"
+version = "0.1.92"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+]
+
+[[package]]
+name = "autocfg"
+version = "1.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
+
 [[package]]
 name = "bindgen"
 version = "0.72.1"
@@ -39,9 +56,18 @@ dependencies = [
 
 [[package]]
 name = "bitflags"
-version = "2.13.1"
+version = "2.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da"
+checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
+
+[[package]]
+name = "block-buffer"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa"
+dependencies = [
+ "hybrid-array",
+]
 
 [[package]]
 name = "bumpalo"
@@ -50,10 +76,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
 
 [[package]]
-name = "cc"
-version = "1.3.0"
+name = "bytes"
+version = "1.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8"
+checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
+
+[[package]]
+name = "cc"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f74872d07caf508b30a21f6836e7d7016a2eaf7d9ff4f48deaa58cd8a0407630"
 dependencies = [
  "find-msvc-tools",
  "shlex 2.0.1",
@@ -70,15 +102,21 @@ dependencies = [
 
 [[package]]
 name = "cfg-if"
-version = "1.0.4"
+version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
+checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600"
+
+[[package]]
+name = "cfg_aliases"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527"
 
 [[package]]
 name = "clang-sys"
-version = "1.8.1"
+version = "1.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4"
+checksum = "157a8ba7b480713b56f4c09fd13fc3e0a22a5dfab8097ba61cbc5feef950788a"
 dependencies = [
  "glob",
  "libc",
@@ -95,10 +133,16 @@ dependencies = [
 ]
 
 [[package]]
-name = "config"
-version = "0.15.25"
+name = "cmov"
+version = "0.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b85f248a4de22d204ceabc6299d89d2c70fbd7f09fea53c06c852369652d8139"
+checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a"
+
+[[package]]
+name = "config"
+version = "0.15.27"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "38e9fbc53afcec54441422f1c596cd054655b32e70e1cbac956651b7b175a404"
 dependencies = [
  "pathdiff",
  "serde_core",
@@ -106,6 +150,21 @@ dependencies = [
  "winnow",
 ]
 
+[[package]]
+name = "const-oid"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c"
+
+[[package]]
+name = "crypto-common"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453"
+dependencies = [
+ "hybrid-array",
+]
+
 [[package]]
 name = "csv"
 version = "1.4.0"
@@ -128,10 +187,31 @@ dependencies = [
 ]
 
 [[package]]
-name = "either"
-version = "1.16.0"
+name = "ctutils"
+version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e"
+checksum = "03bb0e1cc970d482d121d9a1744999169b69a07470b3d644a7894e53fcaf4574"
+dependencies = [
+ "cmov",
+]
+
+[[package]]
+name = "digest"
+version = "0.11.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2"
+dependencies = [
+ "block-buffer",
+ "const-oid",
+ "crypto-common",
+ "ctutils",
+]
+
+[[package]]
+name = "either"
+version = "1.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be"
 
 [[package]]
 name = "equivalent"
@@ -160,9 +240,9 @@ dependencies = [
 
 [[package]]
 name = "find-msvc-tools"
-version = "0.1.9"
+version = "0.1.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
+checksum = "aedcfb3409746eddb02b9e19ebda1c3394f759a152e48ee875a0844d1b955484"
 
 [[package]]
 name = "flume"
@@ -171,9 +251,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be"
 dependencies = [
  "fastrand",
+ "futures-core",
+ "futures-sink",
  "spin",
 ]
 
+[[package]]
+name = "futures-core"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"
+
+[[package]]
+name = "futures-sink"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d"
+
 [[package]]
 name = "getrandom"
 version = "0.4.3"
@@ -189,9 +283,9 @@ dependencies = [
 
 [[package]]
 name = "glob"
-version = "0.3.3"
+version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280"
+checksum = "e4eba85ea1d0a966a983acd07deee566e67395d2d96b6fb39e62b5a833f1eb0b"
 
 [[package]]
 name = "hashbrown"
@@ -200,24 +294,71 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
 
 [[package]]
-name = "icmp-socket2"
-version = "0.6.0"
+name = "hmac"
+version = "0.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "86fc005b72eb9dbc1df87318cf906a415d4f4c0d5eea42849b29dd710cdf670d"
+checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f"
 dependencies = [
+ "digest",
+]
+
+[[package]]
+name = "hybrid-array"
+version = "0.4.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17"
+dependencies = [
+ "typenum",
+]
+
+[[package]]
+name = "icmp-socket2"
+version = "0.6.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "816dc56a4f7f76e42b0616818600b05528cd346cb3843276e82df982d5c3ee26"
+dependencies = [
+ "async-trait",
+ "nix",
  "socket2",
+ "thiserror",
+ "tokio",
 ]
 
 [[package]]
 name = "indexmap"
-version = "2.14.0"
+version = "2.14.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
+checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
 dependencies = [
  "equivalent",
  "hashbrown",
 ]
 
+[[package]]
+name = "irtt-client"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a59c24704bd94f19b84b2b0190aa9163efede499f8cf4eb02922443cfee08339"
+dependencies = [
+ "irtt-proto",
+ "nix",
+ "socket2",
+ "thiserror",
+ "tokio",
+]
+
+[[package]]
+name = "irtt-proto"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a56ff307983b8f6477724e08cf726c5e24975b9affeb853d0830d91c60026ab2"
+dependencies = [
+ "hmac",
+ "md-5",
+ "subtle",
+ "thiserror",
+]
+
 [[package]]
 name = "itertools"
 version = "0.13.0"
@@ -235,9 +376,9 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
 
 [[package]]
 name = "js-sys"
-version = "0.3.103"
+version = "0.3.106"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102"
+checksum = "7883d941dae510fb2d978fc3fe018c71c9e2892fd38854de3e8b92c2e5ad9cc5"
 dependencies = [
  "cfg-if",
  "wasm-bindgen",
@@ -245,9 +386,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.189"
+version = "0.2.190"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
+checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78"
 
 [[package]]
 name = "libloading"
@@ -262,7 +403,7 @@ dependencies = [
 [[package]]
 name = "libuci-sys"
 version = "1.1.0"
-source = "git+https://github.com/Lochnair/rust-uci.git?rev=fcfedd204fb76928ef617bf9631b041c816e03d4#fcfedd204fb76928ef617bf9631b041c816e03d4"
+source = "git+https://github.com/Lochnair/rust-uci.git?rev=6835397f4f4ba8b69e75e727105186865c357417#6835397f4f4ba8b69e75e727105186865c357417"
 dependencies = [
  "bindgen",
  "cc",
@@ -286,19 +427,18 @@ dependencies = [
 
 [[package]]
 name = "log"
-version = "0.4.33"
+version = "0.4.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad"
+checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
 
 [[package]]
-name = "maybe-async"
-version = "0.2.11"
+name = "md-5"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "746873a384ad60adc5db74471dfaba74bd278afbdcfd81db93fafcdfc8b5ca0c"
+checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98"
 dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.119",
+ "cfg-if",
+ "digest",
 ]
 
 [[package]]
@@ -307,6 +447,15 @@ version = "2.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
 
+[[package]]
+name = "memoffset"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a"
+dependencies = [
+ "autocfg",
+]
+
 [[package]]
 name = "minimal-lexical"
 version = "0.2.1"
@@ -314,20 +463,44 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
 
 [[package]]
-name = "netlink-bindings"
-version = "0.3.3"
+name = "mio"
+version = "1.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "114816d03a633cef7cfa951721dafbda16bb1ff6af83393ee568ec2d4e4ded2d"
+checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee"
+dependencies = [
+ "libc",
+ "wasi",
+ "windows-sys",
+]
+
+[[package]]
+name = "netlink-bindings"
+version = "0.3.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15c68f271f059945bc6fb55d210e78c1ceb2d3eac00d18fe9b0806ade81e8a21"
 
 [[package]]
 name = "netlink-socket2"
-version = "0.3.3"
+version = "0.3.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "11193969f046efafdc9d8670d395a6ddcf9b806169f504e520a35dc95cab3241"
+checksum = "0951d4d6a5131a59af90a26fcecd4e366e8fcee3a38880ba58705e479f5bd8c3"
 dependencies = [
  "libc",
- "maybe-async",
  "netlink-bindings",
+ "strip-async",
+]
+
+[[package]]
+name = "nix"
+version = "0.31.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d"
+dependencies = [
+ "bitflags",
+ "cfg-if",
+ "cfg_aliases",
+ "libc",
+ "memoffset",
 ]
 
 [[package]]
@@ -346,12 +519,41 @@ version = "1.21.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
 
+[[package]]
+name = "parking_lot"
+version = "0.12.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a"
+dependencies = [
+ "lock_api",
+ "parking_lot_core",
+]
+
+[[package]]
+name = "parking_lot_core"
+version = "0.9.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "redox_syscall",
+ "smallvec",
+ "windows-link",
+]
+
 [[package]]
 name = "pathdiff"
 version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3"
 
+[[package]]
+name = "pin-project-lite"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
+
 [[package]]
 name = "prettyplease"
 version = "0.2.37"
@@ -386,6 +588,15 @@ version = "6.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
 
+[[package]]
+name = "redox_syscall"
+version = "0.5.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
+dependencies = [
+ "bitflags",
+]
+
 [[package]]
 name = "regex"
 version = "1.13.1"
@@ -400,9 +611,9 @@ dependencies = [
 
 [[package]]
 name = "regex-automata"
-version = "0.4.16"
+version = "0.4.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad"
+checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -418,7 +629,7 @@ checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4"
 [[package]]
 name = "rust-uci"
 version = "0.1.6"
-source = "git+https://github.com/Lochnair/rust-uci.git?rev=fcfedd204fb76928ef617bf9631b041c816e03d4#fcfedd204fb76928ef617bf9631b041c816e03d4"
+source = "git+https://github.com/Lochnair/rust-uci.git?rev=6835397f4f4ba8b69e75e727105186865c357417#6835397f4f4ba8b69e75e727105186865c357417"
 dependencies = [
  "libc",
  "libuci-sys",
@@ -434,9 +645,9 @@ checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d"
 
 [[package]]
 name = "rustix"
-version = "1.1.4"
+version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
+checksum = "891efababe418670775f199f0d233d84843c227a0949a883ce15b37c78d6629d"
 dependencies = [
  "bitflags",
  "errno",
@@ -490,7 +701,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 3.0.3",
+ "syn 3.0.6",
 ]
 
 [[package]]
@@ -508,9 +719,9 @@ dependencies = [
 
 [[package]]
 name = "serde_spanned"
-version = "1.1.1"
+version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26"
+checksum = "4e7523beb55eece201a2356bee0bbca0d1ab466c14c07703b2e0ee6d42cb0c2c"
 dependencies = [
  "serde_core",
 ]
@@ -527,6 +738,22 @@ version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"
 
+[[package]]
+name = "signal-hook-registry"
+version = "1.4.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b"
+dependencies = [
+ "errno",
+ "libc",
+]
+
+[[package]]
+name = "smallvec"
+version = "1.16.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e"
+
 [[package]]
 name = "socket2"
 version = "0.6.5"
@@ -556,6 +783,7 @@ dependencies = [
  "fastrand",
  "flume",
  "icmp-socket2",
+ "irtt-client",
  "libc",
  "log",
  "netlink-bindings",
@@ -563,11 +791,23 @@ dependencies = [
  "rust-uci",
  "rustix",
  "serde",
- "serde_json",
  "thiserror",
+ "tokio",
  "uci-derive",
 ]
 
+[[package]]
+name = "strip-async"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85ec932792e62b2d3936163eacbd2e09984453bcd114f665c1e1694e36ff59c9"
+
+[[package]]
+name = "subtle"
+version = "2.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
+
 [[package]]
 name = "syn"
 version = "2.0.119"
@@ -581,9 +821,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "3.0.3"
+version = "3.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3"
+checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -591,10 +831,10 @@ dependencies = [
 ]
 
 [[package]]
-name = "target-triple"
-version = "1.0.1"
+name = "target-tuple"
+version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3a6bfce3d99adfa72d24750a61f782f3036a81e7f86d8841ee1326deaebd171"
+checksum = "876fef147edbcbddc8ac5cbbba92c7b86519e314e86638596c09673b2ed01e7f"
 
 [[package]]
 name = "termcolor"
@@ -607,29 +847,57 @@ dependencies = [
 
 [[package]]
 name = "thiserror"
-version = "2.0.19"
+version = "2.0.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9"
+checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e"
 dependencies = [
  "thiserror-impl",
 ]
 
 [[package]]
 name = "thiserror-impl"
-version = "2.0.19"
+version = "2.0.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd"
+checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 3.0.3",
+ "syn 3.0.6",
+]
+
+[[package]]
+name = "tokio"
+version = "1.53.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044"
+dependencies = [
+ "bytes",
+ "libc",
+ "mio",
+ "parking_lot",
+ "pin-project-lite",
+ "signal-hook-registry",
+ "socket2",
+ "tokio-macros",
+ "windows-sys",
+]
+
+[[package]]
+name = "tokio-macros"
+version = "2.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
 ]
 
 [[package]]
 name = "toml"
-version = "1.1.3+spec-1.1.0"
+version = "1.1.7+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "53c96ecdfa941c8fc4fcaed14f99ada8ebed502eef533015095a07e3301d4c3c"
+checksum = "7874cd34dd99040fc24e30923aa64a60a81ff3ddfe2b8129ccc84ba24949deed"
 dependencies = [
  "indexmap",
  "serde_core",
@@ -642,64 +910,76 @@ dependencies = [
 
 [[package]]
 name = "toml_datetime"
-version = "1.1.1+spec-1.1.0"
+version = "1.1.2+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7"
+checksum = "2b86d767906c6c42421dcba507eb9d203e779497710a47782a224bb871653053"
 dependencies = [
  "serde_core",
 ]
 
 [[package]]
 name = "toml_parser"
-version = "1.1.2+spec-1.1.0"
+version = "1.1.4+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526"
+checksum = "bed810ee007d8e7f9cf20b4774e7198142fa27186e0f805871da7f3c39eaf333"
 dependencies = [
  "winnow",
 ]
 
 [[package]]
 name = "toml_writer"
-version = "1.1.2+spec-1.1.0"
+version = "1.1.3+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2"
+checksum = "06bdbd8cfc056b8d2e2e85f29b56a3bdbecb527cef81eb39e3e7b98af4652770"
 
 [[package]]
 name = "trybuild"
-version = "1.0.120"
+version = "1.0.122"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e605bf6b39357663d8ba4e984f8be8da8df6bb32e81031d6889024ea8fd68e4"
+checksum = "62db9c92d704393fbf2132041720cc80b689f2d3f28521015c2ac866223c11b8"
 dependencies = [
  "glob",
  "serde",
  "serde_derive",
  "serde_json",
- "target-triple",
+ "target-tuple",
  "termcolor",
  "toml",
 ]
 
+[[package]]
+name = "typenum"
+version = "1.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
+
 [[package]]
 name = "uci-derive"
 version = "0.1.0"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 3.0.3",
+ "syn 3.0.6",
  "trybuild",
 ]
 
 [[package]]
 name = "unicode-ident"
-version = "1.0.24"
+version = "1.0.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
+checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
+
+[[package]]
+name = "wasi"
+version = "0.11.1+wasi-snapshot-preview1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.126"
+version = "0.2.129"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4"
+checksum = "9bb54f33acc68fd454578d9820b0bde1a1a3d17aa17bb7b6595806d02886d409"
 dependencies = [
  "cfg-if",
  "once_cell",
@@ -710,9 +990,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.126"
+version = "0.2.129"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1"
+checksum = "2e29d0c35b16e224a7eeb5cd2d25e3e1968fbd65604117b44d3b789d00ee8535"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -720,22 +1000,22 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.126"
+version = "0.2.129"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e"
+checksum = "6f501a8bc3719dba86ef8ae4728879c08001bea749eb1333ac5b91e040e2a6b7"
 dependencies = [
  "bumpalo",
  "proc-macro2",
  "quote",
- "syn 2.0.119",
+ "syn 3.0.6",
  "wasm-bindgen-shared",
 ]
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.126"
+version = "0.2.129"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24"
+checksum = "23f0c9c52aa7cd7d77769a4cfe2a9adb1b331f489a41d912ce14513d5ab995c6"
 dependencies = [
  "unicode-ident",
 ]

Cargo.toml

@@ -20,16 +20,17 @@ strip = "symbols"
 
 [dependencies]
 anyhow = "1.0"
-icmp-socket2 = "0.6.0"
-fastrand = "2"
-rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
-thiserror = "2.0"
-libc = "0.2"
-flume = { version = "0.12.0",  default-features = false, features = ["eventual-fairness", "select"] }
 config = { version = "0.15.25", default-features = false }
 csv = "1.4.0"
+fastrand = "2"
+flume = { version = "0.12.0",  default-features = false, features = ["async", "eventual-fairness", "select"] }
+icmp-socket2 = { version = "0.6.4", features = ["tokio"] }
+irtt-client = { version = "0.6.0", optional = true, features = ["ancillary", "tokio"] }
+libc = "0.2"
+rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
 serde = { version = "1", features = ["derive"] }
-serde_json = { version = "1", optional = true }
+thiserror = "2.0"
+tokio = { version = "1", features = ["full"] }
 uci-derive = { path = "crates/uci-derive" }
 
 [dependencies.log]
@@ -41,14 +42,14 @@ netlink-bindings = { version = "0.3.3", features = ["rt-link", "tc"] }
 netlink-socket2 = { version = "0.3.3" }
 
 [target.'cfg(unix)'.dependencies]
-rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "fcfedd204fb76928ef617bf9631b041c816e03d4", optional = true, default-features = false, }
+rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "6835397f4f4ba8b69e75e727105186865c357417", optional = true, default-features = false, }
 
 [features]
-default = ["toml", "trace", "uci-vendored"]
+default = ["irtt", "toml", "uci-vendored"]
+irtt = ["dep:irtt-client"]
 toml = ["config/toml"]
 uci = ["dep:rust-uci"]
 uci-vendored = ["uci", "rust-uci/vendored"]
-trace = ["dep:serde_json"]
 
 [workspace]
 members = ["crates/uci-derive"]

Cross.toml

@@ -1,3 +1,40 @@
 [target.x86_64-unknown-freebsd]
 image.name = "ghcr.io/cross-rs/x86_64-unknown-freebsd:local"
 image.toolchain = ["linux/arm64=aarch64-unknown-linux-gnu"]
+
+[build]
+# Vendored UCI searches CMAKE_LIBRARY_PATH for libubox built outside the sysroot.
+pre-build = [
+    "apt-get update && apt-get install -y --no-install-recommends pkg-config",
+    'echo "set(CMAKE_FIND_ROOT_PATH_MODE_LIBRARY BOTH)" >> /opt/toolchain.cmake',
+    # Rust's musl std links -lunwind; MIPS cross GCC supplies the unwinder as libgcc_eh.a.
+    '''
+    for cc in /usr/local/bin/mips*-gcc; do
+        [ -x "$cc" ] || continue
+        archive=$("$cc" -print-file-name=libgcc_eh.a)
+        test -f "$archive" || exit 1
+        ln -sf libgcc_eh.a "$(dirname "$archive")/libunwind.a"
+    done
+    ''',
+]
+
+# This target's image was added after the cross 0.2.5 release.
+[target.riscv64gc-unknown-linux-musl]
+image = "ghcr.io/cross-rs/riscv64gc-unknown-linux-musl:main"
+
+# MIPS images exist but are excluded from cross's automatic image selection.
+[target.mips-unknown-linux-musl]
+image = "ghcr.io/cross-rs/mips-unknown-linux-musl:main"
+build-std = ["std", "panic_abort"]
+
+[target.mipsel-unknown-linux-musl]
+image = "ghcr.io/cross-rs/mipsel-unknown-linux-musl:main"
+build-std = ["std", "panic_abort"]
+
+[target.mips64-unknown-linux-muslabi64]
+image = "ghcr.io/cross-rs/mips64-unknown-linux-muslabi64:main"
+build-std = ["std", "panic_abort"]
+
+[target.mips64el-unknown-linux-muslabi64]
+image = "ghcr.io/cross-rs/mips64el-unknown-linux-muslabi64:main"
+build-std = ["std", "panic_abort"]

README.md

@@ -10,14 +10,6 @@
 * **Configurable via environment or UCI**: reads settings from environment variables or OpenWRT UCI, with sensible defaults and type‑safe parsing .
 * **Minimal dependencies**: relies on Rust crates only (anyhow, socket2, neli, rustix), no external scripts .
 
-## Legacy control trace v1
-
-The additive legacy characterization recorder is excluded from default builds. Build with
-`--features trace` and set `SQMA_TRACE_V1_PATH` to a JSON Lines output path to enable it at
-runtime. The format is versioned as trace v1 and is intended only for replaying the current
-baseliner and rate-controller boundary during the `SqmControlLaw` extraction; it is not the
-final runtime event schema. Existing CSV and metrics outputs remain unchanged.
-
 ## Installation
 
 You can install **sqm‑autorate‑rust** in two ways:

crates/uci-derive/src/lib.rs

@@ -9,13 +9,13 @@ use syn::{Data, DeriveInput, Field, Fields, FieldsNamed, LitStr, Token, parse_ma
 
 /// Generates `UciSectionSchema` from a settings struct.
 ///
-/// Every named field is mapped one-to-one:
+/// Every named field without `#[uci(skip)]` is mapped one-to-one:
 ///
 /// ```text
 /// Rust field name -> UCI option name
 /// Rust field name -> config-rs field name
 /// ```
-#[proc_macro_derive(UciSection)]
+#[proc_macro_derive(UciSection, attributes(uci))]
 pub fn derive_uci_section(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);
 
@@ -56,26 +56,28 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
     let struct_name = &input.ident;
     let fields = require_named_fields(input)?;
 
-    let mappings = fields
-        .named
-        .iter()
-        .map(|field| {
-            let field_ident = field
-                .ident
-                .as_ref()
-                .expect("named fields always have identifiers");
+    let mut mappings = Vec::with_capacity(fields.named.len());
 
-            let field_name = field_ident.unraw().to_string();
-            let field_name = LitStr::new(&field_name, field_ident.span());
+    for field in &fields.named {
+        if parse_skip(field)? {
+            continue;
+        }
 
-            quote! {
-                crate::settings::uci_schema::UciOptionMapping {
-                    uci_option: #field_name,
-                    config_field: #field_name,
-                }
+        let field_ident = field
+            .ident
+            .as_ref()
+            .expect("named fields always have identifiers");
+
+        let field_name = field_ident.unraw().to_string();
+        let field_name = LitStr::new(&field_name, field_ident.span());
+
+        mappings.push(quote! {
+            crate::settings::uci_schema::UciOptionMapping {
+                uci_option: #field_name,
+                config_field: #field_name,
             }
-        })
-        .collect::<Vec<_>>();
+        });
+    }
 
     let (impl_generics, type_generics, where_clause) = input.generics.split_for_impl();
 
@@ -95,6 +97,29 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
     })
 }
 
+fn parse_skip(field: &Field) -> syn::Result<bool> {
+    let mut skip = false;
+
+    for attribute in &field.attrs {
+        if !attribute.path().is_ident("uci") {
+            continue;
+        }
+
+        attribute.parse_nested_meta(|meta| {
+            if !meta.path.is_ident("skip") {
+                return Err(meta.error("expected `skip`"));
+            }
+            if skip {
+                return Err(meta.error("duplicate `skip` attribute"));
+            }
+            skip = true;
+            Ok(())
+        })?;
+    }
+
+    Ok(skip)
+}
+
 fn expand_uci_config(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStream> {
     let struct_name = &input.ident;
     let fields = require_named_fields(input)?;

deny.toml

@@ -0,0 +1,50 @@
+# cargo-deny policy for sqm-autorate-rust.
+# Keep published dependency sources and license terms explicit while allowing
+# advisory reports to be reviewed without blocking unrelated changes.
+
+[graph]
+all-features = true
+no-default-features = false
+targets = []
+
+[output]
+feature-depth = 1
+
+[advisories]
+yanked = "deny"
+unmaintained = "workspace"
+unsound = "workspace"
+ignore = []
+
+[licenses]
+confidence-threshold = 0.8
+allow = [
+    "Apache-2.0",
+    "BSD-2-Clause",
+    "BSD-3-Clause",
+    "CC0-1.0",
+    "ISC",
+    "MIT",
+    "MPL-2.0",
+    "Unicode-3.0",
+    "Zlib",
+]
+
+[bans]
+multiple-versions = "warn"
+wildcards = "allow"
+highlight = "all"
+workspace-default-features = "allow"
+external-default-features = "allow"
+
+[sources]
+unknown-registry = "deny"
+unknown-git = "deny"
+required-git-spec = "rev"
+allow-registry = ["https://github.com/rust-lang/crates.io-index"]
+allow-git = ["https://github.com/Lochnair/rust-uci"]
+
+[sources.allow-org]
+github = []
+gitlab = []
+bitbucket = []

renovate.json

@@ -2,5 +2,18 @@
   "$schema": "https://docs.renovatebot.com/renovate-schema.json",
   "extends": [
     "config:recommended"
+  ],
+  "packageRules": [
+    {
+      "description": "Keep netlink-bindings and netlink-socket2 updates together",
+      "matchManagers": [
+        "cargo"
+      ],
+      "matchPackageNames": [
+        "netlink-bindings",
+        "netlink-socket2"
+      ],
+      "groupName": "netlink dependencies"
+    }
   ]
-}
+}

src/baseliner.rs

@@ -6,10 +6,9 @@
 // SPDX-License-Identifier: MPL-2.0
 
 use crate::SHUTDOWN;
+use crate::measurement::model::{MeasurementEvent, MeasurementObservation};
 use crate::metrics::{Metric, MetricsSender};
-use crate::pinger::PingReply;
 use crate::settings::Settings;
-use crate::trace::{Recorder, v1};
 use flume::{Receiver, Sender};
 use log::{debug, info};
 use std::collections::HashMap;
@@ -30,10 +29,16 @@ pub struct EwmaStats {
 }
 
 impl EwmaStats {
-    fn from_reply(reply: &PingReply) -> Self {
+    fn from_reply(reply: &MeasurementObservation) -> Self {
         Self {
-            down: reply.down_time,
-            up: reply.up_time,
+            down: reply
+                .one_way
+                .map(|o| o.downlink.as_nanos() as f64 / 1000000.0)
+                .unwrap_or(reply.rtt_ms() / 2.0),
+            up: reply
+                .one_way
+                .map(|o| o.uplink.as_nanos() as f64 / 1000000.0)
+                .unwrap_or(reply.rtt_ms() / 2.0),
         }
     }
 
@@ -74,12 +79,12 @@ pub struct Baseliner {
     pub settings: Settings,
     pub reselect_trigger: Sender<bool>,
     pub start_time: Instant,
-    pub stats_rx: Receiver<PingReply>,
+    pub stats_rx: Receiver<MeasurementEvent>,
     pub control_tx: Sender<ControlSnapshot>,
     pub selection_tx: Option<Sender<ControlSnapshot>>,
     pub baseline_metrics: MetricsSender,
     pub event_metrics: MetricsSender,
-    pub trace: Recorder,
+    pub ping_metrics: MetricsSender,
 }
 
 fn ewma_factor(tick: f64, dur: f64) -> f64 {
@@ -117,19 +122,19 @@ fn update_reflector_state(
 
 fn process_reply(
     reflectors: &mut HashMap<IpAddr, ReflectorState>,
-    reply: &PingReply,
+    reply: &MeasurementObservation,
     start_time: Instant,
     slow_factor: f64,
     fast_factor: f64,
 ) -> (ReflectorState, bool) {
     let sample = EwmaStats::from_reply(reply);
     let state = reflectors
-        .entry(reply.reflector)
-        .or_insert_with(|| ReflectorState::new(sample, reply.last_receive_time_s));
+        .entry(reply.stream.peer)
+        .or_insert_with(|| ReflectorState::new(sample, reply.observed_at));
     let anomaly = update_reflector_state(
         state,
         sample,
-        reply.last_receive_time_s,
+        reply.observed_at,
         start_time,
         slow_factor,
         fast_factor,
@@ -167,8 +172,14 @@ impl Baseliner {
                 return Ok(());
             }
 
-            let reply = self.stats_rx.recv()?;
-            let reflector = reply.reflector;
+            let event = self.stats_rx.recv()?;
+
+            let reply = match event {
+                MeasurementEvent::Observation(reply) => reply,
+                MeasurementEvent::Loss(_) => continue,
+            };
+
+            let reflector = reply.stream.peer;
             let (state, anomaly) = process_reply(
                 &mut reflectors,
                 &reply,
@@ -191,6 +202,20 @@ impl Baseliner {
                 let _ = self.reselect_trigger.try_send(true);
             }
 
+            self.ping_metrics.send(Metric::Ping {
+                reflector,
+                measurement_type: reply.stream.source,
+                rtt: reply.rtt_ms(),
+                up_time: reply
+                    .one_way
+                    .map(|o| o.uplink.as_nanos() as f64 / 1000000.0)
+                    .unwrap_or(reply.rtt_ms() / 2.0),
+                down_time: reply
+                    .one_way
+                    .map(|o| o.downlink.as_nanos() as f64 / 1000000.0)
+                    .unwrap_or(reply.rtt_ms() / 2.0),
+            });
+
             self.baseline_metrics.send(Metric::Baseline {
                 reflector,
                 baseline_up_ewma: state.baseline.up,
@@ -209,23 +234,10 @@ impl Baseliner {
             );
 
             let snapshot = control_snapshot(&reflectors, Instant::now());
-            self.trace.linearize(|trace| {
-                if trace.is_enabled() {
-                    trace.record_at(
-                        reply.last_receive_time_s,
-                        v1::Event::PingReply {
-                            reflector: reply.reflector,
-                            down_time_ms: reply.down_time,
-                            up_time_ms: reply.up_time,
-                        },
-                    );
-                }
-
-                if let Some(selection_tx) = &self.selection_tx {
-                    let _ = selection_tx.send(snapshot.clone());
-                }
-                self.control_tx.send(snapshot)
-            })?;
+            if let Some(selection_tx) = &self.selection_tx {
+                let _ = selection_tx.send(snapshot.clone());
+            }
+            self.control_tx.send(snapshot)?;
         }
     }
 }
@@ -233,7 +245,10 @@ impl Baseliner {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::settings::MeasurementType;
+    use crate::measurement::model::{
+        MeasurementSource, MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency,
+        SignedDuration,
+    };
     use std::time::Duration;
 
     fn state(
@@ -260,19 +275,25 @@ mod tests {
         EwmaStats { down, up }
     }
 
-    fn reply(reflector: IpAddr, down_time: f64, up_time: f64, received_at: Instant) -> PingReply {
-        PingReply {
-            reflector,
-            measurement_type: MeasurementType::Icmp,
-            seq: 0,
-            rtt: down_time + up_time,
-            current_time: 0,
-            down_time,
-            up_time,
-            originate_timestamp: 0,
-            receive_timestamp: 0,
-            transmit_timestamp: 0,
-            last_receive_time_s: received_at,
+    fn reply(
+        reflector: IpAddr,
+        down_time: f64,
+        up_time: f64,
+        received_at: Instant,
+    ) -> MeasurementObservation {
+        MeasurementObservation {
+            stream: MeasurementStream {
+                id: MeasurementStreamId(0),
+                peer: reflector,
+                source: MeasurementSource::IcmpEcho,
+            },
+            started_at: received_at,
+            observed_at: received_at,
+            one_way: Some(OneWayLatency {
+                uplink: SignedDuration::from_millis(up_time as i64),
+                downlink: SignedDuration::from_millis(down_time as i64),
+                clock: OneWayClock::Uncalibrated,
+            }),
         }
     }
 

src/controller.rs

@@ -0,0 +1,102 @@
+use std::{collections::HashMap, net::IpAddr, time::Duration};
+
+use flume::{Receiver, SendError, Sender};
+use thiserror::Error;
+
+use crate::measurement::{
+    icmp::IcmpBinding,
+    model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule},
+};
+
+#[derive(Debug, Error)]
+pub enum ControllerError {
+    #[error("ICMP channel error")]
+    IcmpChannelError(#[from] SendError<Vec<IcmpBinding>>),
+}
+
+pub struct Controller {
+    reflector_rx: Receiver<Vec<IpAddr>>,
+    icmp_binding_tx: Sender<Vec<IcmpBinding>>,
+
+    source: MeasurementSource,
+    period: Duration,
+
+    next_stream_id: u64,
+    stream_ids: HashMap<(IpAddr, MeasurementSource), MeasurementStreamId>,
+}
+
+impl Controller {
+    pub fn new(
+        reflector_rx: Receiver<Vec<IpAddr>>,
+        icmp_binding_tx: Sender<Vec<IcmpBinding>>,
+    ) -> Self {
+        Self {
+            reflector_rx,
+            icmp_binding_tx,
+            source: MeasurementSource::IcmpTimestamp,
+            period: Duration::from_millis(500),
+            next_stream_id: 0,
+            stream_ids: HashMap::new(),
+        }
+    }
+
+    pub async fn run(mut self, initial_reflectors: Vec<IpAddr>) -> Result<(), ControllerError> {
+        self.update_reflectors(initial_reflectors).await?;
+
+        while let Ok(reflectors) = self.reflector_rx.recv_async().await {
+            self.update_reflectors(reflectors).await?;
+        }
+
+        Ok(())
+    }
+
+    fn stream_for(&mut self, peer: IpAddr) -> MeasurementStream {
+        let key = (peer, self.source);
+
+        let id = *self.stream_ids.entry(key).or_insert_with(|| {
+            let id = MeasurementStreamId(self.next_stream_id);
+            self.next_stream_id += 1;
+            id
+        });
+
+        MeasurementStream {
+            id,
+            peer,
+            source: self.source,
+        }
+    }
+
+    async fn update_reflectors(&mut self, reflectors: Vec<IpAddr>) -> Result<(), ControllerError> {
+        let count = reflectors.len();
+
+        let bindings = reflectors
+            .into_iter()
+            .enumerate()
+            .map(|(index, peer)| {
+                let stream = self.stream_for(peer);
+
+                IcmpBinding::new(
+                    stream,
+                    ProbeSchedule {
+                        period: self.period,
+                        offset: evenly_spaced_offset(self.period, index, count),
+                    },
+                )
+            })
+            .collect();
+
+        self.icmp_binding_tx
+            .send_async(bindings)
+            .await
+            .map_err(|e| ControllerError::IcmpChannelError(e))
+    }
+}
+
+fn evenly_spaced_offset(period: Duration, index: usize, count: usize) -> Duration {
+    let nanos = period.as_nanos() * index as u128 / count as u128;
+
+    Duration::new(
+        (nanos / 1_000_000_000) as u64,
+        (nanos % 1_000_000_000) as u32,
+    )
+}

src/log.rs

@@ -2,10 +2,9 @@
 //
 // SPDX-License-Identifier: MPL-2.0
 
-use log::{Level, LevelFilter, Metadata, Record, SetLoggerError};
-use rustix::time::ClockId;
+use log::{Level, Metadata, Record, SetLoggerError};
 
-use crate::time::Time;
+use crate::time::{ClockSample, Realtime};
 
 #[derive(Clone, Copy)]
 pub struct SimpleLogger {
@@ -19,7 +18,7 @@ impl log::Log for SimpleLogger {
 
     fn log(&self, record: &Record) {
         if self.enabled(record.metadata()) {
-            let now = Time::new(ClockId::Realtime);
+            let now = ClockSample::<Realtime>::now();
             println!(
                 "{} {:5} {}:{}: {}",
                 now.as_secs_f64(),
@@ -36,5 +35,5 @@ impl log::Log for SimpleLogger {
 
 pub fn init(level: Level) -> Result<(), SetLoggerError> {
     log::set_boxed_logger(Box::new(SimpleLogger { level }))
-        .map(|()| log::set_max_level(LevelFilter::Trace))
+        .map(|()| log::set_max_level(level.to_level_filter()))
 }

src/main.rs

@@ -8,50 +8,42 @@
 extern crate core;
 
 mod baseliner;
+mod controller;
 mod log;
+mod measurement;
 mod metrics;
-mod pinger;
-mod pinger_icmp;
-mod pinger_icmp_ts;
 mod platform;
 mod ratecontroller;
 mod reflector_selector;
 mod settings;
 mod time;
-mod trace;
 mod util;
 
 use crate::baseliner::Baseliner;
+use crate::controller::Controller;
+use crate::measurement::icmp::IcmpEngine;
+use crate::measurement::irtt::IrttEngine;
 use crate::metrics::{Metric, Metrics, MetricsSender};
-use crate::pinger::{InFlightProbeCache, PingListener, PingSender};
-use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender};
-use crate::pinger_icmp_ts::{PingerICMPTimestampListener, PingerICMPTimestampSender};
 use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations};
 use crate::ratecontroller::Ratecontroller;
 use crate::reflector_selector::ReflectorSelector;
-use crate::settings::MeasurementType;
 use crate::settings::Settings;
-use ::log::{debug, info, warn};
+use crate::util::RwLockExt;
+use ::log::{info, warn};
 use flume::RecvTimeoutError;
-use std::collections::HashMap;
 use std::net::IpAddr;
 use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
-use std::sync::{Arc, Mutex, RwLock};
-use std::thread::sleep;
+use std::sync::{Arc, RwLock};
+use std::thread;
+use std::thread::{JoinHandle, sleep};
 use std::time::{Duration, Instant};
-use std::{process, thread};
 
 struct ReflectorSetup {
     peers: Arc<RwLock<Vec<IpAddr>>>,
     pool: Vec<IpAddr>,
     reselection_enabled: bool,
-    active_count: usize,
 }
-
-type PingListenerBox = Box<dyn PingListener + Send>;
-type PingSenderBox = Box<dyn PingSender + Send>;
-
 pub static SHUTDOWN: AtomicBool = AtomicBool::new(false);
 
 extern "C" fn signal_handler(_: libc::c_int) {
@@ -59,47 +51,6 @@ extern "C" fn signal_handler(_: libc::c_int) {
 }
 
 const VERSION: &str = env!("CARGO_PKG_VERSION");
-const RESELECTION_CANDIDATE_BURST: usize = 20;
-const INFLIGHT_CAPACITY_DUPLICATE_FACTOR: usize = 2;
-const INFLIGHT_CAPACITY_MIN: usize = 256;
-
-fn compute_inflight_probe_capacity(
-    settings: &Settings,
-    active_reflector_count: usize,
-    reselection_enabled: bool,
-) -> usize {
-    let tick_interval_ms = (settings.advanced_settings.tick_interval * 1000.0).max(1.0);
-    let expected_path_delay_ms = (settings.advanced_settings.download_delay_ms
-        + settings.advanced_settings.upload_delay_ms)
-        .max(10.0);
-    // Keep enough room for severe queueing events while still adapting to configured delay budgets.
-    let max_rtt_ms = (expected_path_delay_ms * 20.0).clamp(1000.0, 10000.0);
-
-    let burst_reflector_count = if reselection_enabled {
-        active_reflector_count + RESELECTION_CANDIDATE_BURST
-    } else {
-        active_reflector_count
-    };
-    let probes_per_reflector = (max_rtt_ms / tick_interval_ms).ceil() as usize;
-
-    (burst_reflector_count
-        .saturating_mul(probes_per_reflector)
-        .saturating_mul(INFLIGHT_CAPACITY_DUPLICATE_FACTOR))
-    .max(INFLIGHT_CAPACITY_MIN)
-}
-
-fn create_pinger(measurement_type: MeasurementType) -> (PingListenerBox, PingSenderBox) {
-    match measurement_type {
-        MeasurementType::Icmp => (
-            Box::new(PingerICMPEchoListener {}),
-            Box::new(PingerICMPEchoSender {}),
-        ),
-        MeasurementType::IcmpTimestamps => (
-            Box::new(PingerICMPTimestampListener {}),
-            Box::new(PingerICMPTimestampSender {}),
-        ),
-    }
-}
 
 fn initialize_shaper<T: TrafficControlBackend>(
     settings: &Settings,
@@ -156,11 +107,6 @@ fn install_signal_handlers() {
     }
 }
 
-fn probe_identifier() -> u16 {
-    // The ICMP identifier is only two bytes.
-    (process::id() & 0xffff) as u16
-}
-
 fn restore_shaper<T: TrafficControlBackend>(
     settings: &Settings,
     traffic_control: &mut T,
@@ -202,7 +148,7 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
         IpAddr::from_str("94.140.14.14")?,
     ];
 
-    let reselection_enabled = configured_count > settings.advanced_settings.num_reflectors as usize;
+    let reselection_enabled = configured_count > settings.reflectors.num_reflectors as usize;
     let pool = if reselection_enabled {
         configured
     } else {
@@ -213,17 +159,52 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
         peers: Arc::new(RwLock::new(default_reflectors.to_vec())),
         pool,
         reselection_enabled,
-        active_count: default_reflectors
-            .len()
-            .max(settings.advanced_settings.num_reflectors as usize),
     })
 }
 
+fn spawn_task<F, E>(
+    rt: &tokio::runtime::Runtime,
+    error_tx: &flume::Sender<anyhow::Error>,
+    task: F,
+) -> tokio::task::JoinHandle<()>
+where
+    F: Future<Output = Result<(), E>> + Send + 'static,
+    E: Into<anyhow::Error> + Send + 'static,
+{
+    let error_tx = error_tx.clone();
+
+    rt.spawn(async move {
+        if let Err(error) = task.await {
+            let _ = error_tx.send(error.into());
+        }
+    })
+}
+
+fn spawn_worker<F>(
+    name: &str,
+    error_tx: &flume::Sender<anyhow::Error>,
+    worker: F,
+) -> anyhow::Result<JoinHandle<()>>
+where
+    F: FnOnce() -> anyhow::Result<()> + Send + 'static,
+{
+    let error_tx = error_tx.clone();
+
+    thread::Builder::new()
+        .name(name.to_owned())
+        .spawn(move || {
+            if let Err(error) = worker() {
+                let _ = error_tx.send(error);
+            }
+        })
+        .map_err(|e| anyhow::anyhow!(e))
+}
+
 fn wait_for_exit(error_rx: &flume::Receiver<anyhow::Error>) -> anyhow::Result<()> {
     loop {
         match error_rx.recv_timeout(Duration::from_secs(1)) {
             Ok(error) => {
-                return Err(anyhow::anyhow!("thread exited with error: {error}"));
+                return Err(anyhow::anyhow!("worker exited with error: {error}"));
             }
             Err(RecvTimeoutError::Disconnected) => {
                 return Ok(());
@@ -239,31 +220,51 @@ fn wait_for_exit(error_rx: &flume::Receiver<anyhow::Error>) -> anyhow::Result<()
 }
 
 fn run(settings: &Settings) -> anyhow::Result<()> {
+    let rt = tokio::runtime::Runtime::new()?;
+
     if settings.advanced_settings.dry_run {
         info!("*** MONITORING MODE ACTIVE — qdisc rates will NOT be changed ***");
     }
 
     let start_time = Instant::now();
-    let trace = trace::Recorder::from_env(start_time, settings);
-    let probe_id = probe_identifier();
 
     let ReflectorSetup {
         peers: reflector_peers,
         pool: reflector_pool,
         reselection_enabled,
-        active_count: active_reflector_count,
     } = setup_reflectors(settings)?;
 
-    let (baseliner_stats_tx, baseliner_stats_rx) = flume::unbounded();
     let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded();
+    let (error_tx, error_rx) = flume::unbounded::<anyhow::Error>();
+    let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded();
+    let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded();
+    let (measurement_tx, measurement_rx) = flume::unbounded();
+    let (reflector_tx, reflector_rx) = flume::unbounded();
+    let (reselect_tx, reselect_rx) = flume::bounded(1);
     let (selection_snapshot_tx, selection_snapshot_rx) = if reselection_enabled {
         let (tx, rx) = flume::unbounded();
         (Some(tx), Some(rx))
     } else {
         (None, None)
     };
-    let (error_tx, error_rx) = flume::unbounded::<anyhow::Error>();
-    let (reselect_tx, reselect_rx) = flume::bounded(1);
+
+    let controller = Controller::new(reflector_rx, icmp_binding_tx);
+    let icmp_engine = {
+        let _guard = rt.enter();
+        IcmpEngine::new(
+            start_time,
+            &settings.network,
+            measurement_tx.clone(),
+            icmp_binding_rx,
+        )?
+    };
+    let (_irtt_task, irtt_engine) = {
+        let _guard = rt.enter();
+        IrttEngine::new(start_time, measurement_tx, irtt_binding_rx)?
+    };
+
+    spawn_task(&rt, &error_tx, icmp_engine.run());
+    spawn_task(&rt, &error_tx, irtt_engine.run());
 
     let dropped = Arc::new(AtomicU32::new(0));
 
@@ -274,14 +275,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
             metrics_rx: rx,
             metrics_dropped: Arc::clone(&dropped),
         };
-        let err_tx = error_tx.clone();
-        let handle = thread::Builder::new()
-            .name("metrics".to_string())
-            .spawn(move || {
-                if let Err(error) = metrics.run() {
-                    let _ = err_tx.send(error);
-                }
-            })?;
+        let handle = spawn_worker("metrics", &error_tx, move || metrics.run())?;
         (Some(tx), Some(handle))
     } else {
         (None, None)
@@ -311,83 +305,27 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         },
     });
 
-    let (mut ping_listener, mut ping_sender) =
-        create_pinger(settings.advanced_settings.measurement_type);
-
-    let inflight_probe_capacity =
-        compute_inflight_probe_capacity(settings, active_reflector_count, reselection_enabled);
-    info!(
-        "In-flight probe cache capacity: {} (active_reflectors={}, reselection_enabled={}, tick_interval_s={})",
-        inflight_probe_capacity,
-        active_reflector_count,
-        reselection_enabled,
-        settings.advanced_settings.tick_interval
-    );
-
-    let inflight: InFlightProbeCache =
-        Arc::new(Mutex::new(HashMap::with_capacity(inflight_probe_capacity)));
-
     let baseliner = Baseliner {
         settings: settings.clone(),
         reselect_trigger: reselect_tx.clone(),
         start_time,
-        stats_rx: baseliner_stats_rx,
+        stats_rx: measurement_rx,
         control_tx: control_snapshot_tx,
         selection_tx: selection_snapshot_tx,
         baseline_metrics,
         event_metrics: event_metrics.clone(),
-        trace: trace.clone(),
+        ping_metrics,
     };
 
     let mut main_traffic_control = traffic_control_backend();
     let (down_shaper, up_shaper) = initialize_shaper(settings, &mut main_traffic_control)?;
 
-    let err_tx = error_tx.clone();
-    let listener_peers = Arc::clone(&reflector_peers);
-    let listener_inflight = Arc::clone(&inflight);
-    let measurement_type = settings.advanced_settings.measurement_type;
-    thread::Builder::new()
-        .name("receiver".to_string())
-        .spawn(move || {
-            if let Err(error) = ping_listener.listen(
-                probe_id,
-                measurement_type,
-                listener_peers,
-                listener_inflight,
-                baseliner_stats_tx,
-                ping_metrics,
-            ) {
-                let _ = err_tx.send(error);
-            }
-        })?;
-
-    let err_tx = error_tx.clone();
-    thread::Builder::new()
-        .name("baseliner".to_string())
-        .spawn(move || {
-            if let Err(error) = baseliner.run() {
-                let _ = err_tx.send(error);
-            }
-        })?;
-
-    let err_tx = error_tx.clone();
-    let sender_peers = Arc::clone(&reflector_peers);
-    let sender_inflight = Arc::clone(&inflight);
-    let measurement_type = settings.advanced_settings.measurement_type;
-    let tick_interval = settings.advanced_settings.tick_interval;
-    thread::Builder::new()
-        .name("sender".to_string())
-        .spawn(move || {
-            if let Err(error) = ping_sender.send(
-                probe_id,
-                measurement_type,
-                sender_peers,
-                sender_inflight,
-                tick_interval,
-            ) {
-                let _ = err_tx.send(error);
-            }
-        })?;
+    spawn_worker("baseliner", &error_tx, move || baseliner.run())?;
+    spawn_task(
+        &rt,
+        &error_tx,
+        controller.run(reflector_peers.read_anyhow()?.clone()),
+    );
 
     let main_event_metrics = event_metrics.clone();
 
@@ -400,15 +338,9 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
             reflector_pool,
             trigger_channel: reselect_rx,
             metrics: event_metrics,
+            reflector_tx,
         };
-        let err_tx = error_tx.clone();
-        thread::Builder::new()
-            .name("reselection".to_string())
-            .spawn(move || {
-                if let Err(error) = reflector_selector.run() {
-                    let _ = err_tx.send(error);
-                }
-            })?;
+        spawn_worker("reselection", &error_tx, move || reflector_selector.run())?;
     }
 
     // Give the baseliner time to collect initial samples before adjusting rates.
@@ -420,17 +352,9 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         reflector_peers,
         reselect_tx,
         rate_metrics,
-        trace.clone(),
     )?;
 
-    let err_tx = error_tx.clone();
-    thread::Builder::new()
-        .name("ratecontroller".to_string())
-        .spawn(move || {
-            if let Err(error) = ratecontroller.run() {
-                let _ = err_tx.send(error);
-            }
-        })?;
+    spawn_worker("ratecontroller", &error_tx, move || ratecontroller.run())?;
 
     // Drop the original sender so the channel disconnects if all workers exit cleanly.
     drop(error_tx);
@@ -464,8 +388,6 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         &up_shaper,
     );
 
-    trace.finish();
-
     result
 }
 

src/measurement/icmp.rs

@@ -0,0 +1,505 @@
+use std::{
+    collections::{HashMap, HashSet},
+    io,
+    net::IpAddr,
+    time::{Duration, Instant, SystemTime},
+};
+
+use flume::{Receiver, Sender};
+use icmp_socket2::{
+    AsyncIcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet,
+    packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest},
+    tokio::AsyncIcmpV4Socket,
+};
+use log::debug;
+use thiserror::Error;
+
+use crate::{
+    measurement::model::{
+        MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementSource,
+        MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule,
+        SignedDuration,
+    },
+    settings::NetworkSettings,
+    time::{ClockSample, Realtime},
+};
+
+pub struct IcmpBinding {
+    pub stream: MeasurementStream,
+    pub schedule: ProbeSchedule,
+
+    next_probe_at: Instant,
+    next_sequence: u16,
+}
+
+impl IcmpBinding {
+    pub fn new(stream: MeasurementStream, schedule: ProbeSchedule) -> Self {
+        Self {
+            stream,
+            schedule,
+            next_probe_at: Instant::now(),
+            next_sequence: 0,
+        }
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum IcmpError {
+    #[error("I/O error")]
+    Io(#[from] io::Error),
+
+    #[error("failed to build ICMP packet")]
+    PacketBuild(#[from] IcmpPacketBuildError),
+
+    #[error("invalid ICMP packet: {0}")]
+    InvalidPacket(String),
+
+    #[error("unsupported measurement source: {0:?}")]
+    UnsupportedSource(MeasurementSource),
+
+    #[error("probe interval must be non-zero for stream {0:?}")]
+    InvalidInterval(MeasurementStreamId),
+
+    #[error("probe offset must be less than the period for stream {0:?}")]
+    InvalidOffset(MeasurementStreamId),
+
+    #[error(
+        "no matching probe found for peer {:?}, source: {:?}, sequence: {}",
+        .0.0,
+        .0.1,
+        .0.2
+    )]
+    NoMatchingProbe(InFlightProbeKey),
+
+    #[error("wrong identifier: expected {expected}, found {found}")]
+    WrongIdentifier { expected: u16, found: u16 },
+
+    #[error("unsupported ICMP peer address: {0}")]
+    UnsupportedPeer(IpAddr),
+
+    #[error("event channel closed")]
+    EventChannelClosed,
+
+    #[error("binding channel closed")]
+    BindingChannelClosed,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct InFlightProbe {
+    stream: MeasurementStream,
+    sent_at: Instant,
+}
+
+type InFlightProbeKey = (IpAddr, MeasurementSource, u16);
+type InFlightProbes = HashMap<InFlightProbeKey, InFlightProbe>;
+
+pub struct IcmpEngine {
+    socket: AsyncIcmpV4Socket,
+
+    identifier: u16,
+
+    bindings: HashMap<MeasurementStreamId, IcmpBinding>,
+    inflight: HashMap<InFlightProbeKey, InFlightProbe>,
+
+    origin: Instant,
+
+    binding_rx: Receiver<Vec<IcmpBinding>>,
+    observation_tx: Sender<MeasurementEvent>,
+}
+
+async fn wait_for_deadline(deadline: Option<Instant>) {
+    match deadline {
+        Some(deadline) => {
+            tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)).await;
+        }
+
+        None => {
+            std::future::pending::<()>().await;
+        }
+    }
+}
+
+impl IcmpEngine {
+    pub fn new(
+        origin: Instant,
+        #[allow(unused)] settings: &NetworkSettings,
+        observation_tx: Sender<MeasurementEvent>,
+        binding_rx: Receiver<Vec<IcmpBinding>>,
+    ) -> Result<Self, IcmpError> {
+        #[allow(unused_mut)]
+        let mut socket = IcmpSocket4::new()?;
+
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        if let Some(device) = &settings.measurement_bind_device {
+            icmp_socket2::IcmpSocket::bind_device(&mut socket, device)?;
+        }
+
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        if let Some(mark) = settings.measurement_mark {
+            icmp_socket2::IcmpSocket::set_mark(&mut socket, mark)?;
+        }
+
+        #[cfg(target_os = "freebsd")]
+        if let Some(fib) = settings.measurement_fib {
+            icmp_socket2::IcmpSocket::set_fib(&mut socket, fib)?;
+        }
+
+        socket.enable_receive_metadata()?;
+        let identifier = (std::process::id() & 0xffff) as u16;
+        let inflight: InFlightProbes = HashMap::new();
+
+        Ok(Self {
+            socket: socket.into_tokio()?,
+            identifier,
+            bindings: HashMap::new(),
+            inflight,
+            origin,
+            binding_rx,
+            observation_tx,
+        })
+    }
+
+    fn set_bindings(&mut self, desired: Vec<IcmpBinding>) -> Result<(), IcmpError> {
+        // Validate the complete desired state before mutating anything.
+        for binding in &desired {
+            if binding.schedule.period.is_zero() {
+                return Err(IcmpError::InvalidInterval(binding.stream.id));
+            }
+
+            if binding.schedule.offset >= binding.schedule.period {
+                return Err(IcmpError::InvalidOffset(binding.stream.id));
+            }
+
+            if !matches!(
+                binding.stream.source,
+                MeasurementSource::IcmpEcho | MeasurementSource::IcmpTimestamp
+            ) {
+                return Err(IcmpError::UnsupportedSource(binding.stream.source));
+            }
+        }
+
+        let now = Instant::now();
+        let desired_ids = desired
+            .iter()
+            .map(|binding| binding.stream.id)
+            .collect::<HashSet<_>>();
+
+        // Remove bindings that are no longer desired.
+        self.bindings.retain(|id, _| desired_ids.contains(id));
+        self.inflight
+            .retain(|_, probe| desired_ids.contains(&probe.stream.id));
+
+        for desired in desired {
+            match self.bindings.get_mut(&desired.stream.id) {
+                Some(current) if current.schedule == desired.schedule => {
+                    // Same schedule, so leave next_probe_at alone.
+                    current.stream = desired.stream;
+                }
+
+                Some(current) => {
+                    // Same stream, but its schedule changed. Rejoin the new grid.
+                    current.stream = desired.stream;
+                    current.schedule = desired.schedule;
+                    current.next_probe_at = desired.schedule.next_probe_at(self.origin, now);
+                }
+
+                None => {
+                    // New stream. Join its configured schedule at the next
+                    // occurrence after now.
+                    let mut binding = desired;
+                    binding.next_probe_at = binding.schedule.next_probe_at(self.origin, now);
+
+                    self.bindings.insert(binding.stream.id, binding);
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn send_due_probes(&mut self, now: Instant) -> Result<(), IcmpError> {
+        let due = self
+            .bindings
+            .iter()
+            .filter_map(|(&stream_id, binding)| (binding.next_probe_at <= now).then_some(stream_id))
+            .collect::<Vec<_>>();
+
+        for stream_id in due {
+            self.send_probe(stream_id).await?;
+
+            let binding = self
+                .bindings
+                .get_mut(&stream_id)
+                .expect("due binding disappeared");
+
+            binding.next_probe_at = binding.schedule.next_probe_at(self.origin, now);
+        }
+
+        Ok(())
+    }
+
+    async fn send_probe(&mut self, stream_id: MeasurementStreamId) -> Result<(), IcmpError> {
+        let (stream, sequence) = {
+            let binding = self
+                .bindings
+                .get_mut(&stream_id)
+                .expect("binding disappeared");
+
+            let sequence = binding.next_sequence;
+            binding.next_sequence = binding.next_sequence.wrapping_add(1);
+
+            (binding.stream, sequence)
+        };
+
+        let IpAddr::V4(peer) = stream.peer else {
+            return Err(IcmpError::UnsupportedPeer(stream.peer));
+        };
+
+        let packet = match stream.source {
+            MeasurementSource::IcmpEcho => {
+                Icmpv4Packet::with_echo_request(self.identifier, sequence, Vec::new())?
+            }
+            MeasurementSource::IcmpTimestamp => {
+                let originate = ClockSample::<Realtime>::now().as_time_since_midnight() as u32;
+                Icmpv4Packet::with_timestamp_request(self.identifier, sequence, originate, 0, 0)?
+            }
+            source => return Err(IcmpError::UnsupportedSource(source)),
+        };
+
+        let sent_at = Instant::now();
+        let key = (stream.peer, stream.source, sequence);
+
+        self.socket.send_to(peer, packet).await?;
+        self.inflight.insert(key, InFlightProbe { stream, sent_at });
+
+        Ok(())
+    }
+
+    fn next_deadline(&self) -> Option<Instant> {
+        self.bindings
+            .values()
+            .map(|binding| binding.next_probe_at)
+            .min()
+    }
+
+    fn handle_reply(
+        &mut self,
+        packet: Icmpv4Packet,
+        peer: IpAddr,
+        observed_at: Instant,
+        kernel_rx_timestamp: Option<SystemTime>,
+    ) -> Result<MeasurementObservation, IcmpError> {
+        match packet.message {
+            Icmpv4Message::EchoReply {
+                identifier,
+                sequence,
+                ..
+            } => {
+                if identifier != self.identifier {
+                    return Err(IcmpError::WrongIdentifier {
+                        expected: self.identifier,
+                        found: identifier,
+                    });
+                }
+
+                let key = (peer, MeasurementSource::IcmpEcho, sequence);
+                let probe = self
+                    .inflight
+                    .remove(&key)
+                    .ok_or(IcmpError::NoMatchingProbe(key))?;
+
+                debug!(
+                    "ICMP engine > Type: {:4}  | Reflector IP: {:>15}  | Seq: {:5}  | Current time: {:8}  | Originate: {:>8}  | Received time: {:>8}  | Transmit time: {:>8}  | RTT: {:8.3} ms  | UL time: {:>8}  | DL time: {:>8}",
+                    "ICMP",
+                    peer,
+                    sequence,
+                    ClockSample::<Realtime>::now().as_time_since_midnight(),
+                    "N/A",
+                    "N/A",
+                    "N/A",
+                    observed_at.duration_since(probe.sent_at).as_secs_f64() * 1_000.0,
+                    "N/A",
+                    "N/A",
+                );
+
+                Ok(MeasurementObservation {
+                    stream: probe.stream,
+                    started_at: probe.sent_at,
+                    observed_at,
+                    one_way: None,
+                })
+            }
+            Icmpv4Message::TimestampReply {
+                identifier,
+                sequence,
+                originate,
+                receive,
+                transmit,
+            } => {
+                if identifier != self.identifier {
+                    return Err(IcmpError::WrongIdentifier {
+                        expected: self.identifier,
+                        found: identifier,
+                    });
+                }
+
+                if originate >= MS_PER_DAY as u32
+                    || receive >= MS_PER_DAY as u32
+                    || transmit >= MS_PER_DAY as u32
+                {
+                    return Err(IcmpError::InvalidPacket(
+                        "non-standard RFC 792 timestamp".into(),
+                    ));
+                }
+
+                let key = (peer, MeasurementSource::IcmpTimestamp, sequence);
+                let probe = self
+                    .inflight
+                    .remove(&key)
+                    .ok_or(IcmpError::NoMatchingProbe(key))?;
+
+                let now_ms = if let Some(timestamp) = kernel_rx_timestamp {
+                    ClockSample::<Realtime>::from(timestamp).as_time_since_midnight()
+                } else {
+                    ClockSample::<Realtime>::now().as_time_since_midnight()
+                };
+
+                let uplink_ms = timestamp_delta(receive as i64, originate as i64);
+                let downlink_ms = timestamp_delta(now_ms, transmit as i64);
+
+                debug!(
+                    "ICMP engine > Type: {:4}  | Reflector IP: {:>15}  | Seq: {:5}  | Current time: {:8}  | Originate: {:8}  | Received time: {:8}  | Transmit time: {:8}  | RTT: {:8.3} ms  | UL time: {:5} ms  | DL time: {:5} ms",
+                    "ICMP",
+                    peer,
+                    sequence,
+                    now_ms,
+                    originate,
+                    receive,
+                    transmit,
+                    observed_at.duration_since(probe.sent_at).as_secs_f64() * 1_000.0,
+                    uplink_ms,
+                    downlink_ms,
+                );
+
+                Ok(MeasurementObservation {
+                    stream: probe.stream,
+                    started_at: probe.sent_at,
+                    observed_at,
+                    one_way: Some(OneWayLatency {
+                        uplink: SignedDuration::from_millis(uplink_ms),
+                        downlink: SignedDuration::from_millis(downlink_ms),
+                        clock: OneWayClock::Uncalibrated,
+                    }),
+                })
+            }
+            _ => Err(IcmpError::InvalidPacket(
+                "unsupported ICMP message type".to_string(),
+            )),
+        }
+    }
+
+    async fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> {
+        const PROBE_TIMEOUT: Duration = Duration::from_secs(30);
+
+        let expired = {
+            let mut expired = Vec::new();
+
+            self.inflight.retain(|_, probe| {
+                if now.saturating_duration_since(probe.sent_at) >= PROBE_TIMEOUT {
+                    expired.push(*probe);
+                    false
+                } else {
+                    true
+                }
+            });
+
+            expired
+        };
+
+        for probe in expired {
+            self.observation_tx
+                .send_async(MeasurementEvent::Loss(MeasurementLoss {
+                    stream: probe.stream,
+                    started_at: probe.sent_at,
+                    detected_at: now,
+                }))
+                .await
+                .map_err(|_| IcmpError::EventChannelClosed)?;
+        }
+
+        Ok(())
+    }
+
+    pub async fn run(mut self) -> Result<(), IcmpError> {
+        const HOUSEKEEPING_INTERVAL: Duration = Duration::from_millis(500);
+
+        let mut housekeeping = tokio::time::interval_at(
+            tokio::time::Instant::now() + HOUSEKEEPING_INTERVAL,
+            HOUSEKEEPING_INTERVAL,
+        );
+
+        housekeeping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+
+        loop {
+            let deadline = self.next_deadline();
+
+            tokio::select! {
+                bindings = self.binding_rx.recv_async() => {
+                    let bindings = bindings
+                        .map_err(|_| IcmpError::BindingChannelClosed)?;
+
+                    self.set_bindings(bindings)?;
+                }
+
+                _ = wait_for_deadline(deadline) => {
+                    self.send_due_probes(Instant::now()).await?;
+                }
+
+                result = self.socket.rcv_from_with_meta() => {
+                    let result = result?;
+
+                    match self.handle_reply(
+                        result.packet,
+                        result.peer.as_socket().unwrap().ip(),
+                        result.received_at,
+                        result.kernel_rx_timestamp,
+                    ) {
+                        Ok(observation) => {
+                            self.observation_tx
+                                .send_async(MeasurementEvent::Observation(observation))
+                                .await
+                                .map_err(|_| IcmpError::EventChannelClosed)?;
+                        }
+
+                        Err(
+                            IcmpError::WrongIdentifier { .. }
+                            | IcmpError::NoMatchingProbe { .. }
+                            | IcmpError::InvalidPacket(_),
+                        ) => {
+                            // Not ours, malformed, or late.
+                        }
+
+                        Err(err) => return Err(err),
+                    }
+                }
+
+                _ = housekeeping.tick() => {
+                    self.expire_inflight(Instant::now()).await?;
+                }
+            }
+        }
+    }
+}
+
+const MS_PER_DAY: i64 = 86_400_000;
+
+fn timestamp_delta(later: i64, earlier: i64) -> i64 {
+    let delta = later - earlier;
+    if delta < -MS_PER_DAY / 2 {
+        delta + MS_PER_DAY
+    } else if delta > MS_PER_DAY / 2 {
+        delta - MS_PER_DAY
+    } else {
+        delta
+    }
+}

src/measurement/irtt.rs

@@ -0,0 +1,343 @@
+use std::{collections::HashMap, net::SocketAddr, time::Instant};
+
+use flume::{Receiver, Sender};
+use irtt_client::{
+    ClientConfig, ClientEvent,
+    managed::{
+        ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask,
+        ManagedCommandAcknowledgement, ManagedCommandApplyError, ManagedCommandError,
+        ManagedCompletionPolicy, ManagedConfigError, ManagedEndReason, ManagedEvent,
+        ManagedLifecycle, ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason,
+        TargetId,
+    },
+};
+use log::{debug, trace, warn};
+use thiserror::Error;
+use tokio::sync::broadcast::error::RecvError;
+
+use crate::measurement::model::{
+    MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementStream,
+    MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule, SignedDuration,
+};
+
+#[derive(Debug, Error)]
+pub enum IrttError {
+    #[error("failed to build IRTT config")]
+    InvalidConfig(#[from] ManagedConfigError),
+
+    #[error("event channel closed")]
+    EventChannelClosed,
+
+    #[error("binding channel closed")]
+    BindingChannelClosed,
+
+    #[error("error while reciving on channel")]
+    ChannelRecvError(#[from] RecvError),
+
+    #[error("unable to subscribe to irtt events")]
+    SubscriptionError(#[from] ManagedSubscribeError),
+
+    #[error("failed to submit IRTT target update")]
+    CommandError(#[from] ManagedCommandError),
+
+    #[error("failed to apply IRTT target update")]
+    CommandApplyError(#[from] ManagedCommandApplyError),
+
+    #[error("IRTT managed client failed: {0:?}")]
+    ManagedFailure(ManagedEndReason),
+
+    #[error("IRTT managed task was abandoned")]
+    ManagedAbandoned,
+}
+
+pub struct IrttBinding {
+    stream: MeasurementStream,
+    schedule: ProbeSchedule,
+    port: u16,
+}
+
+impl IrttBinding {
+    pub fn new(stream: MeasurementStream, schedule: ProbeSchedule, port: u16) -> Self {
+        Self {
+            stream,
+            schedule,
+            port,
+        }
+    }
+}
+
+pub struct IrttEngine {
+    client: ManagedClientHandle,
+
+    bindings: HashMap<MeasurementStreamId, IrttBinding>,
+
+    origin: Instant,
+
+    binding_rx: Receiver<Vec<IrttBinding>>,
+    observation_tx: Sender<MeasurementEvent>,
+}
+
+impl IrttEngine {
+    pub fn new(
+        origin: Instant,
+        observation_tx: Sender<MeasurementEvent>,
+        binding_rx: Receiver<Vec<IrttBinding>>,
+    ) -> Result<(ManagedClientTask, Self), IrttError> {
+        let config = ManagedClientConfig {
+            client: ClientConfig {
+                request: irtt_client::SessionRequest {
+                    duration: None,
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+            completion: ManagedCompletionPolicy::ExplicitStop,
+            ..Default::default()
+        };
+        let (task, handle) = ManagedClient::task(config, Vec::new())?;
+
+        Ok((
+            task,
+            Self {
+                client: handle,
+                bindings: HashMap::new(),
+                origin,
+                binding_rx,
+                observation_tx,
+            },
+        ))
+    }
+
+    async fn set_bindings(&mut self, bindings: Vec<IrttBinding>) -> Result<(), IrttError> {
+        let targets = bindings
+            .iter()
+            .map(|binding| {
+                ManagedTargetConfig::new(
+                    TargetId::new(binding.stream.id.0.to_string()),
+                    SocketAddr::new(binding.stream.peer, binding.port).to_string(),
+                )
+            })
+            .collect();
+
+        self.client.update_targets(targets)?.await?;
+
+        self.bindings = bindings
+            .into_iter()
+            .map(|binding| (binding.stream.id, binding))
+            .collect();
+
+        Ok(())
+    }
+
+    pub async fn run(mut self) -> Result<(), IrttError> {
+        let mut events = self.client.subscribe()?;
+
+        loop {
+            tokio::select! {
+                bindings = self.binding_rx.recv_async() => {
+                    let bindings = bindings
+                        .map_err(|_| IrttError::BindingChannelClosed)?;
+
+                    self.set_bindings(bindings).await?;
+                }
+
+                event = events.recv() => {
+                    let event = match event {
+                        Ok(event) => event,
+
+                        Err(RecvError::Lagged(n)) => {
+                            warn!("IRTT event receiver dropped {n} events");
+                            continue;
+                        }
+
+                        Err(RecvError::Closed) => {
+                            let status = self.client.status();
+
+                            match status.lifecycle {
+                                ManagedLifecycle::Completed => return Ok(()),
+
+                                ManagedLifecycle::Failed => {
+                                    if let Some(outcome) = &status.final_outcome {
+                                        return Err(IrttError::ManagedFailure(
+                                            outcome.end_reason.clone()
+                                        ));
+                                    }
+
+                                    return Err(IrttError::ManagedAbandoned);
+                                }
+
+                                _ => return Err(IrttError::ManagedAbandoned),
+                            }
+                        }
+                    };
+
+                    match event {
+                        ManagedEvent::Started => {
+                            debug!("IRTT managed client started");
+                        }
+
+                        ManagedEvent::TargetStateChanged {
+                            target,
+                            lifecycle,
+                        } => {
+                            debug!("IRTT target {target:?}: {lifecycle:?}");
+                        }
+
+                        ManagedEvent::Client { target, event } => {
+                            let stream = target.id.as_str()
+                                .parse::<u64>()
+                                .ok()
+                                .and_then(|id| {
+                                    self.bindings.get(&MeasurementStreamId(id))
+                                })
+                                .map(|binding| binding.stream);
+
+                            match event {
+                                ClientEvent::SessionStarted(session) => {
+                                    debug!(
+                                        "IRTT session started for {}: {}",
+                                        target.id,
+                                        session.remote
+                                    );
+                                }
+
+                                ClientEvent::NoTestCompleted(outcome) => {
+                                    debug!(
+                                        "IRTT no-test completed for {}: {}",
+                                        target.id,
+                                        outcome.remote
+                                    );
+                                }
+
+                                ClientEvent::SessionClosed { .. } => {
+                                    debug!("IRTT session closed for {}", target.id);
+                                }
+
+                                ClientEvent::EchoSent { seq, .. } => {
+                                    trace!("IRTT {} sent probe {seq}", target.id);
+                                }
+
+                                ClientEvent::EchoReply {
+                                    sent_at,
+                                    received_at,
+                                    one_way,
+                                    ..
+                                } => {
+                                    let Some(stream) = stream else {
+                                        continue;
+                                    };
+
+                                    let observation = MeasurementObservation {
+                                        stream,
+                                        started_at: sent_at.mono,
+                                        observed_at: received_at.mono,
+                                        one_way: one_way.and_then(map_one_way),
+                                    };
+
+                                    self.observation_tx
+                                        .send_async(MeasurementEvent::Observation(observation))
+                                        .await
+                                        .map_err(|_| IrttError::EventChannelClosed)?;
+                                }
+
+                                ClientEvent::EchoLoss {
+                                    sent_at,
+                                    timeout_at,
+                                    ..
+                                } => {
+                                    let Some(stream) = stream else {
+                                        continue;
+                                    };
+
+                                    let loss = MeasurementLoss {
+                                        stream,
+                                        started_at: sent_at.mono,
+                                        detected_at: timeout_at,
+                                    };
+
+                                    self.observation_tx
+                                        .send_async(MeasurementEvent::Loss(loss))
+                                        .await
+                                        .map_err(|_| IrttError::EventChannelClosed)?;
+                                }
+
+                                ClientEvent::DuplicateReply { seq, .. } => {
+                                    trace!(
+                                        "IRTT {} duplicate reply {seq}",
+                                        target.id
+                                    );
+                                }
+
+                                ClientEvent::LateReply { seq, .. } => {
+                                    trace!(
+                                        "IRTT {} late reply {seq}",
+                                        target.id
+                                    );
+                                }
+
+                                ClientEvent::Warning { kind, message, .. } => {
+                                    debug!(
+                                        "IRTT {} warning ({kind:?}): {message}",
+                                        target.id
+                                    );
+                                }
+                            }
+                        }
+
+                        ManagedEvent::TargetFinished { outcome } => {
+                            match &outcome.end_reason {
+                                ManagedTargetEndReason::Failed(failure) => {
+                                    warn!(
+                                        "IRTT target {} failed: {failure:?}",
+                                        outcome.target.id
+                                    );
+                                }
+
+                                reason => {
+                                    debug!(
+                                        "IRTT target {} finished: {reason:?}",
+                                        outcome.target.id
+                                    );
+                                }
+                            }
+                        }
+
+                        ManagedEvent::Stopping => {
+                            debug!("IRTT managed client stopping");
+                        }
+
+                        ManagedEvent::Completed { outcome } => {
+                            debug!(
+                                "IRTT managed client completed: {:?}",
+                                outcome.end_reason
+                            );
+                            return Ok(());
+                        }
+
+                        ManagedEvent::Failed { outcome } => {
+                            return Err(IrttError::ManagedFailure(
+                                outcome.end_reason.clone()
+                            ));
+                        }
+
+                        ManagedEvent::Abandoned => {
+                            return Err(IrttError::ManagedAbandoned);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+fn map_one_way(sample: irtt_client::OneWayDelaySample) -> Option<OneWayLatency> {
+    let uplink = i64::try_from(sample.client_to_server?.as_nanos()).ok()?;
+
+    let downlink = i64::try_from(sample.server_to_client?.as_nanos()).ok()?;
+
+    Some(OneWayLatency {
+        uplink: SignedDuration::from_nanos(uplink),
+        downlink: SignedDuration::from_nanos(downlink),
+        clock: OneWayClock::Uncalibrated,
+    })
+}

src/measurement/mod.rs

@@ -0,0 +1,4 @@
+pub(crate) mod icmp;
+#[cfg(feature = "irtt")]
+pub(crate) mod irtt;
+pub(crate) mod model;

src/measurement/model.rs

@@ -0,0 +1,149 @@
+use std::fmt::{self, Display};
+use std::{
+    net::IpAddr,
+    time::{Duration, Instant},
+};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct MeasurementStreamId(pub u64);
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum MeasurementSource {
+    IcmpEcho,
+    IcmpTimestamp,
+    Irtt,
+    TcpSeqAck,
+    TcpTimestamp,
+}
+
+impl Display for MeasurementSource {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            MeasurementSource::IcmpEcho => write!(f, "icmp"),
+            MeasurementSource::IcmpTimestamp => write!(f, "icmp-timestamps"),
+            MeasurementSource::Irtt => write!(f, "irtt"),
+            MeasurementSource::TcpSeqAck => write!(f, "tcp-seqack"),
+            MeasurementSource::TcpTimestamp => write!(f, "tcp-timestamp"),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct MeasurementStream {
+    pub id: MeasurementStreamId,
+    pub peer: IpAddr,
+    pub source: MeasurementSource,
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OneWayClock {
+    Uncalibrated,
+    Calibrated,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ProbeSchedule {
+    pub period: Duration,
+    pub offset: Duration,
+}
+
+impl ProbeSchedule {
+    pub(crate) fn next_probe_at(&self, origin: Instant, now: Instant) -> Instant {
+        const NANOS_PER_SEC: u128 = 1_000_000_000;
+
+        let first = origin + self.offset;
+
+        if now < first {
+            return first;
+        }
+
+        let elapsed_ns = now.duration_since(first).as_nanos();
+        let period_ns = self.period.as_nanos();
+        let remainder = elapsed_ns % period_ns;
+
+        let until_next = if remainder == 0 {
+            period_ns
+        } else {
+            period_ns - remainder
+        };
+
+        let until_next = Duration::new(
+            (until_next / NANOS_PER_SEC) as u64,
+            (until_next % NANOS_PER_SEC) as u32,
+        );
+
+        now + until_next
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct SignedDuration {
+    nanos: i64,
+}
+
+#[allow(dead_code)]
+impl SignedDuration {
+    pub const fn from_nanos(nanos: i64) -> Self {
+        Self { nanos }
+    }
+
+    pub const fn from_millis(millis: i64) -> Self {
+        Self {
+            nanos: millis * 1_000_000,
+        }
+    }
+
+    pub const fn as_nanos(self) -> i64 {
+        self.nanos
+    }
+
+    pub fn as_secs_f64(self) -> f64 {
+        self.nanos as f64 / 1_000_000_000.0
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct OneWayLatency {
+    pub uplink: SignedDuration,
+    pub downlink: SignedDuration,
+    pub clock: OneWayClock,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct MeasurementObservation {
+    pub stream: MeasurementStream,
+
+    /// Local monotonic time at which measurement started.
+    pub started_at: Instant,
+
+    /// Local monotonic time at which the result was observed.
+    pub observed_at: Instant,
+
+    /// Directional timing, when the measurement mechanism can provide it.
+    pub one_way: Option<OneWayLatency>,
+}
+
+impl MeasurementObservation {
+    pub fn rtt(&self) -> Duration {
+        self.observed_at.duration_since(self.started_at)
+    }
+
+    pub fn rtt_ms(&self) -> f64 {
+        self.rtt().as_secs_f64() * 1000.0
+    }
+}
+
+#[allow(dead_code)]
+pub struct MeasurementLoss {
+    pub stream: MeasurementStream,
+    pub started_at: Instant,
+    pub detected_at: Instant,
+}
+
+#[allow(dead_code)]
+pub enum MeasurementEvent {
+    Observation(MeasurementObservation),
+    Loss(MeasurementLoss),
+}

src/metrics.rs

@@ -2,12 +2,12 @@
 //
 // SPDX-License-Identifier: MPL-2.0
 
+use crate::measurement::model::MeasurementSource;
+use crate::settings::ObservabilityProtocol;
 use crate::settings::Settings;
-use crate::settings::{MeasurementType, ObservabilityProtocol};
-use crate::time::Time;
+use crate::time::{ClockSample, Realtime};
 use flume::{Receiver, RecvTimeoutError, Sender};
 use log::{error, info, warn};
-use rustix::time::ClockId;
 use std::fmt::Write;
 use std::net::{IpAddr, TcpStream, UdpSocket};
 use std::sync::Arc;
@@ -94,7 +94,7 @@ impl Transport {
 pub enum Metric {
     Ping {
         reflector: IpAddr,
-        measurement_type: MeasurementType,
+        measurement_type: MeasurementSource,
         rtt: f64,
         up_time: f64,
         down_time: f64,
@@ -148,7 +148,7 @@ impl MetricsSender {
 
     pub fn send(&self, metric: Metric) {
         if let Some(ref tx) = self.tx {
-            let ts = Time::new(ClockId::Realtime).as_nanos();
+            let ts = ClockSample::<Realtime>::now().as_nanos();
             if tx.try_send((metric, ts)).is_err() {
                 self.dropped.fetch_add(1, Ordering::Relaxed);
             }
@@ -197,7 +197,7 @@ impl Metrics {
                 Err(RecvTimeoutError::Timeout) => {
                     let dropped = self.metrics_dropped.swap(0, Ordering::Relaxed);
                     if dropped > 0 {
-                        let ts = Time::new(ClockId::Realtime).as_nanos();
+                        let ts = ClockSample::<Realtime>::now().as_nanos();
                         batch.push((Metric::Dropped { count: dropped }, ts));
                     }
                     if !batch.is_empty() {

src/pinger.rs

@@ -1,214 +0,0 @@
-// SPDX-FileCopyrightText: 2022-Present Charles Corrigan mailto:chas-iot@runegate.org (github @chas-iot)
-// SPDX-FileCopyrightText: 2022-Present Daniel Lakeland mailto:dlakelan@street-artists.org (github @dlakelan)
-// SPDX-FileCopyrightText: 2022-Present Mark Baker mailto:mark@vpost.net (github @Fail-Safe)
-// SPDX-FileCopyrightText: 2022-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use crate::MeasurementType;
-use crate::SHUTDOWN;
-use crate::metrics::{Metric, MetricsSender};
-use crate::util::{MutexExt, RwLockExt};
-use flume::Sender;
-use icmp_socket2::socket::IcmpSocket;
-use icmp_socket2::{IcmpSocket4, Icmpv4Packet};
-use log::{debug, info, warn};
-use std::collections::HashMap;
-use std::net::{IpAddr, Ipv4Addr};
-use std::sync::atomic::Ordering;
-use std::sync::{Arc, Mutex, RwLock};
-use std::time::{Duration, Instant};
-use std::{io, thread};
-use thiserror::Error;
-
-#[derive(Error, Debug)]
-pub enum PingError {
-    #[error("Couldn't parse number")]
-    InvalidNumber(#[from] io::Error),
-    #[error("Error parsing packet")]
-    InvalidPacket(String),
-    #[error("Invalid packet type")]
-    InvalidType(String),
-    #[error("Wrong ICMP identifier (expected {expected:?}, found {found:?})")]
-    WrongID { expected: u16, found: u16 },
-}
-
-pub struct PingReply {
-    pub reflector: IpAddr,
-    pub measurement_type: MeasurementType,
-    pub seq: u16,
-    pub rtt: f64,
-    pub current_time: i64,
-    pub down_time: f64,
-    pub up_time: f64,
-    pub originate_timestamp: i64,
-    pub receive_timestamp: i64,
-    pub transmit_timestamp: i64,
-    pub last_receive_time_s: Instant,
-}
-
-#[derive(Clone, Copy, Debug)]
-pub struct InFlightProbe {
-    pub sent_at: Instant,
-    pub originate_timestamp: i64,
-}
-
-pub type InFlightProbeKey = (IpAddr, MeasurementType, u16);
-pub type InFlightProbeCache = Arc<Mutex<HashMap<InFlightProbeKey, InFlightProbe>>>;
-const INFLIGHT_PROBE_TTL: Duration = Duration::from_secs(30);
-
-fn open_socket(type_: MeasurementType) -> io::Result<IcmpSocket4> {
-    match type_ {
-        MeasurementType::Icmp | MeasurementType::IcmpTimestamps => IcmpSocket4::new(),
-    }
-}
-
-pub trait PingListener {
-    fn listen(
-        &mut self,
-        id: u16,
-        type_: MeasurementType,
-        reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
-        inflight: InFlightProbeCache,
-        stats_tx: Sender<PingReply>,
-        ping_metrics: MetricsSender,
-    ) -> anyhow::Result<()> {
-        let socket = &mut open_socket(type_)?;
-
-        loop {
-            if SHUTDOWN.load(Ordering::Relaxed) {
-                info!("Ping listener shutting down");
-                return Ok(());
-            }
-            let (pkt, sender) = match socket.rcv_from() {
-                Ok(val) => val,
-                Err(_) => continue,
-            };
-
-            let addr: IpAddr = sender.as_socket().unwrap().ip();
-
-            let reflectors = reflectors_lock.read_anyhow()?;
-            if !reflectors.contains(&addr) {
-                continue;
-            }
-
-            let reply = match self.parse_packet(id, addr, type_, pkt) {
-                Ok(val) => val,
-                Err(_) => {
-                    // parse_packet will throw an error if it's an unknown protocol etc.
-                    // so just quietly move on
-                    continue;
-                }
-            };
-
-            let key = (addr, type_, reply.seq);
-            let Some(probe) = inflight.lock_anyhow()?.remove(&key) else {
-                continue;
-            };
-            if reply.originate_timestamp != probe.originate_timestamp {
-                continue;
-            }
-
-            ping_metrics.send(Metric::Ping {
-                reflector: reply.reflector,
-                measurement_type: reply.measurement_type,
-                rtt: reply.rtt,
-                up_time: reply.up_time,
-                down_time: reply.down_time,
-            });
-
-            debug!(
-                "Type: {:4}  | Reflector IP: {:>15}  | Seq: {:5}  | Current time: {:8}  |  Originate: {:8}  |  Received time: {:8}  |  Transmit time : {:8}  |  RTT: {:8}  | UL time: {:5}  | DL time: {:5}",
-                "ICMP",
-                addr.to_string(),
-                reply.seq,
-                reply.current_time,
-                reply.originate_timestamp,
-                reply.receive_timestamp,
-                reply.transmit_timestamp,
-                reply.rtt,
-                reply.up_time,
-                reply.down_time
-            );
-            if let Err(e) = stats_tx.send(reply) {
-                warn!("Stats channel closed, stopping listener: {}", e);
-                break Ok(());
-            }
-        }
-    }
-
-    fn parse_packet(
-        &self,
-        id: u16,
-        reflector: IpAddr,
-        measurement_type: MeasurementType,
-        pkt: Icmpv4Packet,
-    ) -> Result<PingReply, PingError>;
-}
-
-pub trait PingSender {
-    fn send(
-        &mut self,
-        id: u16,
-        type_: MeasurementType,
-        reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
-        inflight: InFlightProbeCache,
-        tick_interval: f64,
-    ) -> anyhow::Result<()> {
-        let mut socket = open_socket(type_)?;
-
-        let mut seq: u16 = 0;
-        let tick_duration_ms: u16 = (tick_interval * 1000.0) as u16;
-
-        loop {
-            if SHUTDOWN.load(Ordering::Relaxed) {
-                info!("Ping sender shutting down");
-                return Ok(());
-            }
-            // Clone the reflectors vec and drop the read lock immediately —
-            // holding it for the entire tick would starve the reflector selector's write lock.
-            let reflectors_unlocked = reflectors_lock.read_anyhow()?;
-            let reflectors = reflectors_unlocked.clone();
-            drop(reflectors_unlocked);
-
-            if reflectors.is_empty() {
-                thread::sleep(Duration::from_millis(tick_duration_ms as u64));
-                continue;
-            }
-
-            inflight
-                .lock_anyhow()?
-                .retain(|_, probe| probe.sent_at.elapsed() < INFLIGHT_PROBE_TTL);
-
-            let sleep_duration =
-                Duration::from_millis((tick_duration_ms / reflectors.len() as u16) as u64);
-
-            for reflector in reflectors.iter() {
-                let addr: Ipv4Addr = match reflector {
-                    IpAddr::V4(ipv4) => *ipv4,
-                    IpAddr::V6(_) => unimplemented!(),
-                };
-
-                let (packet, originate_timestamp) = self.craft_packet(id, seq);
-                let sent_at = Instant::now();
-                let key = (*reflector, type_, seq);
-                inflight.lock_anyhow()?.insert(
-                    key,
-                    InFlightProbe {
-                        sent_at,
-                        originate_timestamp,
-                    },
-                );
-                if let Err(e) = socket.send_to(addr, packet) {
-                    inflight.lock_anyhow()?.remove(&key);
-                    return Err(e.into());
-                }
-                thread::sleep(sleep_duration);
-            }
-
-            seq = seq.wrapping_add(1);
-        }
-    }
-
-    fn craft_packet(&self, id: u16, seq: u16) -> (icmp_socket2::packet::Icmpv4Packet, i64);
-}

src/pinger_icmp.rs

@@ -1,97 +0,0 @@
-// SPDX-FileCopyrightText: 2022-Present Charles Corrigan mailto:chas-iot@runegate.org (github @chas-iot)
-// SPDX-FileCopyrightText: 2022-Present Daniel Lakeland mailto:dlakelan@street-artists.org (github @dlakelan)
-// SPDX-FileCopyrightText: 2022-Present Mark Baker mailto:mark@vpost.net (github @Fail-Safe)
-// SPDX-FileCopyrightText: 2022-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use std::net::IpAddr;
-use std::time::Instant;
-
-use crate::pinger::{PingError, PingListener, PingReply, PingSender};
-use crate::settings::MeasurementType;
-use crate::time::Time;
-use icmp_socket2::Icmpv4Message;
-use icmp_socket2::Icmpv4Packet;
-use icmp_socket2::packet::WithEchoRequest;
-use rustix::time::ClockId;
-
-pub struct PingerICMPEchoListener {}
-
-pub struct PingerICMPEchoSender {}
-
-impl PingListener for PingerICMPEchoListener {
-    // Result: RTT, down time, up time
-    fn parse_packet(
-        &self,
-        id: u16,
-        reflector: IpAddr,
-        measurement_type: MeasurementType,
-        pkt: Icmpv4Packet,
-    ) -> Result<PingReply, PingError> {
-        match pkt.typ {
-            0 => {
-                if let Icmpv4Message::EchoReply {
-                    identifier,
-                    sequence,
-                    payload,
-                } = pkt.message
-                {
-                    if identifier != id {
-                        return Err(PingError::WrongID {
-                            expected: id,
-                            found: identifier,
-                        });
-                    }
-
-                    let time_sent = match payload.as_slice().try_into() {
-                        Ok(bytes) => u64::from_be_bytes(bytes) as i64,
-                        Err(_) => {
-                            return Err(PingError::InvalidPacket(format!(
-                                "Expected 8 bytes payload, but found {}",
-                                payload.len()
-                            )));
-                        }
-                    };
-
-                    let clock = Time::new(ClockId::Monotonic);
-                    let time_ms = clock.to_milliseconds() as i64;
-
-                    let rtt = (time_ms - time_sent) as f64;
-                    Ok(PingReply {
-                        reflector,
-                        measurement_type,
-                        seq: sequence,
-                        rtt,
-                        current_time: time_ms,
-                        down_time: (rtt / 2.0) as f64,
-                        up_time: (rtt / 2.0) as f64,
-                        originate_timestamp: time_sent,
-                        receive_timestamp: 0,
-                        transmit_timestamp: 0,
-                        last_receive_time_s: Instant::now(),
-                    })
-                } else {
-                    Err(PingError::InvalidPacket(format!(
-                        "Packet had type {:?}, but did not match the structure",
-                        pkt.typ
-                    )))
-                }
-            }
-            type_ => Err(PingError::InvalidType(format!("{:?}", type_))),
-        }
-    }
-}
-
-impl PingSender for PingerICMPEchoSender {
-    fn craft_packet(&self, id: u16, seq: u16) -> (Icmpv4Packet, i64) {
-        let clock = Time::new(ClockId::Monotonic);
-        let time_ms = clock.to_milliseconds();
-        let payload = time_ms.to_be_bytes().to_vec();
-
-        (
-            Icmpv4Packet::with_echo_request(id, seq, payload).unwrap(),
-            time_ms as i64,
-        )
-    }
-}

src/pinger_icmp_ts.rs

@@ -1,90 +0,0 @@
-// SPDX-FileCopyrightText: 2022-Present Charles Corrigan mailto:chas-iot@runegate.org (github @chas-iot)
-// SPDX-FileCopyrightText: 2022-Present Daniel Lakeland mailto:dlakelan@street-artists.org (github @dlakelan)
-// SPDX-FileCopyrightText: 2022-Present Mark Baker mailto:mark@vpost.net (github @Fail-Safe)
-// SPDX-FileCopyrightText: 2022-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use crate::pinger::{PingError, PingListener, PingReply, PingSender};
-use crate::settings::MeasurementType;
-use crate::time::Time;
-use icmp_socket2::Icmpv4Message;
-use icmp_socket2::Icmpv4Packet;
-use icmp_socket2::packet::WithTimestampRequest;
-use rustix::time::ClockId;
-use std::net::IpAddr;
-use std::time::Instant;
-
-pub struct PingerICMPTimestampListener {}
-
-pub struct PingerICMPTimestampSender {}
-
-impl PingListener for PingerICMPTimestampListener {
-    // Result: RTT, down time, up time
-    fn parse_packet(
-        &self,
-        id: u16,
-        reflector: IpAddr,
-        measurement_type: MeasurementType,
-        pkt: Icmpv4Packet,
-    ) -> Result<PingReply, PingError> {
-        match pkt.typ {
-            // 14 = Timestamp reply
-            14 => {
-                if let Icmpv4Message::TimestampReply {
-                    identifier,
-                    sequence,
-                    originate,
-                    receive,
-                    transmit,
-                } = pkt.message
-                {
-                    if identifier != id {
-                        return Err(PingError::WrongID {
-                            expected: id,
-                            found: identifier,
-                        });
-                    }
-
-                    let time_now = Time::new(ClockId::Realtime);
-                    let time_since_midnight = time_now.get_time_since_midnight();
-
-                    let rtt = (time_since_midnight - originate as i64) as f64;
-                    let dl_time = (time_since_midnight - transmit as i64) as f64;
-                    let ul_time = (receive as i64 - originate as i64) as f64;
-
-                    Ok(PingReply {
-                        reflector,
-                        measurement_type,
-                        seq: sequence,
-                        rtt,
-                        current_time: time_since_midnight,
-                        down_time: dl_time,
-                        up_time: ul_time,
-                        originate_timestamp: originate as i64,
-                        receive_timestamp: receive as i64,
-                        transmit_timestamp: transmit as i64,
-                        last_receive_time_s: Instant::now(),
-                    })
-                } else {
-                    Err(PingError::InvalidPacket(format!(
-                        "Packet had type {:?}, but did not match the structure",
-                        pkt.typ
-                    )))
-                }
-            }
-            type_ => Err(PingError::InvalidType(format!("{:?}", type_))),
-        }
-    }
-}
-
-impl PingSender for PingerICMPTimestampSender {
-    fn craft_packet(&self, id: u16, seq: u16) -> (Icmpv4Packet, i64) {
-        let time_since_midnight = Time::new(ClockId::Realtime).get_time_since_midnight();
-        (
-            Icmpv4Packet::with_timestamp_request(id, seq, time_since_midnight as u32, 0, 0)
-                .unwrap(),
-            time_since_midnight as i64,
-        )
-    }
-}

src/platform/mod.rs

@@ -42,9 +42,6 @@ mod linux;
 #[cfg(target_os = "macos")]
 mod macos;
 
-#[cfg(unix)]
-pub mod unix;
-
 #[cfg(not(any(target_os = "freebsd", target_os = "linux", target_os = "macos")))]
 mod unsupported;
 

src/platform/unix/mod.rs

@@ -1,2 +0,0 @@
-#[cfg(feature = "uci")]
-pub mod uci;

src/ratecontroller.rs

@@ -13,12 +13,10 @@ use crate::platform::{
     interface_stats_provider, traffic_control_backend,
 };
 use crate::settings::Settings;
-use crate::time::Time;
-use crate::trace::{Recorder, v1};
+use crate::time::{ClockSample, Realtime};
 use crate::util::{ArcRwLock, RwLockExt};
 use flume::{Receiver, Sender};
 use log::{debug, info, warn};
-use rustix::time::ClockId;
 use std::fs::File;
 use std::io::Write;
 use std::net::IpAddr;
@@ -146,11 +144,9 @@ pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
     state_dl: State<T::Handle>,
     state_ul: State<T::Handle>,
     rng: fastrand::Rng,
-    next_evaluation_id: u64,
     stats_provider: S,
     traffic_control: T,
     metrics: MetricsSender,
-    trace: Recorder,
 }
 
 impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
@@ -163,7 +159,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         mut stats_provider: S,
         mut traffic_control: T,
         mut rng: fastrand::Rng,
-        trace: Recorder,
     ) -> anyhow::Result<Self> {
         let dl_shaper =
             traffic_control.find_shaper(settings.network.download_interface.as_str())?;
@@ -182,7 +177,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         let (cur_rx, cur_tx) =
             get_interface_stats(&mut stats_provider, &settings.network.upload_interface)?;
 
-        let controller = Self {
+        Ok(Self {
             settings,
             control_snapshot: None,
             control_snapshot_rx,
@@ -191,32 +186,10 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
             state_dl: State::new(dl_shaper, cur_rx, dl_safe_rates),
             state_ul: State::new(ul_shaper, cur_tx, ul_safe_rates),
             rng,
-            next_evaluation_id: 0,
             stats_provider,
             traffic_control,
             metrics,
-            trace,
-        };
-
-        if controller.trace.is_enabled() {
-            controller.trace.record_at(
-                Instant::now(),
-                v1::Event::ControllerInitialized {
-                    previous_rx_bytes: controller.state_dl.previous_bytes,
-                    previous_tx_bytes: controller.state_ul.previous_bytes,
-                    download_prev_mono_ns: controller
-                        .trace
-                        .monotonic_ns(controller.state_dl.prev_t),
-                    upload_prev_mono_ns: controller.trace.monotonic_ns(controller.state_ul.prev_t),
-                    download_history_index: controller.state_dl.nrate as u64,
-                    upload_history_index: controller.state_ul.nrate as u64,
-                    download_safe_rates_kbit: controller.state_dl.safe_rates.clone(),
-                    upload_safe_rates_kbit: controller.state_ul.safe_rates.clone(),
-                },
-            );
-        }
-
-        Ok(controller)
+        })
     }
 
     fn request_initial_rates(&mut self) -> anyhow::Result<()> {
@@ -224,20 +197,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         self.state_dl.current_rate = self.settings.network.download_base_kbits * 0.6;
         self.state_ul.current_rate = self.settings.network.upload_base_kbits * 0.6;
 
-        if self.trace.is_enabled() {
-            self.trace.record_at(
-                Instant::now(),
-                v1::Event::RequestedRates {
-                    evaluation_id: None,
-                    reason: v1::RateRequestReason::Startup,
-                    download_kbit: self.state_dl.current_rate.round() as u64,
-                    upload_kbit: self.state_ul.current_rate.round() as u64,
-                    download_requested: true,
-                    upload_requested: true,
-                },
-            );
-        }
-
         self.traffic_control.set_rate(
             &self.state_dl.shaper,
             self.state_dl.current_rate.round() as u64,
@@ -252,8 +211,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         Ok(())
     }
 
-    fn calculate_rate(&mut self, direction: Direction, evaluation_id: u64) -> anyhow::Result<()> {
-        let trace = self.trace.clone();
+    fn calculate_rate(&mut self, direction: Direction) -> anyhow::Result<()> {
         let (base_rate, delay_ms, min_rate, state) = if direction == Direction::Down {
             (
                 self.settings.network.download_base_kbits,
@@ -272,20 +230,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
 
         let now_t = Instant::now();
         let dur = now_t.duration_since(state.prev_t);
-        let trace_enabled = trace.is_enabled();
-        let trace_direction = match direction {
-            Direction::Down => v1::Direction::Download,
-            Direction::Up => v1::Direction::Upload,
-        };
-        if trace_enabled {
-            trace.record_at(
-                now_t,
-                v1::Event::RateCalculation {
-                    evaluation_id,
-                    direction: trace_direction,
-                },
-            );
-        }
 
         if !state.deltas.is_empty() {
             state.next_rate = state.current_rate;
@@ -324,20 +268,11 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                 }
 
                 if state.delta_stat > delay_ms {
-                    let selected_index = self.rng.usize(..state.safe_rates.len());
-                    match state.safe_rates.get(selected_index) {
+                    match state
+                        .safe_rates
+                        .get(self.rng.usize(..state.safe_rates.len()))
+                    {
                         Some(rnd_rate) => {
-                            if trace_enabled {
-                                trace.record_at(
-                                    Instant::now(),
-                                    v1::Event::RandomSafeRateChoice {
-                                        evaluation_id,
-                                        direction: trace_direction,
-                                        index: selected_index as u64,
-                                        rate_kbit: *rnd_rate,
-                                    },
-                                );
-                            }
                             state.next_rate = rnd_rate.min(0.9 * state.current_rate * state.load);
                         }
                         None => {
@@ -413,9 +348,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                     &mut self.stats_provider,
                     &self.settings.network.upload_interface,
                 )?;
-                let evaluation_id = self.next_evaluation_id;
-                self.next_evaluation_id = self.next_evaluation_id.wrapping_add(1);
-                self.update_deltas(evaluation_id, now_t)?;
+                self.update_deltas()?;
 
                 if self.state_dl.deltas.is_empty() || self.state_ul.deltas.is_empty() {
                     warn!("No reflector data available, dropping to minimum rates");
@@ -428,20 +361,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                     self.state_dl.next_rate = self.settings.network.download_min_kbits();
                     self.state_ul.next_rate = self.settings.network.upload_min_kbits();
 
-                    if self.trace.is_enabled() {
-                        self.trace.record_at(
-                            Instant::now(),
-                            v1::Event::RequestedRates {
-                                evaluation_id: Some(evaluation_id),
-                                reason: v1::RateRequestReason::NoReflectorData,
-                                download_kbit: self.state_dl.next_rate as u64,
-                                upload_kbit: self.state_ul.next_rate as u64,
-                                download_requested: true,
-                                upload_requested: true,
-                            },
-                        );
-                    }
-
                     self.traffic_control.set_rate(
                         &self.state_dl.shaper,
                         self.state_dl.next_rate as u64,
@@ -458,24 +377,8 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                     continue;
                 }
 
-                self.calculate_rate(Direction::Down, evaluation_id)?;
-                self.calculate_rate(Direction::Up, evaluation_id)?;
-
-                if self.trace.is_enabled() {
-                    let download_requested = self.state_dl.next_rate != self.state_dl.current_rate;
-                    let upload_requested = self.state_ul.next_rate != self.state_ul.current_rate;
-                    self.trace.record_at(
-                        Instant::now(),
-                        v1::Event::RequestedRates {
-                            evaluation_id: Some(evaluation_id),
-                            reason: v1::RateRequestReason::Control,
-                            download_kbit: self.state_dl.next_rate as u64,
-                            upload_kbit: self.state_ul.next_rate as u64,
-                            download_requested,
-                            upload_requested,
-                        },
-                    );
-                }
+                self.calculate_rate(Direction::Down)?;
+                self.calculate_rate(Direction::Up)?;
 
                 if self.state_dl.next_rate != self.state_dl.current_rate
                     || self.state_ul.next_rate != self.state_ul.current_rate
@@ -505,7 +408,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                 self.state_dl.current_rate = self.state_dl.next_rate;
                 self.state_ul.current_rate = self.state_ul.next_rate;
 
-                let stats_time = Time::new(ClockId::Realtime);
+                let stats_time = ClockSample::<Realtime>::now();
                 debug!(
                     "{},{},{},{},{},{},{},{}",
                     stats_time.secs(),
@@ -555,7 +458,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                 && now_t.duration_since(lastdump_t).as_secs_f64() > 300.0
             {
                 for i in 0..self.settings.advanced_settings.speed_hist_size as usize {
-                    let hist_time = Time::new(ClockId::Realtime);
+                    let hist_time = ClockSample::<Realtime>::now();
                     if let Err(e) = fd.write_all(
                         format!(
                             "{},{},{},{}\n",
@@ -575,36 +478,18 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         }
     }
 
-    fn update_deltas(&mut self, evaluation_id: u64, loop_time: Instant) -> anyhow::Result<()> {
-        let recorder = self.trace.clone();
-        recorder.linearize(|trace| -> anyhow::Result<()> {
-            drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot);
-            let now_t = Instant::now();
-            let reflectors = self.reflectors_lock.read_anyhow()?;
-            let (down_deltas, up_deltas) = deltas_from_snapshot(
-                self.control_snapshot.as_ref(),
-                &reflectors,
-                now_t,
-                self.settings.advanced_settings.tick_interval,
-            );
-            self.state_dl.deltas = down_deltas;
-            self.state_ul.deltas = up_deltas;
-
-            if trace.is_enabled() {
-                trace.record_at(
-                    now_t,
-                    v1::Event::ControlEvaluation {
-                        evaluation_id,
-                        loop_mono_ns: self.trace.monotonic_ns(loop_time),
-                        rx_bytes: self.state_dl.current_bytes,
-                        tx_bytes: self.state_ul.current_bytes,
-                        active_reflectors: reflectors.clone(),
-                    },
-                );
-            }
-
-            Ok(())
-        })?;
+    fn update_deltas(&mut self) -> anyhow::Result<()> {
+        drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot);
+        let now_t = Instant::now();
+        let reflectors = self.reflectors_lock.read_anyhow()?;
+        let (down_deltas, up_deltas) = deltas_from_snapshot(
+            self.control_snapshot.as_ref(),
+            &reflectors,
+            now_t,
+            self.settings.advanced_settings.tick_interval,
+        );
+        self.state_dl.deltas = down_deltas;
+        self.state_ul.deltas = up_deltas;
 
         if self.state_dl.deltas.len() < 5 || self.state_ul.deltas.len() < 5 {
             // trigger reselection
@@ -627,7 +512,6 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
         reflectors_lock: ArcRwLock<Vec<IpAddr>>,
         reselect_trigger: Sender<bool>,
         metrics: MetricsSender,
-        trace: Recorder,
     ) -> anyhow::Result<Self> {
         Self::new_with_backends(
             settings,
@@ -638,7 +522,6 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
             interface_stats_provider(),
             traffic_control_backend(),
             fastrand::Rng::new(),
-            trace,
         )
     }
 }

src/reflector_selector.rs

@@ -10,7 +10,7 @@ use crate::baseliner::ControlSnapshot;
 use crate::metrics::{Metric, MetricsSender};
 use crate::settings::Settings;
 use crate::util::RwLockExt;
-use flume::{Receiver, RecvError, RecvTimeoutError, Selector};
+use flume::{Receiver, RecvError, RecvTimeoutError, Selector, Sender};
 use log::{debug, info};
 use std::net::IpAddr;
 use std::sync::atomic::Ordering;
@@ -108,6 +108,7 @@ pub struct ReflectorSelector {
     pub settings: Settings,
     pub snapshot_rx: Receiver<ControlSnapshot>,
     pub reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
+    pub reflector_tx: Sender<Vec<IpAddr>>,
     pub reflector_pool: Vec<IpAddr>,
     pub trigger_channel: Receiver<bool>,
     pub metrics: MetricsSender,
@@ -158,10 +159,8 @@ impl ReflectorSelector {
 
             // After 40 reselections, slow down to every 15 minutes
             if reselection_count > 40 {
-                selector_sleep_time = Duration::new(
-                    self.settings.advanced_settings.peer_reselection_time * 60,
-                    0,
-                );
+                selector_sleep_time =
+                    Duration::new(self.settings.reflectors.peer_reselection_time * 60, 0);
             }
 
             let mut next_peers: Vec<IpAddr> = Vec::new();
@@ -186,6 +185,7 @@ impl ReflectorSelector {
             // Clone next_peers because we need it again after the baseline sleep
             // to iterate over candidates for RTT measurement.
             *reflectors_peers = next_peers.clone();
+            self.reflector_tx.send(next_peers.clone())?;
 
             // Drop the MutexGuard explicitly, as Rust won't unlock the mutex by default
             // until the guard goes out of scope
@@ -220,7 +220,7 @@ impl ReflectorSelector {
             candidates.sort_by(|a, b| a.1.cmp(&b.1));
 
             // Now we will just limit the candidates down to 2 * num_reflectors
-            let mut num_reflectors = self.settings.advanced_settings.num_reflectors;
+            let mut num_reflectors = self.settings.reflectors.num_reflectors;
             let candidate_pool_num = (2 * num_reflectors) as usize;
             candidates.truncate(candidate_pool_num);
 
@@ -251,6 +251,7 @@ impl ReflectorSelector {
                 });
             }
 
+            self.reflector_tx.send(new_peers.clone())?;
             *reflectors_peers = new_peers;
         }
     }

src/settings/mod.rs

@@ -11,11 +11,14 @@ use std::net::IpAddr;
 use std::str::FromStr;
 use uci_derive::{UciConfig, UciSection};
 
-pub(crate) mod uci_schema;
-mod validation;
+#[cfg(all(feature = "uci", unix))]
+use crate::settings::uci::UciSource;
 
 #[cfg(all(feature = "uci", unix))]
-use crate::platform::unix::uci::UciSource;
+pub mod uci;
+
+pub(crate) mod uci_schema;
+mod validation;
 
 struct FlexiBool(bool);
 
@@ -75,28 +78,32 @@ where
     deserializer.deserialize_any(FlexiBoolVisitor)
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub enum MeasurementType {
-    Icmp = 1,
+#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Hash)]
+#[serde(rename_all = "kebab-case")]
+pub enum ReflectorProtocol {
+    IcmpEcho = 1,
     IcmpTimestamps,
+    Irtt,
 }
 
-impl Display for MeasurementType {
+impl Display for ReflectorProtocol {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            MeasurementType::Icmp => write!(f, "icmp"),
-            MeasurementType::IcmpTimestamps => write!(f, "icmp-timestamps"),
+            ReflectorProtocol::IcmpEcho => write!(f, "icmp"),
+            ReflectorProtocol::IcmpTimestamps => write!(f, "icmp-timestamps"),
+            ReflectorProtocol::Irtt => write!(f, "irtt"),
         }
     }
 }
 
-impl FromStr for MeasurementType {
+impl FromStr for ReflectorProtocol {
     type Err = ConfigError;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s.to_lowercase().as_str() {
-            "icmp" => Ok(MeasurementType::Icmp),
-            "icmp-timestamps" | "icmp_timestamps" => Ok(MeasurementType::IcmpTimestamps),
+            "icmp" => Ok(ReflectorProtocol::IcmpEcho),
+            "icmp-timestamps" | "icmp_timestamps" => Ok(ReflectorProtocol::IcmpTimestamps),
+            "irtt" => Ok(ReflectorProtocol::Irtt),
             _ => Err(ConfigError::Message(format!(
                 "invalid measurement type: {s}"
             ))),
@@ -124,12 +131,13 @@ impl FromStr for ObservabilityProtocol {
     }
 }
 
-#[derive(Debug, Deserialize)]
-struct ReflectorRecord {
-    reflector_ip: IpAddr,
-    ip_version: u8,
-    #[allow(dead_code)]
-    description: String,
+#[derive(Debug, Clone, Deserialize, UciSection)]
+pub(crate) struct ReflectorDefinition {
+    pub(crate) host: String,
+    pub(crate) country: Option<String>,
+    pub(crate) protocols: Vec<ReflectorProtocol>,
+    pub(crate) irtt_port: Option<u16>,
+    pub(crate) irtt_hmac: Option<String>,
 }
 
 #[derive(Clone, Debug, Deserialize, UciConfig)]
@@ -138,6 +146,10 @@ pub(crate) struct Settings {
     #[uci(section)]
     pub(crate) network: NetworkSettings,
 
+    #[serde(default)]
+    #[uci(section)]
+    pub(crate) reflectors: ReflectorSettings,
+
     #[serde(default)]
     #[uci(section)]
     pub(crate) output: OutputSettings,
@@ -170,37 +182,8 @@ impl Settings {
     }
 
     pub fn load_reflectors(&self) -> anyhow::Result<Vec<IpAddr>> {
-        let mut reader = csv::ReaderBuilder::new()
-            .has_headers(true)
-            .trim(csv::Trim::All)
-            .from_path(&self.advanced_settings.reflector_list_file)
-            .with_context(|| {
-                format!(
-                    "failed to open reflector list: {}",
-                    self.advanced_settings.reflector_list_file
-                )
-            })?;
-
         let mut reflectors = Vec::with_capacity(50);
 
-        for record in reader.deserialize::<ReflectorRecord>() {
-            let record = record.with_context(|| {
-                format!(
-                    "failed to parse reflector list: {}",
-                    self.advanced_settings.reflector_list_file
-                )
-            })?;
-
-            match (record.ip_version, record.reflector_ip) {
-                (4, IpAddr::V4(_)) | (6, IpAddr::V6(_)) => {}
-                (version, address) => {
-                    bail!("reflector {address} has inconsistent IP version {version}");
-                }
-            }
-
-            reflectors.push(record.reflector_ip);
-        }
-
         Ok(reflectors)
     }
 }
@@ -215,6 +198,13 @@ pub(crate) struct NetworkSettings {
     pub(crate) upload_base_kbits: f64,
     #[serde(default = "default_min_percent")]
     pub(crate) upload_min_percent: f64,
+
+    #[allow(dead_code)]
+    pub(crate) measurement_bind_device: Option<String>,
+    #[allow(dead_code)]
+    pub(crate) measurement_mark: Option<u32>,
+    #[allow(dead_code)]
+    pub(crate) measurement_fib: Option<u32>,
 }
 
 impl NetworkSettings {
@@ -227,6 +217,35 @@ impl NetworkSettings {
     }
 }
 
+#[derive(Clone, Debug, Deserialize, UciSection)]
+#[serde(default)]
+pub(crate) struct ReflectorSettings {
+    pub(crate) files: Vec<String>,
+    pub(crate) countries: Vec<String>,
+
+    #[uci(skip)]
+    pub(crate) entries: Vec<ReflectorDefinition>,
+
+    #[uci(skip)]
+    pub(crate) bootstrap: Vec<ReflectorDefinition>,
+
+    pub(crate) num_reflectors: u8,
+    pub(crate) peer_reselection_time: u64,
+}
+
+impl Default for ReflectorSettings {
+    fn default() -> Self {
+        Self {
+            files: vec!["/etc/sqm-autorate-rust/reflectors.csv".into()],
+            countries: Vec::new(),
+            entries: Vec::new(),
+            bootstrap: Vec::new(),
+            num_reflectors: 5,
+            peer_reselection_time: 15,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Deserialize, UciSection)]
 #[serde(default)]
 pub(crate) struct OutputSettings {
@@ -296,10 +315,7 @@ pub(crate) struct AdvancedSettings {
     pub(crate) high_load_level: f64,
     pub(crate) min_change_interval: f64,
     #[serde(deserialize_with = "deserialize_from_str")]
-    pub(crate) measurement_type: MeasurementType,
-    pub(crate) num_reflectors: u8,
-    pub(crate) peer_reselection_time: u64,
-    pub(crate) reflector_list_file: String,
+    pub(crate) measurement_type: ReflectorProtocol,
     pub(crate) speed_hist_size: u32,
     pub(crate) tick_interval: f64,
     pub(crate) upload_delay_ms: f64,
@@ -313,10 +329,7 @@ impl Default for AdvancedSettings {
             download_delay_ms: 15.0,
             high_load_level: 0.8,
             min_change_interval: 0.5,
-            measurement_type: MeasurementType::IcmpTimestamps,
-            num_reflectors: 5,
-            peer_reselection_time: 15,
-            reflector_list_file: "/etc/sqm-autorate/reflectors-icmp.csv".to_owned(),
+            measurement_type: ReflectorProtocol::IcmpTimestamps,
             speed_hist_size: 100,
             tick_interval: 0.5,
             upload_delay_ms: 15.0,

src/settings/uci.rs

@@ -1,6 +1,6 @@
 use std::path::{Path, PathBuf};
 
-use crate::settings::uci_schema::{UciConfigSchema, UciSectionMapping};
+use crate::settings::uci_schema::{UciConfigSchema, UciRepeatedSectionMapping, UciSectionMapping};
 use config::{ConfigError as SourceError, Map, Source, Value, ValueKind};
 
 use rust_uci::config::{
@@ -18,6 +18,7 @@ pub(crate) struct UciSource {
     packages: PackageSelection,
     required: bool,
     schema: &'static [UciSectionMapping],
+    repeated_schema: &'static [UciRepeatedSectionMapping],
     config_dir: Option<PathBuf>,
     save_dir: Option<PathBuf>,
 }
@@ -28,6 +29,7 @@ impl UciSource {
             packages,
             required: true,
             schema: &[],
+            repeated_schema: &[],
             config_dir: None,
             save_dir: None,
         }
@@ -65,6 +67,17 @@ impl UciSource {
         self
     }
 
+    /// Registers section types collected as arrays of tables at their config paths.
+    /// Absent section types leave the destination unset for Serde/source defaults.
+    #[allow(dead_code)]
+    pub(crate) fn with_repeated_schema(
+        mut self,
+        schema: &'static [UciRepeatedSectionMapping],
+    ) -> Self {
+        self.repeated_schema = schema;
+        self
+    }
+
     #[allow(dead_code)]
     pub(crate) fn with_directories(
         mut self,
@@ -145,6 +158,7 @@ impl UciSource {
         package_name: &str,
         package: &UciPackage,
         values: &mut Map<String, Value>,
+        repeated_values: &mut Map<String, Vec<Value>>,
     ) -> Result<(), SourceError> {
         let sections = package.sections().map_err(|e| {
             SourceError::Message(format!(
@@ -154,6 +168,12 @@ impl UciSource {
         })?;
 
         for section in sections {
+            let repeated_mapping = self
+                .repeated_schema
+                .iter()
+                .find(|mapping| mapping.uci_section == section.type_());
+            let mut section_values = Map::new();
+
             let selector = section.selector().ok_or_else(|| {
                 SourceError::Message(format!(
                     "loaded UCI section of type {:?} in package \
@@ -181,9 +201,21 @@ impl UciSource {
 
                 let source_key = format!("{package_name}.{config_relative_key}");
 
-                let destination_key = self
-                    .mapped_config_key(&selector, option_name)
-                    .unwrap_or(source_key);
+                let repeated_option = repeated_mapping.and_then(|mapping| {
+                    mapping
+                        .options
+                        .iter()
+                        .find(|option| option.uci_option == option_name)
+                });
+
+                let destination_key = if let Some(mapping) = repeated_option {
+                    mapping.config_field.to_owned()
+                } else if repeated_mapping.is_some() {
+                    option_name.to_owned()
+                } else {
+                    self.mapped_config_key(&selector, option_name)
+                        .unwrap_or(source_key)
+                };
 
                 let Some(value) = option.get().map_err(|e| {
                     SourceError::Message(format!(
@@ -210,13 +242,27 @@ impl UciSource {
                     }
                 };
 
-                if values.insert(destination_key.clone(), value).is_some() {
+                let destination = if repeated_mapping.is_some() {
+                    &mut section_values
+                } else {
+                    &mut *values
+                };
+
+                if destination.insert(destination_key.clone(), value).is_some() {
                     return Err(SourceError::Message(format!(
                         "duplicate UCI destination key \
                          {destination_key:?} while mapping {uci_key:?}"
                     )));
                 }
             }
+
+            if let Some(mapping) = repeated_mapping {
+                let origin = format!("UCI section {package_name}.{selector}");
+                repeated_values
+                    .entry(mapping.config_path.to_owned())
+                    .or_default()
+                    .push(Value::new(Some(&origin), ValueKind::Table(section_values)));
+            }
         }
 
         Ok(())
@@ -262,7 +308,7 @@ impl Source for UciSource {
         let (_uci, packages) = match resolved {
             Ok(resolved) => resolved,
 
-            Err(_) if !self.required => {
+            Err(SourceError::NotFound(_)) if !self.required => {
                 return Ok(Map::new());
             }
 
@@ -272,6 +318,7 @@ impl Source for UciSource {
         };
 
         let mut values = Map::new();
+        let mut repeated_values = Map::new();
 
         for package in packages {
             let package_name = package
@@ -279,9 +326,50 @@ impl Source for UciSource {
                 .map_err(|e| SourceError::Message(format!("failed to read UCI package name: {e}")))?
                 .to_owned();
 
-            self.collect_package(&package_name, &package, &mut values)?;
+            self.collect_package(&package_name, &package, &mut values, &mut repeated_values)?;
+        }
+
+        for (config_path, sections) in repeated_values {
+            let origin = "UCI repeated sections".to_owned();
+            if values
+                .insert(
+                    config_path.clone(),
+                    Value::new(Some(&origin), ValueKind::Array(sections)),
+                )
+                .is_some()
+            {
+                return Err(SourceError::Message(format!(
+                    "duplicate UCI destination key {config_path:?} while mapping repeated sections"
+                )));
+            }
         }
 
         Ok(values)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::settings::Settings;
+    use std::fs;
+
+    #[test]
+    fn malformed_optional_package_is_an_error() {
+        let root = std::env::temp_dir().join(format!("sqma-uci-{}", std::process::id()));
+        let config_dir = root.join("config");
+        let save_dir = root.join("save");
+        fs::create_dir_all(&config_dir).unwrap();
+        fs::create_dir_all(&save_dir).unwrap();
+        let config_file = config_dir.join("sqm-autorate-rust");
+        fs::write(&config_file, "config network 'unterminated\n").unwrap();
+
+        let source = UciSource::from_schema::<Settings>()
+            .with_directories(&config_dir, &save_dir)
+            .required(false);
+        let result = source.collect();
+
+        fs::remove_dir_all(root).unwrap();
+        assert!(result.is_err());
+    }
+}

src/settings/uci_schema.rs

@@ -19,6 +19,18 @@ pub(crate) struct UciSectionMapping {
     pub(crate) options: &'static [UciOptionMapping],
 }
 
+#[cfg(all(feature = "uci", unix))]
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct UciRepeatedSectionMapping {
+    /// UCI section type, for both named and anonymous sections.
+    pub(crate) uci_section: &'static str,
+
+    /// Config-rs / Serde path receiving an array of section tables.
+    pub(crate) config_path: &'static str,
+
+    pub(crate) options: &'static [UciOptionMapping],
+}
+
 pub(crate) trait UciSectionSchema {
     const UCI_OPTIONS: &'static [UciOptionMapping];
 }

src/settings/validation.rs

@@ -3,6 +3,8 @@
 // SPDX-License-Identifier: MPL-2.0
 //
 
+use crate::settings::ReflectorSettings;
+
 use super::{AdvancedSettings, NetworkSettings, ObservabilitySettings, OutputSettings, Settings};
 use std::error::Error;
 use std::fmt::{self, Display, Formatter};
@@ -88,6 +90,7 @@ impl Settings {
         let mut report = ValidationReport::default();
 
         self.network.validate(&mut report);
+        self.reflectors.validate(&mut report);
         self.output.validate(&mut report);
         self.observability.validate(&mut report);
         self.advanced_settings.validate(&mut report);
@@ -153,7 +156,12 @@ impl NetworkSettings {
         if download_base_valid && download_percent_valid {
             let minimum = self.download_min_kbits();
 
-            if minimum < WARN_DOWNLOAD_MIN_KBITS {
+            if minimum < 1.0 {
+                report.error(
+                    "network.download_min_percent",
+                    "minimum download rate floors to zero",
+                );
+            } else if minimum < WARN_DOWNLOAD_MIN_KBITS {
                 report.warning(
                     "network.download_min_percent",
                     format!(
@@ -168,7 +176,12 @@ impl NetworkSettings {
         if upload_base_valid && upload_percent_valid {
             let minimum = self.upload_min_kbits();
 
-            if minimum < WARN_UPLOAD_MIN_KBITS {
+            if minimum < 1.0 {
+                report.error(
+                    "network.upload_min_percent",
+                    "minimum upload rate floors to zero",
+                );
+            } else if minimum < WARN_UPLOAD_MIN_KBITS {
                 report.warning(
                     "network.upload_min_percent",
                     format!(
@@ -182,6 +195,45 @@ impl NetworkSettings {
     }
 }
 
+impl ReflectorSettings {
+    fn validate(&self, report: &mut ValidationReport) {
+        if self.num_reflectors == 0 {
+            report.error("reflectors.num_reflectors", "must be greater than 0");
+        }
+
+        if self.peer_reselection_time == 0 {
+            report.error("reflectors.peer_reselection_time", "must be greater than 0");
+        }
+
+        if self.files.is_empty() && self.entries.is_empty() {
+            report.error(
+                "reflectors",
+                "at least one reflector file or inline entry is required",
+            );
+        }
+
+        for (index, file) in self.files.iter().enumerate() {
+            if file.trim().is_empty() {
+                report.error(
+                    "reflectors.files",
+                    format!("entry {index} must not be empty"),
+                );
+            }
+        }
+
+        for (index, country) in self.countries.iter().enumerate() {
+            if country.len() != 2 || !country.bytes().all(|c| c.is_ascii_alphabetic()) {
+                report.error(
+                    "reflectors.countries",
+                    format!(
+                        "entry {index} ({country:?}) must be an ISO 3166-1 alpha-2 country code"
+                    ),
+                );
+            }
+        }
+    }
+}
+
 impl OutputSettings {
     fn validate(&self, report: &mut ValidationReport) {
         if self.suppress_statistics {
@@ -288,24 +340,6 @@ impl AdvancedSettings {
             self.tick_interval,
         );
 
-        if self.num_reflectors < 5 {
-            report.error(
-                "advanced_settings.num_reflectors",
-                format!("must be at least 5, got {}", self.num_reflectors,),
-            );
-        }
-
-        if self.peer_reselection_time == 0 {
-            report.error(
-                "advanced_settings.peer_reselection_time",
-                "must be greater than 0",
-            );
-        }
-
-        if self.reflector_list_file.trim().is_empty() {
-            report.error("advanced_settings.reflector_list_file", "must not be empty");
-        }
-
         if self.speed_hist_size == 0 {
             report.error(
                 "advanced_settings.speed_hist_size",
@@ -376,3 +410,41 @@ fn validate_percent(report: &mut ValidationReport, path: &'static str, value: f6
         true
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn network(download_base_kbits: f64, upload_base_kbits: f64) -> NetworkSettings {
+        NetworkSettings {
+            download_interface: "ifb0".into(),
+            upload_interface: "eth0".into(),
+            download_base_kbits,
+            download_min_percent: 1.0,
+            upload_base_kbits,
+            upload_min_percent: 1.0,
+            measurement_bind_device: None,
+            measurement_mark: None,
+            measurement_fib: None,
+        }
+    }
+
+    #[test]
+    fn minimum_rates_that_floor_to_zero_are_errors_for_both_directions() {
+        let mut report = ValidationReport::default();
+        network(50.0, 50.0).validate(&mut report);
+        let error = report.finish().unwrap_err().to_string();
+
+        assert!(error.contains("network.download_min_percent"));
+        assert!(error.contains("network.upload_min_percent"));
+        assert!(error.contains("floors to zero"));
+    }
+
+    #[test]
+    fn representable_low_minimum_rates_remain_warnings() {
+        let mut report = ValidationReport::default();
+        network(10_000.0, 10_000.0).validate(&mut report);
+
+        assert!(report.finish().is_ok());
+    }
+}

src/time.rs

@@ -2,27 +2,45 @@
 //
 // SPDX-License-Identifier: MPL-2.0
 
-use rustix::fs::Timespec;
+use std::{
+    marker::PhantomData,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
 use rustix::time::{ClockId, clock_gettime};
 
-pub struct Time {
-    time_s: u64,
-    time_ns: u64,
+pub trait Clock {
+    const ID: ClockId;
 }
 
-impl Time {
-    pub fn new(id: ClockId) -> Self {
-        let time: Timespec = clock_gettime(id);
+pub enum Realtime {}
+pub enum Monotonic {}
+
+impl Clock for Realtime {
+    const ID: ClockId = ClockId::Realtime;
+}
+
+impl Clock for Monotonic {
+    const ID: ClockId = ClockId::Monotonic;
+}
+
+pub struct ClockSample<C> {
+    time_s: u64,
+    time_ns: u64,
+    _clock: PhantomData<C>,
+}
+
+impl<C: Clock> ClockSample<C> {
+    pub fn now() -> Self {
+        let time = clock_gettime(C::ID);
+
         Self {
             time_s: time.tv_sec as u64,
             time_ns: time.tv_nsec as u64,
+            _clock: PhantomData,
         }
     }
 
-    pub fn get_time_since_midnight(&self) -> i64 {
-        (self.time_s as i64 % 86400 * 1000) + (self.time_ns as i64 / 1000000)
-    }
-
     pub fn secs(&self) -> u64 {
         self.time_s
     }
@@ -43,3 +61,23 @@ impl Time {
         (self.time_s * 1000) + (self.time_ns / 1000000)
     }
 }
+
+impl ClockSample<Realtime> {
+    pub fn as_time_since_midnight(&self) -> i64 {
+        (self.time_s as i64 % 86_400 * 1000) + self.time_ns as i64 / 1_000_000
+    }
+}
+
+impl From<SystemTime> for ClockSample<Realtime> {
+    fn from(time: SystemTime) -> Self {
+        let duration = time
+            .duration_since(UNIX_EPOCH)
+            .expect("realtime timestamp predates Unix epoch");
+
+        Self {
+            time_s: duration.as_secs(),
+            time_ns: u64::from(duration.subsec_nanos()),
+            _clock: PhantomData,
+        }
+    }
+}

src/trace/characterization.rs

@@ -1,497 +0,0 @@
-// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use super::reader::{TraceCaptureV1, TraceHeaderV1};
-use super::v1::{Direction, Event, RateRequestReason, Record};
-use std::mem;
-use thiserror::Error;
-
-#[derive(Clone, Debug)]
-pub(super) struct CharacterizationV1 {
-    pub(super) header: TraceHeaderV1,
-    pub(super) initialization: Record,
-    pub(super) startup_rates: Record,
-    pub(super) evaluations: Vec<EvaluationCaseV1>,
-    pub(super) trailing_ping_replies: Vec<Record>,
-}
-
-#[derive(Clone, Debug)]
-pub(super) struct EvaluationCaseV1 {
-    pub(super) ping_replies_before: Vec<Record>,
-    pub(super) boundary: Record,
-    pub(super) download_calculation: Option<Record>,
-    pub(super) upload_calculation: Option<Record>,
-    pub(super) random_choices: Vec<Record>,
-    pub(super) expected_rates: Record,
-    pub(super) deferred_ping_replies: Vec<Record>,
-}
-
-#[derive(Debug, Error)]
-pub(super) enum CharacterizationErrorV1 {
-    #[error("trace has no controller initialization")]
-    MissingInitialization,
-    #[error("trace has no startup requested-rates record")]
-    MissingStartupRates,
-    #[error("record {seq} is structurally unexpected: {detail}")]
-    UnexpectedRecord { seq: u64, detail: &'static str },
-    #[error("evaluation {evaluation_id} is incomplete at EOF")]
-    IncompleteEvaluation { evaluation_id: u64 },
-    #[error("evaluation {evaluation_id} has invalid structure: {detail}")]
-    InvalidEvaluation {
-        evaluation_id: u64,
-        detail: &'static str,
-    },
-}
-
-struct OpenEvaluation {
-    evaluation_id: u64,
-    ping_replies_before: Vec<Record>,
-    boundary: Record,
-    download_calculation: Option<Record>,
-    upload_calculation: Option<Record>,
-    random_choices: Vec<Record>,
-    deferred_ping_replies: Vec<Record>,
-}
-
-pub(super) fn build_characterization_v1(
-    capture: &TraceCaptureV1,
-) -> Result<CharacterizationV1, CharacterizationErrorV1> {
-    let mut initialization = None;
-    let mut startup_rates = None;
-    let mut evaluations = Vec::new();
-    let mut pending_ping_replies = Vec::new();
-    let mut open = None::<OpenEvaluation>;
-
-    for record in &capture.records {
-        match &record.event {
-            Event::PingReply { .. } => {
-                if let Some(evaluation) = &mut open {
-                    evaluation.deferred_ping_replies.push(record.clone());
-                } else {
-                    pending_ping_replies.push(record.clone());
-                }
-            }
-            Event::ControllerInitialized { .. } => {
-                if initialization.is_some() || open.is_some() || !evaluations.is_empty() {
-                    return unexpected(record, "duplicate or late controller initialization");
-                }
-                initialization = Some(record.clone());
-            }
-            Event::ControlEvaluation { evaluation_id, .. } => {
-                if initialization.is_none() || startup_rates.is_none() {
-                    return unexpected(record, "evaluation precedes controller startup");
-                }
-                if open.is_some() {
-                    return unexpected(record, "evaluation begins before the previous one ends");
-                }
-                if evaluations
-                    .last()
-                    .is_none_or(|previous: &EvaluationCaseV1| {
-                        evaluation_id_of(&previous.boundary).wrapping_add(1) != *evaluation_id
-                    })
-                    && (!evaluations.is_empty() || *evaluation_id != 0)
-                {
-                    return invalid(*evaluation_id, "evaluation ids are not contiguous");
-                }
-                open = Some(OpenEvaluation {
-                    evaluation_id: *evaluation_id,
-                    ping_replies_before: mem::take(&mut pending_ping_replies),
-                    boundary: record.clone(),
-                    download_calculation: None,
-                    upload_calculation: None,
-                    random_choices: Vec::new(),
-                    deferred_ping_replies: Vec::new(),
-                });
-            }
-            Event::RateCalculation {
-                evaluation_id,
-                direction,
-            } => {
-                let evaluation = matching_open(&mut open, record, *evaluation_id)?;
-                let calculation = match direction {
-                    Direction::Download => &mut evaluation.download_calculation,
-                    Direction::Upload => &mut evaluation.upload_calculation,
-                };
-                if calculation.replace(record.clone()).is_some() {
-                    return invalid(*evaluation_id, "duplicate directional calculation");
-                }
-            }
-            Event::RandomSafeRateChoice {
-                evaluation_id,
-                direction,
-                ..
-            } => {
-                let evaluation = matching_open(&mut open, record, *evaluation_id)?;
-                let has_calculation = match direction {
-                    Direction::Download => evaluation.download_calculation.is_some(),
-                    Direction::Upload => evaluation.upload_calculation.is_some(),
-                };
-                if !has_calculation {
-                    return invalid(*evaluation_id, "random choice precedes its calculation");
-                }
-                evaluation.random_choices.push(record.clone());
-            }
-            Event::RequestedRates {
-                evaluation_id: None,
-                reason: RateRequestReason::Startup,
-                ..
-            } => {
-                if startup_rates.is_some() || open.is_some() || !evaluations.is_empty() {
-                    return unexpected(record, "duplicate or late startup rates");
-                }
-                startup_rates = Some(record.clone());
-            }
-            Event::RequestedRates {
-                evaluation_id: Some(evaluation_id),
-                reason,
-                ..
-            } => {
-                let evaluation = matching_open(&mut open, record, *evaluation_id)?;
-                match reason {
-                    RateRequestReason::Control
-                        if evaluation.download_calculation.is_none()
-                            || evaluation.upload_calculation.is_none() =>
-                    {
-                        return invalid(*evaluation_id, "control evaluation lacks a calculation");
-                    }
-                    RateRequestReason::NoReflectorData
-                        if evaluation.download_calculation.is_some()
-                            || evaluation.upload_calculation.is_some()
-                            || !evaluation.random_choices.is_empty() =>
-                    {
-                        return invalid(
-                            *evaluation_id,
-                            "no-reflector evaluation contains calculation metadata",
-                        );
-                    }
-                    RateRequestReason::Startup => {
-                        return invalid(*evaluation_id, "startup rates have an evaluation id");
-                    }
-                    _ => {}
-                }
-                let evaluation = open.take().expect("matching evaluation checked above");
-                evaluations.push(EvaluationCaseV1 {
-                    ping_replies_before: evaluation.ping_replies_before,
-                    boundary: evaluation.boundary,
-                    download_calculation: evaluation.download_calculation,
-                    upload_calculation: evaluation.upload_calculation,
-                    random_choices: evaluation.random_choices,
-                    expected_rates: record.clone(),
-                    deferred_ping_replies: evaluation.deferred_ping_replies,
-                });
-            }
-            Event::RequestedRates { .. } => {
-                return unexpected(record, "requested-rates reason and evaluation id disagree");
-            }
-            Event::Header { .. } | Event::End { .. } => {
-                return unexpected(record, "header or end leaked from the trace reader");
-            }
-        }
-    }
-
-    if let Some(evaluation) = open {
-        return Err(CharacterizationErrorV1::IncompleteEvaluation {
-            evaluation_id: evaluation.evaluation_id,
-        });
-    }
-
-    Ok(CharacterizationV1 {
-        header: capture.header.clone(),
-        initialization: initialization.ok_or(CharacterizationErrorV1::MissingInitialization)?,
-        startup_rates: startup_rates.ok_or(CharacterizationErrorV1::MissingStartupRates)?,
-        evaluations,
-        trailing_ping_replies: pending_ping_replies,
-    })
-}
-
-fn matching_open<'a>(
-    open: &'a mut Option<OpenEvaluation>,
-    record: &Record,
-    evaluation_id: u64,
-) -> Result<&'a mut OpenEvaluation, CharacterizationErrorV1> {
-    let Some(evaluation) = open else {
-        return unexpected(record, "evaluation-scoped record has no open evaluation");
-    };
-    if evaluation.evaluation_id != evaluation_id {
-        return invalid(
-            evaluation_id,
-            "record refers to a different open evaluation",
-        );
-    }
-    Ok(evaluation)
-}
-
-fn evaluation_id_of(record: &Record) -> u64 {
-    let Event::ControlEvaluation { evaluation_id, .. } = record.event else {
-        unreachable!("builder only stores control-evaluation boundaries")
-    };
-    evaluation_id
-}
-
-fn unexpected<T>(record: &Record, detail: &'static str) -> Result<T, CharacterizationErrorV1> {
-    Err(CharacterizationErrorV1::UnexpectedRecord {
-        seq: record.seq,
-        detail,
-    })
-}
-
-fn invalid<T>(evaluation_id: u64, detail: &'static str) -> Result<T, CharacterizationErrorV1> {
-    Err(CharacterizationErrorV1::InvalidEvaluation {
-        evaluation_id,
-        detail,
-    })
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::trace::reader::{CaptureCompletionV1, read_trace_v1};
-    use crate::trace::v1::{RateRequestReason, Timestamp};
-    use std::collections::HashSet;
-    use std::io::Cursor;
-
-    const LIVE_FIXTURE: &str = include_str!("../../tests/fixtures/trace-v1/speedtests.jsonl");
-
-    fn record(seq: u64, event: Event) -> Record {
-        Record {
-            seq,
-            timestamp: Timestamp {
-                mono_ns: seq,
-                unix_us: seq as i64,
-            },
-            event,
-        }
-    }
-
-    fn jsonl(records: &[Record]) -> String {
-        records
-            .iter()
-            .map(|record| format!("{}\n", serde_json::to_string(record).unwrap()))
-            .collect()
-    }
-
-    fn initialization(seq: u64) -> Record {
-        record(
-            seq,
-            Event::ControllerInitialized {
-                previous_rx_bytes: 10,
-                previous_tx_bytes: 20,
-                download_prev_mono_ns: 1,
-                upload_prev_mono_ns: 2,
-                download_history_index: 0,
-                upload_history_index: 0,
-                download_safe_rates_kbit: vec![100.0],
-                upload_safe_rates_kbit: vec![200.0],
-            },
-        )
-    }
-
-    fn requested(seq: u64, evaluation_id: Option<u64>, reason: RateRequestReason) -> Record {
-        record(
-            seq,
-            Event::RequestedRates {
-                evaluation_id,
-                reason,
-                download_kbit: 100,
-                upload_kbit: 200,
-                download_requested: true,
-                upload_requested: true,
-            },
-        )
-    }
-
-    fn ping(seq: u64, reflector: &str) -> Record {
-        record(
-            seq,
-            Event::PingReply {
-                reflector: reflector.parse().unwrap(),
-                down_time_ms: 1.0,
-                up_time_ms: 2.0,
-            },
-        )
-    }
-
-    fn boundary(seq: u64, evaluation_id: u64) -> Record {
-        record(
-            seq,
-            Event::ControlEvaluation {
-                evaluation_id,
-                loop_mono_ns: seq,
-                rx_bytes: 30,
-                tx_bytes: 40,
-                active_reflectors: vec!["192.0.2.1".parse().unwrap()],
-            },
-        )
-    }
-
-    fn calculation(seq: u64, evaluation_id: u64, direction: Direction) -> Record {
-        record(
-            seq,
-            Event::RateCalculation {
-                evaluation_id,
-                direction,
-            },
-        )
-    }
-
-    fn synthetic_capture(records: &[Record]) -> TraceCaptureV1 {
-        let mut all = vec![record(0, crate::trace::v1::tests::header_event())];
-        all.extend_from_slice(records);
-        let input = jsonl(&all);
-        read_trace_v1(Cursor::new(input)).unwrap()
-    }
-
-    #[test]
-    fn freezes_evaluation_before_interleaved_ping_replies() {
-        let capture = synthetic_capture(&[
-            initialization(1),
-            requested(2, None, RateRequestReason::Startup),
-            ping(3, "192.0.2.1"),
-            boundary(4, 0),
-            calculation(5, 0, Direction::Download),
-            ping(6, "192.0.2.2"),
-            calculation(7, 0, Direction::Upload),
-            requested(8, Some(0), RateRequestReason::Control),
-            ping(9, "192.0.2.3"),
-            boundary(10, 1),
-            calculation(11, 1, Direction::Download),
-            calculation(12, 1, Direction::Upload),
-            requested(13, Some(1), RateRequestReason::Control),
-        ]);
-
-        let characterization = build_characterization_v1(&capture).unwrap();
-
-        assert_eq!(characterization.header.seq, 0);
-        assert!(characterization.trailing_ping_replies.is_empty());
-        assert_eq!(
-            characterization.evaluations[0].ping_replies_before[0].seq,
-            3
-        );
-        assert_eq!(
-            characterization.evaluations[0].deferred_ping_replies[0].seq,
-            6
-        );
-        assert_eq!(
-            characterization.evaluations[1].ping_replies_before[0].seq,
-            9
-        );
-        assert!(
-            characterization.evaluations[1]
-                .deferred_ping_replies
-                .is_empty()
-        );
-    }
-
-    #[test]
-    fn live_fixture_parses_as_an_incomplete_observed_prefix() {
-        let capture = read_trace_v1(Cursor::new(LIVE_FIXTURE)).unwrap();
-
-        assert_eq!(capture.completion, CaptureCompletionV1::IncompleteEof);
-        assert_eq!(capture.header.program_version, "0.4.1");
-        assert_eq!(capture.records.last().unwrap().seq, 1515);
-    }
-
-    #[test]
-    fn live_fixture_covers_useful_legacy_decisions() {
-        let capture = read_trace_v1(Cursor::new(LIVE_FIXTURE)).unwrap();
-        let characterization = build_characterization_v1(&capture).unwrap();
-
-        assert!(matches!(
-            characterization.initialization.event,
-            Event::ControllerInitialized { .. }
-        ));
-        assert!(matches!(
-            characterization.startup_rates.event,
-            Event::RequestedRates {
-                reason: RateRequestReason::Startup,
-                ..
-            }
-        ));
-        assert!(characterization.evaluations.iter().any(|evaluation| {
-            let Event::ControlEvaluation {
-                active_reflectors, ..
-            } = &evaluation.boundary.event
-            else {
-                unreachable!();
-            };
-            active_reflectors.len() > 5
-        }));
-
-        let stochastic_index = characterization
-            .evaluations
-            .iter()
-            .position(|evaluation| !evaluation.random_choices.is_empty())
-            .expect("fixture must exercise the legacy stochastic branch");
-        let stochastic = &characterization.evaluations[stochastic_index];
-        assert_eq!(evaluation_id_of(&stochastic.boundary), 150);
-        assert!(matches!(
-            stochastic.random_choices.as_slice(),
-            [Record {
-                event: Event::RandomSafeRateChoice {
-                    direction: Direction::Download,
-                    index: 92,
-                    ..
-                },
-                ..
-            }]
-        ));
-        let Event::RequestedRates {
-            download_kbit,
-            download_requested,
-            ..
-        } = stochastic.expected_rates.event
-        else {
-            unreachable!();
-        };
-        assert_eq!(download_kbit, 256_374);
-        assert!(download_requested);
-        let Event::RequestedRates {
-            download_kbit: previous_download_kbit,
-            ..
-        } = characterization.evaluations[stochastic_index - 1]
-            .expected_rates
-            .event
-        else {
-            unreachable!();
-        };
-        assert!(download_kbit < previous_download_kbit);
-
-        let distinct_rates = characterization
-            .evaluations
-            .iter()
-            .map(|evaluation| match evaluation.expected_rates.event {
-                Event::RequestedRates {
-                    download_kbit,
-                    upload_kbit,
-                    ..
-                } => (download_kbit, upload_kbit),
-                _ => unreachable!(),
-            })
-            .collect::<HashSet<_>>();
-        assert!(distinct_rates.len() > 3);
-    }
-
-    #[test]
-    fn live_expected_rates_have_one_unambiguous_evaluation() {
-        let capture = read_trace_v1(Cursor::new(LIVE_FIXTURE)).unwrap();
-        let characterization = build_characterization_v1(&capture).unwrap();
-        let mut ids = HashSet::new();
-
-        for evaluation in &characterization.evaluations {
-            let evaluation_id = evaluation_id_of(&evaluation.boundary);
-            let Event::RequestedRates {
-                evaluation_id: Some(expected_id),
-                reason: RateRequestReason::Control,
-                ..
-            } = evaluation.expected_rates.event
-            else {
-                panic!("fixture evaluation must have normal control output");
-            };
-            assert_eq!(expected_id, evaluation_id);
-            assert!(ids.insert(evaluation_id));
-            assert!(evaluation.download_calculation.is_some());
-            assert!(evaluation.upload_calculation.is_some());
-        }
-    }
-}

src/trace/mod.rs

@@ -1,422 +0,0 @@
-// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-pub(crate) mod v1;
-
-#[cfg(all(feature = "trace", test))]
-mod characterization;
-#[cfg(all(feature = "trace", test))]
-mod reader;
-
-#[cfg(feature = "trace")]
-mod writer;
-
-use crate::settings::Settings;
-use std::time::Instant;
-
-#[cfg(feature = "trace")]
-use std::time::SystemTime;
-
-#[cfg(feature = "trace")]
-use flume::Sender;
-#[cfg(feature = "trace")]
-use log::warn;
-#[cfg(feature = "trace")]
-use std::fs::File;
-#[cfg(feature = "trace")]
-use std::io::{BufWriter, Write};
-#[cfg(feature = "trace")]
-use std::path::Path;
-#[cfg(feature = "trace")]
-use std::sync::atomic::{AtomicBool, Ordering};
-#[cfg(feature = "trace")]
-use std::sync::{Arc, Mutex};
-#[cfg(feature = "trace")]
-use std::thread::JoinHandle;
-
-const TRACE_PATH_ENV: &str = "SQMA_TRACE_V1_PATH";
-#[cfg(feature = "trace")]
-const QUEUE_CAPACITY: usize = 4_096;
-
-#[derive(Clone, Default)]
-pub(crate) struct Recorder {
-    #[cfg(feature = "trace")]
-    inner: Option<Arc<Inner>>,
-}
-
-#[cfg(feature = "trace")]
-struct Inner {
-    origin: Instant,
-    records: Sender<v1::Record>,
-    next_seq: Mutex<u64>,
-    healthy: Arc<AtomicBool>,
-    writer: Mutex<Option<JoinHandle<()>>>,
-}
-
-pub(crate) struct OrderedRecorder<'a> {
-    #[cfg(feature = "trace")]
-    inner: Option<&'a Inner>,
-    #[cfg(feature = "trace")]
-    next_seq: Option<&'a mut u64>,
-    #[cfg(not(feature = "trace"))]
-    marker: std::marker::PhantomData<&'a ()>,
-}
-
-impl Recorder {
-    pub(crate) fn from_env(origin: Instant, settings: &Settings) -> Self {
-        let Some(path) = std::env::var_os(TRACE_PATH_ENV) else {
-            return Self::disabled();
-        };
-
-        #[cfg(feature = "trace")]
-        {
-            let header = v1::Event::Header {
-                format: v1::FORMAT_NAME.to_owned(),
-                version: v1::FORMAT_VERSION,
-                program_version: env!("CARGO_PKG_VERSION").to_owned(),
-                control: v1::ControlConfig::from_settings(settings),
-            };
-            Self::start_path(Path::new(&path), origin, header)
-        }
-
-        #[cfg(not(feature = "trace"))]
-        {
-            let _ = (origin, settings, path);
-            log::warn!(
-                "{TRACE_PATH_ENV} is set, but this binary was built without the `trace` feature"
-            );
-            Self::disabled()
-        }
-    }
-
-    pub(crate) fn disabled() -> Self {
-        Self::default()
-    }
-
-    pub(crate) fn is_enabled(&self) -> bool {
-        #[cfg(feature = "trace")]
-        {
-            self.inner
-                .as_ref()
-                .is_some_and(|inner| inner.healthy.load(Ordering::Acquire))
-        }
-
-        #[cfg(not(feature = "trace"))]
-        {
-            false
-        }
-    }
-
-    pub(crate) fn monotonic_ns(&self, instant: Instant) -> u64 {
-        #[cfg(feature = "trace")]
-        {
-            self.inner
-                .as_ref()
-                .map_or(0, |inner| v1::monotonic_ns(inner.origin, instant))
-        }
-
-        #[cfg(not(feature = "trace"))]
-        {
-            let _ = instant;
-            0
-        }
-    }
-
-    pub(crate) fn record_at(&self, instant: Instant, event: v1::Event) -> Option<u64> {
-        self.linearize(|recorder| recorder.record_at(instant, event))
-    }
-
-    pub(crate) fn linearize<R>(&self, callback: impl FnOnce(&mut OrderedRecorder<'_>) -> R) -> R {
-        #[cfg(feature = "trace")]
-        {
-            if self
-                .inner
-                .as_ref()
-                .is_some_and(|inner| inner.healthy.load(Ordering::Acquire))
-            {
-                let inner = self
-                    .inner
-                    .as_ref()
-                    .expect("healthy trace recorder must have inner state");
-                match inner.next_seq.lock() {
-                    Ok(mut next_seq) => {
-                        if inner.healthy.load(Ordering::Acquire) {
-                            let mut recorder = OrderedRecorder {
-                                inner: Some(inner),
-                                next_seq: Some(&mut next_seq),
-                            };
-                            return callback(&mut recorder);
-                        }
-                    }
-                    Err(_) => {
-                        if inner.healthy.swap(false, Ordering::AcqRel) {
-                            warn!("Legacy control trace sequencing lock failed; tracing disabled");
-                        }
-                    }
-                }
-            }
-
-            let mut recorder = OrderedRecorder {
-                inner: None,
-                next_seq: None,
-            };
-            callback(&mut recorder)
-        }
-
-        #[cfg(not(feature = "trace"))]
-        {
-            let mut recorder = OrderedRecorder {
-                marker: std::marker::PhantomData,
-            };
-            callback(&mut recorder)
-        }
-    }
-
-    pub(crate) fn finish(&self) {
-        #[cfg(feature = "trace")]
-        if let Some(inner) = self.inner.as_ref() {
-            self.linearize(|recorder| {
-                recorder.record_at(Instant::now(), v1::Event::End { clean: true });
-                inner.healthy.store(false, Ordering::Release);
-            });
-
-            let writer = match inner.writer.lock() {
-                Ok(mut writer) => writer.take(),
-                Err(_) => {
-                    warn!("Legacy control trace writer lock failed during shutdown");
-                    None
-                }
-            };
-            let writer_panicked = writer.map(|writer| writer.join().is_err()).unwrap_or(false);
-            if writer_panicked {
-                warn!("Legacy control trace writer panicked during shutdown");
-            }
-        }
-    }
-
-    #[cfg(feature = "trace")]
-    fn start_path(path: &Path, origin: Instant, header: v1::Event) -> Self {
-        let file = match File::options()
-            .create(true)
-            .write(true)
-            .truncate(true)
-            .open(path)
-        {
-            Ok(file) => file,
-            Err(error) => {
-                warn!(
-                    "Failed to open legacy control trace at {}: {error}; tracing disabled",
-                    path.display()
-                );
-                return Self::disabled();
-            }
-        };
-
-        Self::start_writer(BufWriter::new(file), origin, header)
-    }
-
-    #[cfg(feature = "trace")]
-    fn start_writer<W>(writer: W, origin: Instant, header: v1::Event) -> Self
-    where
-        W: Write + Send + 'static,
-    {
-        let (records, record_rx) = flume::bounded(QUEUE_CAPACITY);
-        let healthy = Arc::new(AtomicBool::new(true));
-        let header = v1::Record {
-            seq: 0,
-            timestamp: v1::timestamp(origin, origin, SystemTime::now()),
-            event: header,
-        };
-        let writer = match writer::spawn(writer, record_rx, header, Arc::clone(&healthy)) {
-            Ok(writer) => writer,
-            Err(error) => {
-                warn!("Failed to start legacy control trace writer: {error}; tracing disabled");
-                return Self::disabled();
-            }
-        };
-
-        Self {
-            inner: Some(Arc::new(Inner {
-                origin,
-                records,
-                next_seq: Mutex::new(1),
-                healthy,
-                writer: Mutex::new(Some(writer)),
-            })),
-        }
-    }
-}
-
-impl OrderedRecorder<'_> {
-    pub(crate) fn is_enabled(&self) -> bool {
-        #[cfg(feature = "trace")]
-        {
-            self.inner.is_some()
-        }
-
-        #[cfg(not(feature = "trace"))]
-        {
-            false
-        }
-    }
-
-    pub(crate) fn record_at(&mut self, instant: Instant, event: v1::Event) -> Option<u64> {
-        #[cfg(feature = "trace")]
-        {
-            let inner = self.inner?;
-            let next_seq = self.next_seq.as_deref_mut()?;
-            let seq = *next_seq;
-            let record = v1::Record {
-                seq,
-                timestamp: v1::timestamp(inner.origin, instant, SystemTime::now()),
-                event,
-            };
-
-            if inner.records.send(record).is_err() {
-                if inner.healthy.swap(false, Ordering::AcqRel) {
-                    warn!("Legacy control trace channel closed; tracing disabled");
-                }
-                return None;
-            }
-
-            *next_seq = (*next_seq).saturating_add(1);
-            Some(seq)
-        }
-
-        #[cfg(not(feature = "trace"))]
-        {
-            let _ = (instant, event);
-            None
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn disabled_recorder_is_a_no_op() {
-        let recorder = Recorder::disabled();
-
-        assert!(!recorder.is_enabled());
-        assert_eq!(
-            recorder.record_at(Instant::now(), v1::Event::End { clean: true }),
-            None
-        );
-        recorder.finish();
-    }
-
-    #[cfg(feature = "trace")]
-    mod enabled {
-        use super::*;
-        use std::io;
-        use std::net::{IpAddr, Ipv4Addr};
-        use std::path::PathBuf;
-        use std::sync::Barrier;
-        use std::sync::atomic::{AtomicU64, Ordering};
-        use std::thread;
-
-        static NEXT_PATH_ID: AtomicU64 = AtomicU64::new(0);
-
-        struct FailingWriter;
-
-        impl Write for FailingWriter {
-            fn write(&mut self, _buffer: &[u8]) -> io::Result<usize> {
-                Err(io::Error::other("injected writer failure"))
-            }
-
-            fn flush(&mut self) -> io::Result<()> {
-                Err(io::Error::other("injected writer failure"))
-            }
-        }
-
-        fn temp_path() -> PathBuf {
-            let id = NEXT_PATH_ID.fetch_add(1, Ordering::Relaxed);
-            std::env::temp_dir().join(format!(
-                "sqm-autorate-trace-v1-{}-{id}.jsonl",
-                std::process::id()
-            ))
-        }
-
-        fn ping_event(reflector: Ipv4Addr, value: f64) -> v1::Event {
-            v1::Event::PingReply {
-                reflector: IpAddr::V4(reflector),
-                down_time_ms: value,
-                up_time_ms: value,
-            }
-        }
-
-        #[test]
-        fn writer_preserves_global_sequence_order_from_multiple_producers() {
-            let path = temp_path();
-            let origin = Instant::now();
-            let recorder = Recorder::start_path(&path, origin, v1::tests::header_event());
-            let barrier = Arc::new(Barrier::new(3));
-            let mut producers = Vec::new();
-
-            for producer in 1..=2 {
-                let recorder = recorder.clone();
-                let barrier = Arc::clone(&barrier);
-                producers.push(thread::spawn(move || {
-                    barrier.wait();
-                    for value in 0..50 {
-                        recorder.record_at(
-                            origin,
-                            ping_event(Ipv4Addr::new(192, 0, 2, producer), value as f64),
-                        );
-                    }
-                }));
-            }
-
-            barrier.wait();
-            for producer in producers {
-                producer.join().unwrap();
-            }
-            recorder.finish();
-
-            let contents = std::fs::read_to_string(&path).unwrap();
-            let records = contents
-                .lines()
-                .map(|line| super::reader::parse_record_v1(line).unwrap())
-                .collect::<Vec<_>>();
-
-            assert_eq!(records.len(), 102);
-            assert!(matches!(
-                &records.first().unwrap().event,
-                v1::Event::Header { .. }
-            ));
-            assert!(matches!(
-                &records.last().unwrap().event,
-                v1::Event::End { clean: true }
-            ));
-            for (expected, record) in records.iter().enumerate() {
-                assert_eq!(record.seq, expected as u64);
-            }
-
-            std::fs::remove_file(path).unwrap();
-        }
-
-        #[test]
-        fn writer_failure_disables_recorder_without_propagating() {
-            let recorder =
-                Recorder::start_writer(FailingWriter, Instant::now(), v1::tests::header_event());
-
-            recorder.record_at(Instant::now(), ping_event(Ipv4Addr::new(192, 0, 2, 1), 1.0));
-            recorder.finish();
-
-            assert!(!recorder.is_enabled());
-        }
-
-        #[test]
-        fn open_failure_returns_disabled_recorder() {
-            let path = temp_path().join("missing-parent").join("trace.jsonl");
-            let recorder = Recorder::start_path(&path, Instant::now(), v1::tests::header_event());
-
-            assert!(!recorder.is_enabled());
-            recorder.record_at(Instant::now(), v1::Event::End { clean: true });
-            recorder.finish();
-        }
-    }
-}

src/trace/reader.rs

@@ -1,343 +0,0 @@
-// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use super::v1::{self, ControlConfig, Event, Record, Timestamp};
-use std::io::{self, BufRead};
-use thiserror::Error;
-
-#[derive(Clone, Debug, PartialEq)]
-pub(super) struct TraceHeaderV1 {
-    pub(super) seq: u64,
-    pub(super) timestamp: Timestamp,
-    pub(super) program_version: String,
-    pub(super) control: ControlConfig,
-}
-
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub(super) enum CaptureCompletionV1 {
-    Complete,
-    IncompleteEof,
-    IncompleteEnd,
-    TruncatedFinalLine { line: usize },
-}
-
-#[derive(Clone, Debug, PartialEq)]
-pub(super) struct TraceCaptureV1 {
-    pub(super) header: TraceHeaderV1,
-    pub(super) records: Vec<Record>,
-    pub(super) completion: CaptureCompletionV1,
-}
-
-#[derive(Debug, Error)]
-pub(super) enum TraceReadErrorV1 {
-    #[error("failed to read trace line {line}: {source}")]
-    Io {
-        line: usize,
-        #[source]
-        source: io::Error,
-    },
-    #[error("trace line {line} is not a valid v1 record: {source}")]
-    MalformedRecord {
-        line: usize,
-        #[source]
-        source: serde_json::Error,
-    },
-    #[error("trace is missing its v1 header")]
-    MissingHeader,
-    #[error("first trace record must be the v1 header")]
-    FirstRecordNotHeader,
-    #[error("trace header sequence must be 0, got {actual}")]
-    InvalidHeaderSequence { actual: u64 },
-    #[error("unsupported trace format {actual:?}")]
-    InvalidFormat { actual: String },
-    #[error("unsupported trace version {actual}")]
-    InvalidVersion { actual: u16 },
-    #[error("trace line {line} has sequence {actual}, expected {expected}")]
-    InvalidSequence {
-        line: usize,
-        expected: u64,
-        actual: u64,
-    },
-    #[error("trace line {line} contains another header")]
-    UnexpectedHeader { line: usize },
-    #[error("trace line {line} follows an end record")]
-    RecordAfterEnd { line: usize },
-}
-
-pub(super) fn read_trace_v1(mut reader: impl BufRead) -> Result<TraceCaptureV1, TraceReadErrorV1> {
-    let mut header = None;
-    let mut records = Vec::new();
-    let mut next_seq = 0;
-    let mut completion = None;
-    let mut line_number = 0;
-
-    loop {
-        let mut line = String::new();
-        line_number += 1;
-        let bytes_read = reader
-            .read_line(&mut line)
-            .map_err(|source| TraceReadErrorV1::Io {
-                line: line_number,
-                source,
-            })?;
-        if bytes_read == 0 {
-            break;
-        }
-
-        if completion.is_some() {
-            return Err(TraceReadErrorV1::RecordAfterEnd { line: line_number });
-        }
-
-        let terminated = line.ends_with('\n');
-        let json = line.strip_suffix('\n').unwrap_or(&line);
-        let json = json.strip_suffix('\r').unwrap_or(json);
-        let record = match parse_record_v1(json) {
-            Ok(record) => record,
-            Err(source)
-                if !terminated
-                    && !json.trim().is_empty()
-                    && source.is_eof()
-                    && header.is_some() =>
-            {
-                return Ok(TraceCaptureV1 {
-                    header: header.expect("header checked above"),
-                    records,
-                    completion: CaptureCompletionV1::TruncatedFinalLine { line: line_number },
-                });
-            }
-            Err(source) => {
-                return Err(TraceReadErrorV1::MalformedRecord {
-                    line: line_number,
-                    source,
-                });
-            }
-        };
-
-        if record.seq != next_seq {
-            if header.is_none() && next_seq == 0 {
-                return Err(TraceReadErrorV1::InvalidHeaderSequence { actual: record.seq });
-            }
-            return Err(TraceReadErrorV1::InvalidSequence {
-                line: line_number,
-                expected: next_seq,
-                actual: record.seq,
-            });
-        }
-        next_seq = next_seq.saturating_add(1);
-
-        if header.is_none() {
-            let Event::Header {
-                format,
-                version,
-                program_version,
-                control,
-            } = record.event
-            else {
-                return Err(TraceReadErrorV1::FirstRecordNotHeader);
-            };
-            if format != v1::FORMAT_NAME {
-                return Err(TraceReadErrorV1::InvalidFormat { actual: format });
-            }
-            if version != v1::FORMAT_VERSION {
-                return Err(TraceReadErrorV1::InvalidVersion { actual: version });
-            }
-            header = Some(TraceHeaderV1 {
-                seq: record.seq,
-                timestamp: record.timestamp,
-                program_version,
-                control,
-            });
-            continue;
-        }
-
-        match record.event {
-            Event::Header { .. } => {
-                return Err(TraceReadErrorV1::UnexpectedHeader { line: line_number });
-            }
-            Event::End { clean } => {
-                completion = Some(if clean {
-                    CaptureCompletionV1::Complete
-                } else {
-                    CaptureCompletionV1::IncompleteEnd
-                });
-            }
-            _ => records.push(record),
-        }
-    }
-
-    let header = header.ok_or(TraceReadErrorV1::MissingHeader)?;
-    Ok(TraceCaptureV1 {
-        header,
-        records,
-        completion: completion.unwrap_or(CaptureCompletionV1::IncompleteEof),
-    })
-}
-
-pub(super) fn parse_record_v1(json: &str) -> Result<Record, serde_json::Error> {
-    serde_json::from_str(json)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::io::Cursor;
-
-    fn record(seq: u64, event: Event) -> Record {
-        Record {
-            seq,
-            timestamp: Timestamp {
-                mono_ns: seq,
-                unix_us: seq as i64,
-            },
-            event,
-        }
-    }
-
-    fn jsonl(records: &[Record]) -> String {
-        records
-            .iter()
-            .map(|record| format!("{}\n", serde_json::to_string(record).unwrap()))
-            .collect()
-    }
-
-    fn header(seq: u64) -> Record {
-        record(seq, v1::tests::header_event())
-    }
-
-    fn ping(seq: u64) -> Record {
-        record(
-            seq,
-            Event::PingReply {
-                reflector: "192.0.2.1".parse().unwrap(),
-                down_time_ms: 1.0,
-                up_time_ms: 2.0,
-            },
-        )
-    }
-
-    #[test]
-    fn parses_clean_v1_trace() {
-        let input = jsonl(&[header(0), ping(1), record(2, Event::End { clean: true })]);
-
-        let capture = read_trace_v1(Cursor::new(input)).unwrap();
-
-        assert_eq!(capture.header.seq, 0);
-        assert_eq!(capture.records, vec![ping(1)]);
-        assert_eq!(capture.completion, CaptureCompletionV1::Complete);
-    }
-
-    #[test]
-    fn retains_records_before_truncated_final_line() {
-        let input = format!(
-            "{}{{\"seq\":2,\"timestamp\":{{",
-            jsonl(&[header(0), ping(1)])
-        );
-
-        let capture = read_trace_v1(Cursor::new(input)).unwrap();
-
-        assert_eq!(capture.records, vec![ping(1)]);
-        assert_eq!(
-            capture.completion,
-            CaptureCompletionV1::TruncatedFinalLine { line: 3 }
-        );
-    }
-
-    #[test]
-    fn marks_valid_eof_without_end_incomplete() {
-        let capture = read_trace_v1(Cursor::new(jsonl(&[header(0), ping(1)]))).unwrap();
-
-        assert_eq!(capture.records, vec![ping(1)]);
-        assert_eq!(capture.completion, CaptureCompletionV1::IncompleteEof);
-    }
-
-    #[test]
-    fn rejects_malformed_middle_line() {
-        let input = format!("{}not-json\n{}", jsonl(&[header(0)]), jsonl(&[ping(1)]));
-
-        assert!(matches!(
-            read_trace_v1(Cursor::new(input)),
-            Err(TraceReadErrorV1::MalformedRecord { line: 2, .. })
-        ));
-    }
-
-    #[test]
-    fn rejects_sequence_gap_duplicate_and_out_of_order() {
-        let cases = [
-            (vec![header(0), ping(2)], 2, 1, 2),
-            (vec![header(0), ping(1), ping(1)], 3, 2, 1),
-            (vec![header(0), ping(1), ping(2), ping(1)], 4, 3, 1),
-        ];
-
-        for (records, line, expected, actual) in cases {
-            let input = jsonl(&records);
-            assert!(matches!(
-                read_trace_v1(Cursor::new(input)),
-                Err(TraceReadErrorV1::InvalidSequence {
-                    line: seen_line,
-                    expected: seen_expected,
-                    actual: seen,
-                }) if seen_line == line && seen_expected == expected && seen == actual
-            ));
-        }
-    }
-
-    #[test]
-    fn rejects_non_truncation_error_on_final_line() {
-        let input = format!("{}{{\"seq\":1,]", jsonl(&[header(0)]));
-
-        assert!(matches!(
-            read_trace_v1(Cursor::new(input)),
-            Err(TraceReadErrorV1::MalformedRecord { line: 2, .. })
-        ));
-    }
-
-    #[test]
-    fn rejects_records_after_clean_end() {
-        let input = jsonl(&[header(0), record(1, Event::End { clean: true }), ping(2)]);
-
-        assert!(matches!(
-            read_trace_v1(Cursor::new(input)),
-            Err(TraceReadErrorV1::RecordAfterEnd { line: 3 })
-        ));
-    }
-
-    #[test]
-    fn validates_format_name_and_version() {
-        let Event::Header {
-            program_version,
-            control,
-            ..
-        } = v1::tests::header_event()
-        else {
-            unreachable!();
-        };
-        let invalid_format = record(
-            0,
-            Event::Header {
-                format: "other".to_owned(),
-                version: v1::FORMAT_VERSION,
-                program_version: program_version.clone(),
-                control: control.clone(),
-            },
-        );
-        let invalid_version = record(
-            0,
-            Event::Header {
-                format: v1::FORMAT_NAME.to_owned(),
-                version: v1::FORMAT_VERSION + 1,
-                program_version,
-                control,
-            },
-        );
-
-        assert!(matches!(
-            read_trace_v1(Cursor::new(jsonl(&[invalid_format]))),
-            Err(TraceReadErrorV1::InvalidFormat { .. })
-        ));
-        assert!(matches!(
-            read_trace_v1(Cursor::new(jsonl(&[invalid_version]))),
-            Err(TraceReadErrorV1::InvalidVersion { .. })
-        ));
-    }
-}

src/trace/v1.rs

@@ -1,269 +0,0 @@
-// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use serde::de::{Deserializer, Visitor};
-use serde::{Deserialize, Serialize};
-use std::fmt;
-use std::net::IpAddr;
-
-#[cfg(feature = "trace")]
-use crate::settings::Settings;
-#[cfg(any(feature = "trace", test))]
-use std::time::Duration;
-#[cfg(any(feature = "trace", test))]
-use std::time::Instant;
-#[cfg(feature = "trace")]
-use std::time::{SystemTime, UNIX_EPOCH};
-
-#[cfg(feature = "trace")]
-pub(crate) const FORMAT_VERSION: u16 = 1;
-#[cfg(feature = "trace")]
-pub(crate) const FORMAT_NAME: &str = "legacy-control-trace-v1";
-
-#[cfg(feature = "trace")]
-#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
-pub(crate) struct Timestamp {
-    pub(crate) mono_ns: u64,
-    pub(crate) unix_us: i64,
-}
-
-#[cfg(feature = "trace")]
-#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
-pub(crate) struct Record {
-    pub(crate) seq: u64,
-    pub(crate) timestamp: Timestamp,
-    #[serde(flatten)]
-    pub(crate) event: Event,
-}
-
-#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
-#[serde(tag = "type", rename_all = "snake_case")]
-pub(crate) enum Event {
-    #[cfg(feature = "trace")]
-    Header {
-        format: String,
-        version: u16,
-        program_version: String,
-        control: ControlConfig,
-    },
-    PingReply {
-        reflector: IpAddr,
-        down_time_ms: f64,
-        up_time_ms: f64,
-    },
-    ControllerInitialized {
-        #[serde(deserialize_with = "deserialize_i128")]
-        previous_rx_bytes: i128,
-        #[serde(deserialize_with = "deserialize_i128")]
-        previous_tx_bytes: i128,
-        download_prev_mono_ns: u64,
-        upload_prev_mono_ns: u64,
-        download_history_index: u64,
-        upload_history_index: u64,
-        download_safe_rates_kbit: Vec<f64>,
-        upload_safe_rates_kbit: Vec<f64>,
-    },
-    ControlEvaluation {
-        evaluation_id: u64,
-        loop_mono_ns: u64,
-        #[serde(deserialize_with = "deserialize_i128")]
-        rx_bytes: i128,
-        #[serde(deserialize_with = "deserialize_i128")]
-        tx_bytes: i128,
-        active_reflectors: Vec<IpAddr>,
-    },
-    RateCalculation {
-        evaluation_id: u64,
-        direction: Direction,
-    },
-    RandomSafeRateChoice {
-        evaluation_id: u64,
-        direction: Direction,
-        index: u64,
-        rate_kbit: f64,
-    },
-    RequestedRates {
-        evaluation_id: Option<u64>,
-        reason: RateRequestReason,
-        download_kbit: u64,
-        upload_kbit: u64,
-        download_requested: bool,
-        upload_requested: bool,
-    },
-    End {
-        clean: bool,
-    },
-}
-
-fn deserialize_i128<'de, D>(deserializer: D) -> Result<i128, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    struct I128Visitor;
-
-    impl Visitor<'_> for I128Visitor {
-        type Value = i128;
-
-        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-            formatter.write_str("a signed or unsigned integer")
-        }
-
-        fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E> {
-            Ok(i128::from(value))
-        }
-
-        fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
-            Ok(i128::from(value))
-        }
-    }
-
-    deserializer.deserialize_any(I128Visitor)
-}
-
-#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub(crate) enum Direction {
-    Download,
-    Upload,
-}
-
-#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub(crate) enum RateRequestReason {
-    Startup,
-    Control,
-    NoReflectorData,
-}
-
-#[cfg(feature = "trace")]
-#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
-pub(crate) struct ControlConfig {
-    pub(crate) tick_interval_secs: f64,
-    pub(crate) min_change_interval_secs: f64,
-    pub(crate) download_base_kbit: f64,
-    pub(crate) upload_base_kbit: f64,
-    pub(crate) download_min_kbit: f64,
-    pub(crate) upload_min_kbit: f64,
-    pub(crate) download_delay_ms: f64,
-    pub(crate) upload_delay_ms: f64,
-    pub(crate) high_load_level: f64,
-    pub(crate) speed_history_size: u32,
-}
-
-#[cfg(feature = "trace")]
-impl ControlConfig {
-    pub(crate) fn from_settings(settings: &Settings) -> Self {
-        Self {
-            tick_interval_secs: settings.advanced_settings.tick_interval,
-            min_change_interval_secs: settings.advanced_settings.min_change_interval,
-            download_base_kbit: settings.network.download_base_kbits,
-            upload_base_kbit: settings.network.upload_base_kbits,
-            download_min_kbit: settings.network.download_min_kbits(),
-            upload_min_kbit: settings.network.upload_min_kbits(),
-            download_delay_ms: settings.advanced_settings.download_delay_ms,
-            upload_delay_ms: settings.advanced_settings.upload_delay_ms,
-            high_load_level: settings.advanced_settings.high_load_level,
-            speed_history_size: settings.advanced_settings.speed_hist_size,
-        }
-    }
-}
-
-#[cfg(feature = "trace")]
-pub(crate) fn timestamp(origin: Instant, monotonic: Instant, realtime: SystemTime) -> Timestamp {
-    Timestamp {
-        mono_ns: monotonic_ns(origin, monotonic),
-        unix_us: unix_us(realtime),
-    }
-}
-
-#[cfg(any(feature = "trace", test))]
-pub(crate) fn monotonic_ns(origin: Instant, monotonic: Instant) -> u64 {
-    duration_ns(monotonic.saturating_duration_since(origin))
-}
-
-#[cfg(any(feature = "trace", test))]
-fn duration_ns(duration: Duration) -> u64 {
-    duration.as_nanos().min(u64::MAX as u128) as u64
-}
-
-#[cfg(feature = "trace")]
-fn unix_us(realtime: SystemTime) -> i64 {
-    match realtime.duration_since(UNIX_EPOCH) {
-        Ok(duration) => duration
-            .as_micros()
-            .min(i64::MAX as u128)
-            .try_into()
-            .unwrap_or(i64::MAX),
-        Err(error) => {
-            let micros = error.duration().as_micros().min(i64::MAX as u128) as i64;
-            -micros
-        }
-    }
-}
-
-#[cfg(test)]
-pub(super) mod tests {
-    use super::*;
-
-    #[cfg(feature = "trace")]
-    fn control_config() -> ControlConfig {
-        ControlConfig {
-            tick_interval_secs: 0.5,
-            min_change_interval_secs: 0.5,
-            download_base_kbit: 100_000.0,
-            upload_base_kbit: 20_000.0,
-            download_min_kbit: 20_000.0,
-            upload_min_kbit: 4_000.0,
-            download_delay_ms: 15.0,
-            upload_delay_ms: 15.0,
-            high_load_level: 0.8,
-            speed_history_size: 100,
-        }
-    }
-
-    #[cfg(feature = "trace")]
-    pub(crate) fn header_event() -> Event {
-        Event::Header {
-            format: FORMAT_NAME.to_owned(),
-            version: FORMAT_VERSION,
-            program_version: "test".to_owned(),
-            control: control_config(),
-        }
-    }
-
-    #[cfg(feature = "trace")]
-    #[test]
-    fn serializes_versioned_header() {
-        let origin = Instant::now();
-        let record = Record {
-            seq: 0,
-            timestamp: timestamp(origin, origin, UNIX_EPOCH + Duration::from_secs(1)),
-            event: header_event(),
-        };
-
-        let json = serde_json::to_value(&record).unwrap();
-
-        assert_eq!(json["seq"], 0);
-        assert_eq!(json["type"], "header");
-        assert_eq!(json["format"], FORMAT_NAME);
-        assert_eq!(json["version"], FORMAT_VERSION);
-        assert_eq!(json["timestamp"]["mono_ns"], 0);
-        assert_eq!(
-            crate::trace::reader::parse_record_v1(&serde_json::to_string(&record).unwrap())
-                .unwrap(),
-            record
-        );
-    }
-
-    #[test]
-    fn converts_existing_instants_relative_to_origin() {
-        let origin = Instant::now();
-
-        assert_eq!(
-            monotonic_ns(origin, origin + Duration::from_nanos(123_456)),
-            123_456
-        );
-        assert_eq!(monotonic_ns(origin, origin - Duration::from_nanos(1)), 0);
-    }
-}

src/trace/writer.rs

@@ -1,54 +0,0 @@
-// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use super::v1::{Event, Record};
-use flume::Receiver;
-use log::warn;
-use std::io::{self, Write};
-use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::thread::{self, JoinHandle};
-
-pub(super) fn spawn<W>(
-    writer: W,
-    records: Receiver<Record>,
-    header: Record,
-    healthy: Arc<AtomicBool>,
-) -> io::Result<JoinHandle<()>>
-where
-    W: Write + Send + 'static,
-{
-    thread::Builder::new()
-        .name("trace-v1-writer".to_owned())
-        .spawn(move || {
-            if let Err(error) = write_records(writer, records, header) {
-                warn!("Legacy control trace writer failed; tracing disabled: {error}");
-                healthy.store(false, Ordering::Release);
-            }
-        })
-}
-
-fn write_records<W>(mut writer: W, records: Receiver<Record>, header: Record) -> io::Result<()>
-where
-    W: Write,
-{
-    write_record(&mut writer, &header)?;
-
-    while let Ok(record) = records.recv() {
-        let clean_end = matches!(record.event, Event::End { clean: true });
-        write_record(&mut writer, &record)?;
-
-        if clean_end {
-            writer.flush()?;
-            return Ok(());
-        }
-    }
-
-    writer.flush()
-}
-
-fn write_record(writer: &mut impl Write, record: &Record) -> io::Result<()> {
-    serde_json::to_writer(&mut *writer, record).map_err(io::Error::other)?;
-    writer.write_all(b"\n")
-}

tests/fixtures/trace-v1/README.md

@@ -1,7 +0,0 @@
-# Trace v1 characterization fixture
-
-`speedtests.jsonl` is an observed prefix of a live legacy-controller capture
-recorded during speed and LibreQoS testing. It preserves the original records,
-sequence values, and order through evaluation 150. The prefix intentionally has
-no synthetic `end` record and may include undesirable legacy behavior; it is an
-extraction-characterization artifact, not a correctness oracle.