refactor(measurement): migrate ICMP engine to Tokio async runtime

Remove direct `rustix::time::ClockId` imports by centralizing its dependency within the `time` module.

.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,63 @@
+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
+        run: |
+          "$(go env GOPATH)/bin/actionlint" \
+            .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.
+# This workflow uploads SARIF results to GitHub Code Scanning.
-# They are provided by a third-party and are governed by
-# separate terms of service, privacy policy, and support
-# documentation.
-
 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.
-# 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.
 
-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)
+      contents: read
-      security-events: write # for uploading SARIF files
+      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.6.0
-    uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@v2.5.0"
     with:
-      # Just scan the root directory and docs, since everything else is fixtures
+      scan-args: ./
-      scan-args: |-
-        ./
-        ./docs/
   scan-pr:
     permissions:
-      contents: read # to fetch code (actions/checkout)
+      contents: read
-      security-events: write # for uploading SARIF files
+      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.6.0
-    uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@v2.5.0"
     with:
-      # Just scan the root directory and docs, since everything else is fixtures
+      scan-args: ./
-      scan-args: |-
-        ./
-        ./docs/

.github/workflows/rust-clippy.yml

@@ -1,55 +1,61 @@
-# This workflow uses actions that are not certified by GitHub.
+name: Clippy SARIF
-# 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
 
 on:
   push:
-    branches: [ "master" ]
+    branches: [master]
-  pull_request:
-    # The branches below must be a subset of the branches above
-    branches: [ "master" ]
   schedule:
-    - cron: '28 20 * * 3'
+    - cron: "17 12 * * 4"
+  workflow_dispatch:
 
-jobs:
-  rust-clippy-analyze:
-    name: Run rust-clippy analyzing
-    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
+  actions: read
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  clippy-sarif:
+    name: Clippy SARIF
+    runs-on: ubuntu-latest
+    timeout-minutes: 25
     steps:
-      - name: Checkout code
+      - name: Checkout
-        uses: actions/checkout@v7
+        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
+          cache: false
-
+      - name: Install SARIF tools
-      - name: Install required cargo
+        run: cargo install --locked clippy-sarif sarif-fmt
-        run: cargo install clippy-sarif sarif-fmt
+      - name: Run Clippy and generate SARIF
-
+        id: clippy
-      - name: Run rust-clippy
+        shell: bash
-        run:
+        run: |
-          cargo clippy
+          set +e
-          --all-features
+          cargo clippy \
-          --message-format=json | clippy-sarif | tee rust-clippy-results.sarif | sarif-fmt
+            --workspace \
-        continue-on-error: true
+            --all-targets \
-
+            --all-features \
-      - name: Upload analysis results to GitHub
+            --locked \
-        uses: github/codeql-action/upload-sarif@v4
+            --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@2892aa5e19bbd11bc0cff5427e3b750a04d9e3c2 # 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

@@ -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"
@@ -49,6 +66,12 @@ version = "3.20.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
 
+[[package]]
+name = "bytes"
+version = "1.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
+
 [[package]]
 name = "cc"
 version = "1.3.0"
@@ -74,6 +97,12 @@ version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
 
+[[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"
@@ -96,9 +125,9 @@ dependencies = [
 
 [[package]]
 name = "config"
-version = "0.15.25"
+version = "0.15.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b85f248a4de22d204ceabc6299d89d2c70fbd7f09fea53c06c852369652d8139"
+checksum = "38e9fbc53afcec54441422f1c596cd054655b32e70e1cbac956651b7b175a404"
 dependencies = [
  "pathdiff",
  "serde_core",
@@ -171,9 +200,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.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
+
 [[package]]
 name = "getrandom"
 version = "0.4.3"
@@ -201,11 +244,15 @@ checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
 
 [[package]]
 name = "icmp-socket2"
-version = "0.6.0"
+version = "0.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "86fc005b72eb9dbc1df87318cf906a415d4f4c0d5eea42849b29dd710cdf670d"
+checksum = "816dc56a4f7f76e42b0616818600b05528cd346cb3843276e82df982d5c3ee26"
 dependencies = [
+ "async-trait",
+ "nix",
  "socket2",
+ "thiserror",
+ "tokio",
 ]
 
 [[package]]
@@ -245,9 +292,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"
@@ -286,20 +333,9 @@ 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"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "746873a384ad60adc5db74471dfaba74bd278afbdcfd81db93fafcdfc8b5ca0c"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.119",
-]
 
 [[package]]
 name = "memchr"
@@ -307,6 +343,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 +359,44 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
 
 [[package]]
-name = "netlink-bindings"
+name = "mio"
-version = "0.3.3"
+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 +415,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 +484,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"
@@ -434,9 +541,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 +597,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 3.0.3",
+ "syn 3.0.6",
 ]
 
 [[package]]
@@ -527,6 +634,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"
@@ -564,9 +687,16 @@ dependencies = [
  "rustix",
  "serde",
  "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 = "syn"
 version = "2.0.119"
@@ -580,9 +710,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",
@@ -590,10 +720,10 @@ dependencies = [
 ]
 
 [[package]]
-name = "target-triple"
+name = "target-tuple"
-version = "1.0.1"
+version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3a6bfce3d99adfa72d24750a61f782f3036a81e7f86d8841ee1326deaebd171"
+checksum = "876fef147edbcbddc8ac5cbbba92c7b86519e314e86638596c09673b2ed01e7f"
 
 [[package]]
 name = "termcolor"
@@ -606,22 +736,50 @@ 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]]
@@ -665,15 +823,15 @@ checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2"
 
 [[package]]
 name = "trybuild"
-version = "1.0.120"
+version = "1.0.121"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e605bf6b39357663d8ba4e984f8be8da8df6bb32e81031d6889024ea8fd68e4"
+checksum = "c0cabaa10be1917331a313866bd94526343e03c77bcf69144b62b072ad35d47c"
 dependencies = [
  "glob",
  "serde",
  "serde_derive",
  "serde_json",
- "target-triple",
+ "target-tuple",
  "termcolor",
  "toml",
 ]
@@ -684,7 +842,7 @@ version = "0.1.0"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 3.0.3",
+ "syn 3.0.6",
  "trybuild",
 ]
 
@@ -694,6 +852,12 @@ version = "1.0.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
 
+[[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"

Cargo.toml

@@ -20,15 +20,16 @@ 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"] }
+libc = "0.2"
+rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
 serde = { version = "1", features = ["derive"] }
+thiserror = "2.0"
+tokio = { version = "1", features = ["full"] }
 uci-derive = { path = "crates/uci-derive" }
 
 [dependencies.log]

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/log.rs

@@ -2,10 +2,9 @@
 //
 // SPDX-License-Identifier: MPL-2.0
 
-use log::{Level, LevelFilter, Metadata, Record, SetLoggerError};
+use log::{Level, Metadata, Record, SetLoggerError};
-use rustix::time::ClockId;
 
-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

@@ -9,6 +9,7 @@ extern crate core;
 
 mod baseliner;
 mod log;
+mod measurement;
 mod metrics;
 mod pinger;
 mod pinger_icmp;
@@ -21,6 +22,10 @@ mod time;
 mod util;
 
 use crate::baseliner::Baseliner;
+use crate::measurement::icmp::{IcmpBinding, IcmpEngine};
+use crate::measurement::model::{
+    MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule,
+};
 use crate::metrics::{Metric, Metrics, MetricsSender};
 use crate::pinger::{InFlightProbeCache, PingListener, PingSender};
 use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender};
@@ -30,10 +35,10 @@ use crate::ratecontroller::Ratecontroller;
 use crate::reflector_selector::ReflectorSelector;
 use crate::settings::MeasurementType;
 use crate::settings::Settings;
-use ::log::{debug, info, warn};
+use ::log::{info, warn};
 use flume::RecvTimeoutError;
 use std::collections::HashMap;
-use std::net::IpAddr;
+use std::net::{IpAddr, Ipv4Addr};
 use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
@@ -245,6 +250,24 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     let start_time = Instant::now();
     let probe_id = probe_identifier();
 
+    let (ie_txt, _) = flume::unbounded();
+    let (binding_tx, binding_rx) = flume::unbounded();
+    let engine = IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)?;
+    binding_tx.send(vec![IcmpBinding::new(
+        MeasurementStream {
+            id: MeasurementStreamId(0),
+            peer: IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
+            source: MeasurementSource::IcmpEcho,
+        },
+        ProbeSchedule {
+            period: Duration::from_secs(1),
+            offset: Duration::from_secs(0),
+        },
+    )])?;
+
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    rt.spawn(engine.run());
+
     let ReflectorSetup {
         peers: reflector_peers,
         pool: reflector_pool,

src/measurement/icmp.rs

@@ -0,0 +1,459 @@
+use std::{
+    collections::{HashMap, HashSet},
+    io,
+    net::IpAddr,
+    sync::{Arc, Mutex, atomic::Ordering},
+    thread,
+    time::{Duration, Instant, SystemTime},
+};
+
+use flume::{Receiver, RecvTimeoutError, Sender};
+use icmp_socket2::{
+    AsyncIcmpSocket, IcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet,
+    packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest},
+    tokio::AsyncIcmpV4Socket,
+};
+use thiserror::Error;
+
+use crate::{
+    SHUTDOWN,
+    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,
+}
+
+impl IcmpBinding {
+    pub fn new(stream: MeasurementStream, schedule: ProbeSchedule) -> Self {
+        Self {
+            stream,
+            schedule,
+            next_probe_at: Instant::now(),
+        }
+    }
+}
+
+#[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,
+    next_sequence: 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 {
+            IcmpSocket::bind_device(&mut socket, device)?;
+        }
+
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        if let Some(mark) = settings.measurement_mark {
+            IcmpSocket::set_mark(&mut socket, mark)?;
+        }
+
+        #[cfg(target_os = "freebsd")]
+        if let Some(fib) = settings.measurement_fib {
+            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,
+            next_sequence: 0,
+            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));
+
+        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");
+
+            while binding.next_probe_at <= now {
+                binding.next_probe_at += binding.schedule.period;
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn send_probe(&mut self, stream_id: MeasurementStreamId) -> Result<(), IcmpError> {
+        let stream = self
+            .bindings
+            .get(&stream_id)
+            .expect("binding disappeared")
+            .stream;
+
+        let IpAddr::V4(peer) = stream.peer else {
+            return Err(IcmpError::UnsupportedPeer(stream.peer));
+        };
+
+        let sequence = self.next_sequence;
+        self.next_sequence = self.next_sequence.wrapping_add(1);
+
+        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.inflight.insert(key, InFlightProbe { stream, sent_at });
+
+        if let Err(err) = self.socket.send_to(peer, packet).await {
+            return Err(err.into());
+        }
+
+        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))?;
+
+                Ok(MeasurementObservation {
+                    stream: probe.stream,
+                    started_at: probe.sent_at,
+                    observed_at,
+                    one_way: None,
+                })
+            }
+            Icmpv4Message::TimestampReply {
+                identifier,
+                sequence,
+                originate,
+                receive,
+                transmit,
+            } => {
+                const NON_STANDARD_TIMESTAMP: u32 = 1 << 31;
+
+                if identifier != self.identifier {
+                    return Err(IcmpError::WrongIdentifier {
+                        expected: self.identifier,
+                        found: identifier,
+                    });
+                }
+
+                if receive & NON_STANDARD_TIMESTAMP != 0 || transmit & NON_STANDARD_TIMESTAMP != 0 {
+                    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);
+
+                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(),
+            )),
+        }
+    }
+
+    fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> {
+        const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+
+        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(MeasurementEvent::Loss(MeasurementLoss {
+                    stream: probe.stream,
+                    started_at: probe.sent_at,
+                    detected_at: now,
+                }))
+                .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?;
+
+                    self.handle_reply(
+                        result.packet,
+                        result.peer.as_socket().unwrap().ip(),
+                        result.received_at,
+                        result.kernel_rx_timestamp,
+                    )?;
+                }
+
+                _ = housekeeping.tick() => {
+                    self.expire_inflight(Instant::now())?;
+                }
+            }
+        }
+    }
+}
+
+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/mod.rs

@@ -0,0 +1,2 @@
+pub(crate) mod icmp;
+pub(crate) mod model;

src/measurement/model.rs

@@ -0,0 +1,138 @@
+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,
+}
+
+#[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,
+}
+
+#[allow(dead_code)]
+#[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>,
+}
+
+#[allow(dead_code)]
+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

@@ -4,10 +4,9 @@
 
 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;
@@ -148,7 +147,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 +196,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_icmp.rs

@@ -10,11 +10,10 @@ use std::time::Instant;
 
 use crate::pinger::{PingError, PingListener, PingReply, PingSender};
 use crate::settings::MeasurementType;
-use crate::time::Time;
+use crate::time::{ClockSample, Monotonic};
 use icmp_socket2::Icmpv4Message;
 use icmp_socket2::Icmpv4Packet;
 use icmp_socket2::packet::WithEchoRequest;
-use rustix::time::ClockId;
 
 pub struct PingerICMPEchoListener {}
 
@@ -54,7 +53,7 @@ impl PingListener for PingerICMPEchoListener {
                         }
                     };
 
-                    let clock = Time::new(ClockId::Monotonic);
+                    let clock = ClockSample::<Monotonic>::now();
                     let time_ms = clock.to_milliseconds() as i64;
 
                     let rtt = (time_ms - time_sent) as f64;
@@ -85,7 +84,7 @@ impl PingListener for PingerICMPEchoListener {
 
 impl PingSender for PingerICMPEchoSender {
     fn craft_packet(&self, id: u16, seq: u16) -> (Icmpv4Packet, i64) {
-        let clock = Time::new(ClockId::Monotonic);
+        let clock = ClockSample::<Monotonic>::now();
         let time_ms = clock.to_milliseconds();
         let payload = time_ms.to_be_bytes().to_vec();
 

src/pinger_icmp_ts.rs

@@ -7,11 +7,10 @@
 
 use crate::pinger::{PingError, PingListener, PingReply, PingSender};
 use crate::settings::MeasurementType;
-use crate::time::Time;
+use crate::time::{ClockSample, Realtime};
 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;
 
@@ -19,6 +18,19 @@ pub struct PingerICMPTimestampListener {}
 
 pub struct PingerICMPTimestampSender {}
 
+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
+    }
+}
+
 impl PingListener for PingerICMPTimestampListener {
     // Result: RTT, down time, up time
     fn parse_packet(
@@ -46,12 +58,18 @@ impl PingListener for PingerICMPTimestampListener {
                         });
                     }
 
-                    let time_now = Time::new(ClockId::Realtime);
+                    if receive & 0x8000_0000 != 0 || transmit & 0x8000_0000 != 0 {
-                    let time_since_midnight = time_now.get_time_since_midnight();
+                        return Err(PingError::InvalidPacket(
+                            "non-standard RFC 792 timestamp".into(),
+                        ));
+                    }
 
-                    let rtt = (time_since_midnight - originate as i64) as f64;
+                    let time_since_midnight =
-                    let dl_time = (time_since_midnight - transmit as i64) as f64;
+                        ClockSample::<Realtime>::now().as_time_since_midnight();
-                    let ul_time = (receive as i64 - originate as i64) as f64;
+
+                    let rtt = timestamp_delta(time_since_midnight, originate as i64) as f64;
+                    let dl_time = timestamp_delta(time_since_midnight, transmit as i64) as f64;
+                    let ul_time = timestamp_delta(receive as i64, originate as i64) as f64;
 
                     Ok(PingReply {
                         reflector,
@@ -80,7 +98,7 @@ impl PingListener for PingerICMPTimestampListener {
 
 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();
+        let time_since_midnight = ClockSample::<Realtime>::now().as_time_since_midnight();
         (
             Icmpv4Packet::with_timestamp_request(id, seq, time_since_midnight as u32, 0, 0)
                 .unwrap(),
@@ -88,3 +106,27 @@ impl PingSender for PingerICMPTimestampSender {
         )
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use icmp_socket2::packet::WithTimestampReply;
+
+    #[test]
+    fn timestamps_wrap_at_midnight_in_both_directions() {
+        assert_eq!(timestamp_delta(20, 86_399_900), 120);
+        assert_eq!(timestamp_delta(86_399_900, 20), -120);
+    }
+
+    #[test]
+    fn non_standard_rfc_792_timestamps_are_rejected() {
+        let packet = Icmpv4Packet::with_timestamp_reply(7, 1, 10, 0x8000_0001, 20).unwrap();
+        let result = PingerICMPTimestampListener {}.parse_packet(
+            7,
+            "192.0.2.1".parse().unwrap(),
+            MeasurementType::IcmpTimestamps,
+            packet,
+        );
+        assert!(matches!(result, Err(PingError::InvalidPacket(_))));
+    }
+}

src/platform/unix/uci.rs

@@ -262,7 +262,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());
             }
 
@@ -285,3 +285,29 @@ impl Source for UciSource {
         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/ratecontroller.rs

@@ -13,11 +13,10 @@ use crate::platform::{
     interface_stats_provider, traffic_control_backend,
 };
 use crate::settings::Settings;
-use crate::time::Time;
+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;
@@ -409,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(),
@@ -459,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",

src/settings/mod.rs

@@ -192,7 +192,10 @@ impl Settings {
             })?;
 
             match (record.ip_version, record.reflector_ip) {
-                (4, IpAddr::V4(_)) | (6, IpAddr::V6(_)) => {}
+                (4, IpAddr::V4(_)) => {}
+                (6, address @ IpAddr::V6(_)) => {
+                    bail!("reflector {address} is IPv6; only IPv4 reflectors are supported");
+                }
                 (version, address) => {
                     bail!("reflector {address} has inconsistent IP version {version}");
                 }
@@ -215,6 +218,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 {

src/settings/validation.rs

@@ -153,7 +153,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 +173,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!(
@@ -376,3 +386,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 {
+pub trait Clock {
-    time_s: u64,
+    const ID: ClockId;
-    time_ns: u64,
 }
 
-impl Time {
+pub enum Realtime {}
-    pub fn new(id: ClockId) -> Self {
+pub enum Monotonic {}
-        let time: Timespec = clock_gettime(id);
+
+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,
+        }
+    }
+}