diff --git a/.github/actions/install-risc0/action.yml b/.github/actions/install-risc0/action.yml deleted file mode 100644 index fef3a467..00000000 --- a/.github/actions/install-risc0/action.yml +++ /dev/null @@ -1,10 +0,0 @@ -name: Install risc0 -description: Installs risc0 in the environment -runs: - using: "composite" - steps: - - name: Install risc0 - run: | - curl -L https://risczero.com/install | bash - /home/runner/.risc0/bin/rzup install - shell: bash diff --git a/.github/actions/install-system-deps/action.yml b/.github/actions/install-system-deps/action.yml deleted file mode 100644 index 28c4b41c..00000000 --- a/.github/actions/install-system-deps/action.yml +++ /dev/null @@ -1,10 +0,0 @@ -name: Install system dependencies -description: Installs system dependencies in the environment -runs: - using: "composite" - steps: - - name: Install system dependencies - run: | - sudo apt-get update - sudo apt-get install -y build-essential clang libclang-dev libssl-dev pkg-config - shell: bash diff --git a/.github/actions/run-in-ci-image/action.yml b/.github/actions/run-in-ci-image/action.yml new file mode 100644 index 00000000..3fafe38c --- /dev/null +++ b/.github/actions/run-in-ci-image/action.yml @@ -0,0 +1,54 @@ +name: Run in CI image +description: > + Runs a command inside the CI image on a plain runner, for jobs that need the + host's Docker daemon. `container:` cannot work for those: it puts the + workspace on a path the host daemon cannot resolve, and it forbids + `--network host`. + +inputs: + image: + description: CI image reference, from the ci-image workflow's output. + required: true + run: + description: Command to run inside the image. + required: true + env: + description: Newline-separated NAME=VALUE pairs to pass into the container. + required: false + default: "" + +runs: + using: "composite" + steps: + - name: Run in CI image + shell: bash + env: + INPUT_IMAGE: ${{ inputs.image }} + INPUT_RUN: ${{ inputs.run }} + INPUT_ENV: ${{ inputs.env }} + run: | + set -euo pipefail + + env_args=() + while IFS= read -r pair; do + [[ -z "$pair" ]] && continue + env_args+=(--env "$pair") + done <<< "$INPUT_ENV" + + # Cargo's registry is the only part of CARGO_HOME worth sharing with the + # host: mounting all of it would shadow the tools baked into the image. + mkdir -p "$HOME/.cargo/registry" + + # Mounting the workspace on its own path is what makes this work. The + # daemon resolves compose bind mounts against the host, and the test + # binaries bake CARGO_MANIFEST_DIR in at compile time, so the path has + # to mean the same thing inside and outside the container. + docker run --rm \ + --network host \ + --volume /var/run/docker.sock:/var/run/docker.sock \ + --volume "$PWD:$PWD" \ + --volume "$HOME/.cargo/registry:/usr/local/cargo/registry" \ + --workdir "$PWD" \ + "${env_args[@]}" \ + "$INPUT_IMAGE" \ + bash -e -o pipefail -c "$INPUT_RUN" diff --git a/.github/docker/ci.Dockerfile b/.github/docker/ci.Dockerfile new file mode 100644 index 00000000..a79be15f --- /dev/null +++ b/.github/docker/ci.Dockerfile @@ -0,0 +1,97 @@ +# Image behind the `container:` of the CI jobs. Everything the jobs used to +# install per-run is baked in here instead, so a CI run no longer asks the risc0 +# install servers for anything. +# +# The image tag is a hash of this file plus rust-toolchain.toml, so editing +# either rebuilds the image on the PR that changes it. See ci-image.yml. +# +# Keep the base tag in sync with rust-toolchain.toml. +FROM rust:1.94.0-trixie + +# GitHub sets HOME=/github/home inside container jobs, which hides anything we +# bake into the image's ~. rzup defaults to $HOME/.risc0, so pin it to a fixed +# path; RUSTUP_HOME and CARGO_HOME already point at /usr/local from the base +# image, and rzup honours all three. +ENV RISC0_HOME=/usr/local/risc0 +ENV PATH=/usr/local/risc0/bin:$PATH + +RUN apt-get update && apt-get install -y --no-install-recommends \ + build-essential \ + clang \ + libclang-dev \ + libssl-dev \ + pkg-config \ + curl \ + git \ + jq \ + python3-dev \ + && rm -rf /var/lib/apt/lists/* + +# The runner pre-creates the workspace as another uid, so git in a container job +# calls it dubious. checkout's own fix is --global, which only holds while HOME +# still points at the gitconfig it wrote; --system holds regardless. +RUN git config --system --add safe.directory '*' + +# The base image already has this toolchain at the minimal profile, so the +# `profile = "default"` in rust-toolchain.toml is a no-op here: rustup sees the +# toolchain as installed and never applies it. Add what the jobs need by hand. +COPY rust-toolchain.toml /tmp/toolchain/rust-toolchain.toml +RUN cd /tmp/toolchain \ + && rustup toolchain install \ + && rustup component add clippy rustfmt \ + && rm -rf /tmp/toolchain + +# For `cargo +nightly fmt` in the fmt-rs job. +RUN rustup toolchain install nightly --profile minimal --component rustfmt + +# The bootstrap script only ever drops rzup in $HOME/.risc0/bin, so run it +# against the build-time HOME and keep just the binary. rzup itself then honours +# RISC0_HOME and installs each component under /usr/local. +# +# Versions pinned to the set currently validated in local dev (`rzup show`); +# bare `rzup install` would float all four default components to latest. r0vm +# and cargo-risczero track each other and the risc0-zkvm crate version. +RUN curl -L https://risczero.com/install | bash \ + && mv /root/.risc0/bin/rzup /usr/local/bin/rzup \ + && rm -rf /root/.risc0 \ + && rzup install rust 1.94.1 \ + && rzup install cpp 2024.1.5 \ + && rzup install r0vm 3.0.5 \ + && rzup install cargo-risczero 3.0.5 + +# Copying docker. Static Go binaries, so the alpine-built ones run fine here. +COPY --from=docker:29.6.1-cli /usr/local/bin/docker /usr/local/bin/docker +COPY --from=docker:29.6.1-cli /usr/local/libexec/docker/cli-plugins/docker-compose \ + /usr/local/libexec/docker/cli-plugins/docker-compose +COPY --from=docker:29.6.1-cli /usr/local/libexec/docker/cli-plugins/docker-buildx \ + /usr/local/libexec/docker/cli-plugins/docker-buildx + +# Prebuilt binaries; compiling these from source would dominate the image build. +# Versions pinned so a rebuild of the same image ships the same tools. +ENV BINSTALL_VERSION=1.21.0 +RUN curl -L --proto '=https' --tlsv1.2 -sSf \ + https://raw.githubusercontent.com/cargo-bins/cargo-binstall/main/install-from-binstall-release.sh | bash \ + && cargo binstall --no-confirm \ + cargo-nextest@0.9.140 \ + taplo-cli@0.10.0 \ + cargo-machete@0.9.2 \ + cargo-deny@0.20.2 \ + just@1.56.0 + +# Smoke-test every tool the jobs use. A bad install should fail here, not later +# in some CI job that reuses this cached image. pyo3's `auto-initialize` links +# libpython, and the base image ships python3 without it, so check the .so too. +RUN cargo --version \ + && cargo +nightly fmt --version \ + && cargo clippy --version \ + && ls /usr/lib/*/libpython3*.so \ + && r0vm --version \ + && cargo risczero --version \ + && cargo nextest --version \ + && taplo --version \ + && cargo machete --version \ + && cargo deny --version \ + && just --version \ + && docker --version \ + && docker compose version \ + && docker buildx version diff --git a/.github/workflows/bench-regression.yml b/.github/workflows/bench-regression.yml index d82c0f11..89c18e32 100644 --- a/.github/workflows/bench-regression.yml +++ b/.github/workflows/bench-regression.yml @@ -9,12 +9,25 @@ on: permissions: contents: read pull-requests: write + packages: read name: bench-regression jobs: + ci-image: + permissions: + contents: read + packages: write + uses: ./.github/workflows/ci-image.yml + crypto-primitives: + needs: ci-image runs-on: ubuntu-latest + container: + image: ${{ needs.ci-image.outputs.image }} + credentials: + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} timeout-minutes: 60 steps: - uses: actions/checkout@v5 @@ -24,13 +37,6 @@ jobs: # working tree, so we need the full history. fetch-depth: 0 - - uses: ./.github/actions/install-system-deps - - - uses: ./.github/actions/install-risc0 - - - name: Install active toolchain - run: rustup install - - name: Run criterion-compare against base branch uses: boa-dev/criterion-compare-action@v3 with: diff --git a/.github/workflows/ci-image.yml b/.github/workflows/ci-image.yml new file mode 100644 index 00000000..c93782de --- /dev/null +++ b/.github/workflows/ci-image.yml @@ -0,0 +1,75 @@ +name: CI image + +concurrency: + group: ci-image-${{ github.sha }} + cancel-in-progress: false + +# Resolves the CI image the calling workflow's jobs run in, building and pushing +# it only when it isn't published yet. +on: + workflow_call: + outputs: + image: + description: Image reference to pass to a job's `container:`. + value: ${{ jobs.ci-image.outputs.image }} + +jobs: + ci-image: + runs-on: ubuntu-latest + timeout-minutes: 60 + outputs: + image: ${{ steps.image-ref.outputs.image }} + steps: + - uses: actions/checkout@v5 + with: + ref: ${{ github.event.pull_request.head.sha || github.head_ref }} + + # Tagging by content hash means a PR that touches the Dockerfile or the + # toolchain builds and tests against its own image, while every other PR + # reuses the one already in the registry. + - name: Compute image reference + id: image-ref + run: | + repo="$(echo "${{ github.repository }}" | tr '[:upper:]' '[:lower:]')" + tag="$(cat .github/docker/ci.Dockerfile rust-toolchain.toml | sha256sum | cut -c1-16)" + echo "image=ghcr.io/${repo}/ci:${tag}" >> "$GITHUB_OUTPUT" + echo "CI image: ghcr.io/${repo}/ci:${tag}" + + - name: Log in to GHCR + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Check whether the image is already published + id: probe + run: | + if docker manifest inspect "${{ steps.image-ref.outputs.image }}" > /dev/null 2>&1; then + echo "exists=true" >> "$GITHUB_OUTPUT" + echo "Image already published, skipping build." + else + echo "exists=false" >> "$GITHUB_OUTPUT" + echo "Image not published yet, building it." + fi + + - name: Set up Docker Buildx + if: steps.probe.outputs.exists == 'false' + uses: docker/setup-buildx-action@v3 + + # A pull_request from a fork gets a read-only GITHUB_TOKEN, so this step + # fails there. It only runs when the fork changed the image, which is the + # case that needs a maintainer's eyes anyway. + - name: Build and push CI image + if: steps.probe.outputs.exists == 'false' + uses: docker/build-push-action@v5 + with: + context: . + file: ./.github/docker/ci.Dockerfile + push: true + tags: ${{ steps.image-ref.outputs.image }} + # Links the package to the repo, so it shows up there and inherits its + # access rules. + labels: org.opencontainers.image.source=https://github.com/${{ github.repository }} + cache-from: type=gha,scope=ci-image + cache-to: type=gha,mode=max,scope=ci-image diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 194646f3..ff9b4a8d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,67 +15,91 @@ on: permissions: contents: read pull-requests: read + packages: read name: General jobs: + ci-image: + permissions: + contents: read + packages: write + uses: ./.github/workflows/ci-image.yml + fmt-rs: + needs: ci-image runs-on: ubuntu-latest + container: + image: ${{ needs.ci-image.outputs.image }} + credentials: + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} steps: - uses: actions/checkout@v5 with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - name: Install nightly toolchain for rustfmt - run: rustup install nightly --profile minimal --component rustfmt - - name: Check Rust files are formatted run: cargo +nightly fmt --check fmt-toml: + needs: ci-image runs-on: ubuntu-latest + container: + image: ${{ needs.ci-image.outputs.image }} + credentials: + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} steps: - uses: actions/checkout@v5 with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - name: Install taplo-cli - run: cargo install --locked taplo-cli - + # No path argument: taplo reads a `.` as a literal file, excludes it for + # not being a .toml, and checks nothing. - name: Check TOML files are formatted - run: taplo fmt --check . + run: taplo fmt --check machete: + needs: ci-image runs-on: ubuntu-latest + container: + image: ${{ needs.ci-image.outputs.image }} + credentials: + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} steps: - uses: actions/checkout@v5 with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - name: Install active toolchain - run: rustup install - - - name: Install cargo-machete - run: cargo install cargo-machete - - name: Check for unused dependencies run: cargo machete deny: + needs: ci-image runs-on: ubuntu-latest + container: + image: ${{ needs.ci-image.outputs.image }} + credentials: + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} steps: - uses: actions/checkout@v5 with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - name: Install cargo-deny - run: cargo install --locked cargo-deny - - name: Check licenses and advisories run: cargo deny check lint: + needs: ci-image runs-on: ubuntu-latest + container: + image: ${{ needs.ci-image.outputs.image }} + credentials: + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} timeout-minutes: 60 name: lint @@ -84,13 +108,6 @@ jobs: with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - uses: ./.github/actions/install-system-deps - - - uses: ./.github/actions/install-risc0 - - - name: Install active toolchain - run: rustup install - - name: Restore Rust cache uses: Swatinem/rust-cache@v2 with: @@ -108,36 +125,35 @@ jobs: run: cargo clippy -p "*program" -- -D warnings unit-tests: + needs: ci-image runs-on: ubuntu-latest + container: + image: ${{ needs.ci-image.outputs.image }} + credentials: + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} timeout-minutes: 60 steps: - uses: actions/checkout@v5 with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - uses: ./.github/actions/install-system-deps - - - uses: ./.github/actions/install-risc0 - - - name: Install active toolchain - run: rustup install - - name: Restore Rust cache uses: Swatinem/rust-cache@v2 with: shared-key: ci-rust-cache save-if: ${{ github.ref == 'refs/heads/main' || github.ref == 'refs/heads/dev' }} - - name: Install nextest - run: cargo install --locked cargo-nextest - - name: Run tests env: RISC0_DEV_MODE: "1" RUST_LOG: "info" run: cargo nextest run --workspace --exclude integration_tests --exclude test_fixtures --all-features + # Not a `container:` job: the tests drive the host's Docker daemon through + # testcontainers, so they run in the CI image via `run-in-ci-image` instead. test-fixtures-tests: + needs: ci-image runs-on: ubuntu-latest timeout-minutes: 60 steps: @@ -145,12 +161,12 @@ jobs: with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - uses: ./.github/actions/install-system-deps - - - uses: ./.github/actions/install-risc0 - - - name: Install active toolchain - run: rustup install + - name: Log in to GHCR + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} - name: Restore Rust cache uses: Swatinem/rust-cache@v2 @@ -158,16 +174,20 @@ jobs: shared-key: ci-rust-cache save-if: ${{ github.ref == 'refs/heads/main' || github.ref == 'refs/heads/dev' }} - - name: Install nextest - run: cargo install --locked cargo-nextest - - name: Run test_fixtures tests - env: - RISC0_DEV_MODE: "1" - RUST_LOG: "info" - run: cargo nextest run -p test_fixtures + uses: ./.github/actions/run-in-ci-image + with: + image: ${{ needs.ci-image.outputs.image }} + env: | + RISC0_DEV_MODE=1 + RUST_LOG=info + run: cargo nextest run -p test_fixtures + # Not a `container:` job: the test binaries bake CARGO_MANIFEST_DIR in at + # compile time, so the archive has to be built on the same path the matrix + # jobs later run it from. integration-tests-prebuild: + needs: ci-image runs-on: ubuntu-latest outputs: targets: ${{ steps.discover-targets.outputs.targets }} @@ -176,12 +196,12 @@ jobs: with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - uses: ./.github/actions/install-system-deps - - - uses: ./.github/actions/install-risc0 - - - name: Install active toolchain - run: rustup install + - name: Log in to GHCR + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} - name: Restore Rust cache uses: Swatinem/rust-cache@v2 @@ -189,13 +209,12 @@ jobs: shared-key: ci-rust-cache save-if: ${{ github.ref == 'refs/heads/main' || github.ref == 'refs/heads/dev' }} - - name: Install nextest - run: cargo install --locked cargo-nextest - - name: Build integration test archive - env: - RISC0_DEV_MODE: "1" - run: cargo nextest archive -p integration_tests --archive-file integration-tests.tar.zst --no-pager + uses: ./.github/actions/run-in-ci-image + with: + image: ${{ needs.ci-image.outputs.image }} + env: RISC0_DEV_MODE=1 + run: cargo nextest archive -p integration_tests --archive-file integration-tests.tar.zst --no-pager - name: Upload integration test archive uses: actions/upload-artifact@v4 @@ -203,15 +222,22 @@ jobs: name: integration-tests-archive path: integration-tests.tar.zst + - name: List integration test binaries + uses: ./.github/actions/run-in-ci-image + with: + image: ${{ needs.ci-image.outputs.image }} + run: | + cargo nextest list \ + --archive-file integration-tests.tar.zst \ + --list-type binaries-only \ + --message-format json \ + --no-pager > integration-tests-binaries.json + + # $GITHUB_OUTPUT is outside the mounted workspace, so the container cannot + # write to it. jq is on the runner, so do this pass here. - name: Discover integration test targets from archive id: discover-targets run: | - cargo nextest list \ - --archive-file integration-tests.tar.zst \ - --list-type binaries-only \ - --message-format json \ - --no-pager > integration-tests-binaries.json - targets_json="$(jq -c '[."rust-binaries" | to_entries[] | select(.value.kind == "test" and .value."binary-name" != "tps") | .value."binary-name"] | sort | unique' integration-tests-binaries.json)" if [[ "$targets_json" == "[]" ]]; then @@ -223,7 +249,7 @@ jobs: echo "Discovered integration targets: $targets_json" integration-tests: - needs: [test-fixtures-tests, integration-tests-prebuild] + needs: [ci-image, test-fixtures-tests, integration-tests-prebuild] runs-on: ubuntu-latest timeout-minutes: 90 strategy: @@ -236,28 +262,29 @@ jobs: with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - uses: ./.github/actions/install-system-deps - - - uses: ./.github/actions/install-risc0 - - - name: Install active toolchain - run: rustup install + - name: Log in to GHCR + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} - name: Download integration test archive uses: actions/download-artifact@v4 with: name: integration-tests-archive - - name: Install nextest - run: cargo install --locked cargo-nextest - - name: Run tests - env: - RISC0_DEV_MODE: "1" - RUST_LOG: "info" - run: cargo nextest run --archive-file integration-tests.tar.zst -E "binary(${{ matrix.target }})" + uses: ./.github/actions/run-in-ci-image + with: + image: ${{ needs.ci-image.outputs.image }} + env: | + RISC0_DEV_MODE=1 + RUST_LOG=info + run: cargo nextest run --archive-file integration-tests.tar.zst -E "binary(${{ matrix.target }})" valid-proof-test: + needs: ci-image runs-on: ubuntu-latest timeout-minutes: 90 steps: @@ -265,12 +292,12 @@ jobs: with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - uses: ./.github/actions/install-system-deps - - - uses: ./.github/actions/install-risc0 - - - name: Install active toolchain - run: rustup install + - name: Log in to GHCR + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} - name: Restore Rust cache uses: Swatinem/rust-cache@v2 @@ -279,11 +306,16 @@ jobs: save-if: ${{ github.ref == 'refs/heads/main' || github.ref == 'refs/heads/dev' }} - name: Test valid proof - env: - RUST_LOG: "info" - run: cargo test -p integration_tests -- --exact private::private_transfer_to_owned_account + uses: ./.github/actions/run-in-ci-image + with: + image: ${{ needs.ci-image.outputs.image }} + env: RUST_LOG=info + run: cargo test -p integration_tests -- --exact private::private_transfer_to_owned_account + # `just build-artifacts` drives the host's Docker daemon via `cargo risczero + # build`, so it goes through the image rather than `container:`. artifacts: + needs: ci-image runs-on: ubuntu-latest timeout-minutes: 60 @@ -293,7 +325,12 @@ jobs: with: ref: ${{ github.event.pull_request.head.sha || github.head_ref }} - - uses: ./.github/actions/install-risc0 + - name: Log in to GHCR + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} - name: Restore Rust cache uses: Swatinem/rust-cache@v2 @@ -301,11 +338,11 @@ jobs: shared-key: ci-rust-cache save-if: ${{ github.ref == 'refs/heads/main' || github.ref == 'refs/heads/dev' }} - - name: Install just - run: cargo install --locked just - - name: Build artifacts - run: just build-artifacts + uses: ./.github/actions/run-in-ci-image + with: + image: ${{ needs.ci-image.outputs.image }} + run: just build-artifacts - name: Check if artifacts match repository run: | diff --git a/Justfile b/Justfile index cf7c833e..277403c7 100644 --- a/Justfile +++ b/Justfile @@ -131,5 +131,6 @@ clean: rm -rf lez/sequencer/service/rocksdb rm -rf lez/indexer/service/rocksdb* rm -rf lez/wallet/configs/debug/storage.json + rm -rf lez/wallet/configs/debug/statistics.json rm -rf rocksdb* cd bedrock && docker compose down -v diff --git a/flake.nix b/flake.nix index 0aaad40c..d0058bcb 100644 --- a/flake.nix +++ b/flake.nix @@ -117,6 +117,7 @@ done if [ -n "$tool" ]; then unset DEVELOPER_DIR SDKROOT + export xcrun_nocache=1 fi exec /usr/bin/xcrun "$@" ''; diff --git a/integration_tests/tests/bridge.rs b/integration_tests/tests/bridge.rs index 68088bc0..8da35f4d 100644 --- a/integration_tests/tests/bridge.rs +++ b/integration_tests/tests/bridge.rs @@ -440,7 +440,7 @@ async fn bedrock_deposit_claim_and_withdraw_round_trip_succeeds() -> anyhow::Res // Now claim funds from vault back to recipient let nonces = ctx - .wallet_mut() + .wallet() .get_accounts_nonces(&[recipient_id]) .await .context("Failed to get nonce for vault claim")?; diff --git a/integration_tests/tests/wallet_ffi.rs b/integration_tests/tests/wallet_ffi.rs index 3b8ea05e..a36eb069 100644 --- a/integration_tests/tests/wallet_ffi.rs +++ b/integration_tests/tests/wallet_ffi.rs @@ -28,7 +28,6 @@ use lee::{ }; use lee_core::program::DEFAULT_PROGRAM_ID; use log::info; -use tempfile::tempdir; use wallet::{account::HumanReadableAccount, program_facades::vault::Vault}; use wallet_ffi::{ FfiAccount, FfiAccountIdWithPrivacy, FfiAccountIdentity, FfiAccountList, FfiBytes32, @@ -371,28 +370,6 @@ fn new_wallet_ffi_with_test_context_config( Ok(create_wallet_result) } -fn new_wallet_ffi_with_default_config(password: &str) -> Result { - let tempdir = tempdir()?; - let config_path = tempdir.path().join("wallet_config.json"); - let storage_path = tempdir.path().join("storage.json"); - let metrics_path = tempdir.path().join("metrics.json"); - let config_path_c = CString::new(config_path.to_str().unwrap())?; - let storage_path_c = CString::new(storage_path.to_str().unwrap())?; - let metrics_path_c = CString::new(metrics_path.to_str().unwrap())?; - let password = CString::new(password)?; - - let create_wallet_result = unsafe { - wallet_ffi_create_new( - config_path_c.as_ptr(), - storage_path_c.as_ptr(), - metrics_path_c.as_ptr(), - password.as_ptr(), - ) - }; - - Ok(create_wallet_result) -} - fn load_existing_ffi_wallet(home: &Path) -> Result<*mut WalletHandle> { let config_path = home.join("wallet_config.json"); let storage_path = home.join("storage.json"); @@ -412,17 +389,18 @@ fn load_existing_ffi_wallet(home: &Path) -> Result<*mut WalletHandle> { #[test] fn wallet_ffi_create_public_accounts() -> Result<()> { - let password = "password_for_tests"; + let ctx = BlockingTestContext::new()?; let n_accounts = 10; // Create `n_accounts` public accounts with wallet FFI let new_public_account_ids_ffi = unsafe { let mut account_ids = Vec::new(); + let home = tempfile::tempdir()?; let FfiCreateWalletOutput { wallet: wallet_ffi_handle, mnemonic: _, - } = new_wallet_ffi_with_default_config(password)?; + } = new_wallet_ffi_with_test_context_config(&ctx, home.path())?; for _ in 0..n_accounts { let mut out_account_id = FfiBytes32::from_bytes([0; 32]); wallet_ffi_create_account_public(wallet_ffi_handle, &raw mut out_account_id).unwrap(); @@ -452,16 +430,17 @@ fn wallet_ffi_create_public_accounts() -> Result<()> { #[test] fn wallet_ffi_create_private_accounts() -> Result<()> { - let password = "password_for_tests"; + let ctx = BlockingTestContext::new()?; let n_accounts = 10; // Create `n_accounts` receiving keys with wallet FFI let new_npks_ffi = unsafe { let mut npks = Vec::new(); + let home = tempfile::tempdir()?; let FfiCreateWalletOutput { wallet: wallet_ffi_handle, mnemonic: _, - } = new_wallet_ffi_with_default_config(password)?; + } = new_wallet_ffi_with_test_context_config(&ctx, home.path())?; for _ in 0..n_accounts { let mut out_keys = FfiPrivateAccountKeys::default(); wallet_ffi_create_private_accounts_key(wallet_ffi_handle, &raw mut out_keys).unwrap(); @@ -526,14 +505,14 @@ fn wallet_ffi_save_and_load_persistent_storage() -> Result<()> { #[test] fn test_wallet_ffi_list_accounts() -> Result<()> { - let password = "password_for_tests"; - + let ctx = BlockingTestContext::new()?; // Create the wallet FFI and track which account IDs were created as public/private let (wallet_ffi_handle, created_public_ids) = unsafe { + let home = tempfile::tempdir()?; let FfiCreateWalletOutput { wallet: handle, mnemonic: _, - } = new_wallet_ffi_with_default_config(password)?; + } = new_wallet_ffi_with_test_context_config(&ctx, home.path())?; let mut public_ids: Vec<[u8; 32]> = Vec::new(); // Create 5 public accounts and 5 receiving keys diff --git a/lez/wallet-ffi/src/lib.rs b/lez/wallet-ffi/src/lib.rs index 361ae672..c91185e6 100644 --- a/lez/wallet-ffi/src/lib.rs +++ b/lez/wallet-ffi/src/lib.rs @@ -47,6 +47,7 @@ pub mod error; pub mod generic_transaction; pub mod keys; pub mod label; +pub mod pda; pub mod pinata; pub mod program_deployment; pub mod sync; diff --git a/lez/wallet-ffi/src/pda.rs b/lez/wallet-ffi/src/pda.rs new file mode 100644 index 00000000..35a9482b --- /dev/null +++ b/lez/wallet-ffi/src/pda.rs @@ -0,0 +1,142 @@ +use lee::AccountId; + +use crate::{ + error::WalletFfiError, FfiBytes32, FfiNullifierPublicKey, FfiPdaSeed, FfiPrivateAccountKeys, + FfiProgramId, FfiU128, +}; + +/// Produce account id for public PDA. +/// +/// # Parameters +/// - `program_id`: Id of the owner program +/// - `pda_seed`: 32 byte seed +/// +/// # Returns +/// - `FfiBytes32` representing account id bytes +#[no_mangle] +pub extern "C" fn wallet_ffi_account_id_for_public_pda( + program_id: FfiProgramId, + pda_seed: FfiPdaSeed, +) -> FfiBytes32 { + AccountId::for_public_pda(&program_id.data, &pda_seed.into()).into() +} + +/// Produce account id for private PDA. +/// +/// # Parameters +/// - `program_id`: Id of the owner program +/// - `pda_seed`: 32 byte seed +/// - `npk`: 32 byte nullifier public key (can be obtained from +/// `wallet_ffi_get_private_account_keys`) +/// - `viewing_public_key`: pointer to u8 (can be obtained from +/// `wallet_ffi_get_private_account_keys`) +/// - `viewing_public_key_len`: length of a `viewing_public_key` (can be obtained from +/// `wallet_ffi_get_private_account_keys`), must be `1184` +/// - `identifier`: little endian encoded `u128` +/// - `account_id`: valid pointer to `FfiBytes32` +/// +/// # Returns +/// - `Success` on successful parsing +/// - Error code on failure +/// +/// # Safety +/// - `viewing_public_key` must be a valid pointer to a `u8` +/// - `account_id` must be a valid pointer to a `FfiBytes32` struct +#[no_mangle] +pub unsafe extern "C" fn wallet_ffi_account_id_for_private_pda( + program_id: FfiProgramId, + pda_seed: FfiPdaSeed, + npk: FfiNullifierPublicKey, + viewing_public_key: *const u8, + viewing_public_key_len: usize, + identifier: FfiU128, + account_id: *mut FfiBytes32, +) -> WalletFfiError { + if viewing_public_key.is_null() { + return WalletFfiError::NullPointer; + } + + let ffi_private_keys = FfiPrivateAccountKeys { + nullifier_public_key: npk, + viewing_public_key, + viewing_public_key_len, + }; + + let vpk = ffi_private_keys.vpk(); + + if vpk.is_err() { + return vpk.err().unwrap(); + } + + unsafe { + *account_id = AccountId::for_private_pda( + &program_id.data, + &pda_seed.into(), + &ffi_private_keys.npk(), + &vpk.unwrap(), + identifier.into(), + ) + .into(); + } + + WalletFfiError::Success +} + +#[cfg(test)] +mod tests { + use lee::AccountId; + use lee_core::{encryption::ViewingPublicKey, NullifierPublicKey}; + use vault_core::PdaSeed; + + use crate::{ + error::WalletFfiError, + pda::{wallet_ffi_account_id_for_private_pda, wallet_ffi_account_id_for_public_pda}, + FfiBytes32, + }; + + #[test] + fn public_pda_consistent_derivation() { + let program_id = [100_u32, 101, 102, 103, 104, 105, 106, 107]; + let pda_seed = PdaSeed::new([42; 32]); + + let pda_id = AccountId::for_public_pda(&program_id, &pda_seed); + let ffi_pda_id = wallet_ffi_account_id_for_public_pda(program_id.into(), pda_seed.into()); + + assert_eq!(pda_id.into_value(), ffi_pda_id.data); + } + + #[test] + fn private_pda_consistent_derivation() { + let program_id = [100_u32, 101, 102, 103, 104, 105, 106, 107]; + let pda_seed = PdaSeed::new([42; 32]); + let vpk = ViewingPublicKey::from_bytes(vec![43; 1184]).unwrap(); + let npk = NullifierPublicKey([44; 32]); + let identifier = 100_000_u128; + + let pda_id = AccountId::for_private_pda(&program_id, &pda_seed, &npk, &vpk, identifier); + + let vpk_ptr = Box::into_raw(vpk.to_bytes().to_vec().into_boxed_slice()) as *const u8; + + let mut ffi_pda_id_base = FfiBytes32 { data: [0; 32] }; + let ffi_pda_id = &raw mut ffi_pda_id_base; + + let err = unsafe { + wallet_ffi_account_id_for_private_pda( + program_id.into(), + pda_seed.into(), + npk.into(), + vpk_ptr, + 1184, + identifier.into(), + ffi_pda_id, + ) + }; + + assert_eq!(err, WalletFfiError::Success); + + assert_eq!(pda_id.into_value(), unsafe { (*ffi_pda_id).data }); + + let vpk_slice = unsafe { std::slice::from_raw_parts_mut(vpk_ptr.cast_mut(), 1184) }; + drop(unsafe { Box::from_raw(std::ptr::from_mut(vpk_slice)) }); + } +} diff --git a/lez/wallet-ffi/src/types.rs b/lez/wallet-ffi/src/types.rs index 4f041c55..3779ba01 100644 --- a/lez/wallet-ffi/src/types.rs +++ b/lez/wallet-ffi/src/types.rs @@ -8,7 +8,7 @@ use std::{ }; use lee::{Data, ProgramId, SharedSecretKey}; -use lee_core::{encryption::MlKem768EncapsulationKey, NullifierPublicKey}; +use lee_core::{encryption::MlKem768EncapsulationKey, program::PdaSeed, NullifierPublicKey}; use wallet::{account::AccountIdWithPrivacy, AccountIdentity}; use crate::error::WalletFfiError; @@ -29,6 +29,36 @@ pub struct FfiBytes32 { pub data: [u8; 32], } +pub type FfiPdaSeed = FfiBytes32; + +impl From for PdaSeed { + fn from(value: FfiPdaSeed) -> Self { + Self::new(value.data) + } +} + +impl From for FfiPdaSeed { + fn from(value: PdaSeed) -> Self { + Self { + data: *value.as_bytes(), + } + } +} + +pub type FfiNullifierPublicKey = FfiBytes32; + +impl From for NullifierPublicKey { + fn from(value: FfiNullifierPublicKey) -> Self { + Self(value.data) + } +} + +impl From for FfiNullifierPublicKey { + fn from(value: NullifierPublicKey) -> Self { + Self { data: value.0 } + } +} + /// Program ID - 8 u32 values (32 bytes total). #[repr(C)] #[derive(Clone, Copy, Default)] diff --git a/lez/wallet-ffi/src/wallet.rs b/lez/wallet-ffi/src/wallet.rs index 703da3a5..b2846912 100644 --- a/lez/wallet-ffi/src/wallet.rs +++ b/lez/wallet-ffi/src/wallet.rs @@ -86,7 +86,7 @@ fn c_str_to_path(ptr: *const c_char, name: &str) -> Result Result FfiCreateWalletOutput { let Ok(config_path) = c_str_to_path(config_path, "config_path") else { @@ -114,14 +114,14 @@ pub unsafe extern "C" fn wallet_ffi_create_new( return FfiCreateWalletOutput::default(); }; - let Ok(metrics_path) = c_str_to_path(metrics_path, "metrics_path") else { + let Ok(statistics_path) = c_str_to_path(statistics_path, "statistics_path") else { return FfiCreateWalletOutput::default(); }; match block_on(WalletCore::new_init_storage( config_path, storage_path, - metrics_path, + statistics_path, None, &password, )) { @@ -154,9 +154,9 @@ pub unsafe extern "C" fn wallet_ffi_create_new( /// This loads a wallet that was previously created with `wallet_ffi_create_new()`. /// /// # Parameters -/// - `handle` - Valid wallet handle /// - `config_path`: Path to the wallet configuration file (JSON) -/// - `metrics_path`: Path to the wallet metrics file (JSON) +/// - `storage_path`: Path to the wallet storage (JSON) +/// - `statistics_path`: Path to the wallet statistics file (JSON) /// /// # Returns /// - Opaque wallet handle on success @@ -164,12 +164,11 @@ pub unsafe extern "C" fn wallet_ffi_create_new( /// /// # Safety /// All string parameters must be valid null-terminated UTF-8 strings. -/// `handle` must be a valid wallet handle from `wallet_ffi_create_new` or `wallet_ffi_open`. #[no_mangle] pub unsafe extern "C" fn wallet_ffi_open( config_path: *const c_char, storage_path: *const c_char, - metrics_path: *const c_char, + statistics_path: *const c_char, ) -> *mut WalletHandle { let Ok(config_path) = c_str_to_path(config_path, "config_path") else { return ptr::null_mut(); @@ -179,14 +178,14 @@ pub unsafe extern "C" fn wallet_ffi_open( return ptr::null_mut(); }; - let Ok(metrics_path) = c_str_to_path(metrics_path, "metrics_path") else { + let Ok(statistics_path) = c_str_to_path(statistics_path, "statistics_path") else { return ptr::null_mut(); }; match block_on(WalletCore::new_update_chain( config_path, storage_path, - metrics_path, + statistics_path, None, )) { Ok(core) => { @@ -250,7 +249,7 @@ pub unsafe extern "C" fn wallet_ffi_save(handle: *mut WalletHandle) -> WalletFfi match wallet .store_persistent_data() - .and_then(|()| block_on(wallet.store_metrics())) + .and_then(|()| block_on(wallet.client_rotation())) { Ok(()) => WalletFfiError::Success, Err(e) => { diff --git a/lez/wallet-ffi/wallet_ffi.h b/lez/wallet-ffi/wallet_ffi.h index e977e946..bbd7da1f 100644 --- a/lez/wallet-ffi/wallet_ffi.h +++ b/lez/wallet-ffi/wallet_ffi.h @@ -320,6 +320,10 @@ typedef struct LabelList { enum WalletFfiError error; } LabelList; +typedef struct FfiBytes32 FfiPdaSeed; + +typedef struct FfiBytes32 FfiNullifierPublicKey; + typedef struct FfiCreateWalletOutput { struct WalletHandle *wallet; /** @@ -914,6 +918,50 @@ struct LabelList wallet_ffi_get_all_labels_for_account(struct WalletHandle *hand */ enum WalletFfiError wallet_ffi_free_label_list(struct LabelList *label_list); +/** + * Produce account id for public PDA. + * + * # Parameters + * - `program_id`: Id of the owner program + * - `pda_seed`: 32 byte seed + * + * # Returns + * - `FfiBytes32` representing account id bytes + */ +struct FfiBytes32 wallet_ffi_account_id_for_public_pda(struct FfiProgramId program_id, + FfiPdaSeed pda_seed); + +/** + * Produce account id for private PDA. + * + * # Parameters + * - `program_id`: Id of the owner program + * - `pda_seed`: 32 byte seed + * - `npk`: 32 byte nullifier public key (can be obtained from + * `wallet_ffi_get_private_account_keys`) + * - `viewing_public_key`: pointer to u8 (can be obtained from + * `wallet_ffi_get_private_account_keys`) + * - `viewing_public_key_len`: length of a `viewing_public_key` (can be obtained from + * `wallet_ffi_get_private_account_keys`), must be `1184` + * - `identifier`: little endian encoded `u128` + * - `account_id`: valid pointer to `FfiBytes32` + * + * # Returns + * - `Success` on successful parsing + * - Error code on failure + * + * # Safety + * - `viewing_public_key` must be a valid pointer to a `u8` + * - `account_id` must be a valid pointer to a `FfiBytes32` struct + */ +enum WalletFfiError wallet_ffi_account_id_for_private_pda(struct FfiProgramId program_id, + FfiPdaSeed pda_seed, + FfiNullifierPublicKey npk, + const uint8_t *viewing_public_key, + uintptr_t viewing_public_key_len, + struct FfiU128 identifier, + struct FfiBytes32 *account_id); + /** * Claim a pinata reward using a public transaction. * @@ -1564,7 +1612,7 @@ enum WalletFfiError wallet_ffi_vault_claim_private(struct WalletHandle *handle, * # Parameters * - `config_path`: Path to the wallet configuration file (JSON) * - `storage_path`: Path where wallet data will be stored - * - `metrics_path`: Path to the wallet metrics file (JSON) + * - `statistics_path`: Path to the wallet statistics file (JSON) * - `password`: Password for encrypting the wallet seed * * # Returns @@ -1576,7 +1624,7 @@ enum WalletFfiError wallet_ffi_vault_claim_private(struct WalletHandle *handle, */ struct FfiCreateWalletOutput wallet_ffi_create_new(const char *config_path, const char *storage_path, - const char *metrics_path, + const char *statistics_path, const char *password); /** @@ -1585,9 +1633,9 @@ struct FfiCreateWalletOutput wallet_ffi_create_new(const char *config_path, * This loads a wallet that was previously created with `wallet_ffi_create_new()`. * * # Parameters - * - `handle` - Valid wallet handle * - `config_path`: Path to the wallet configuration file (JSON) - * - `metrics_path`: Path to the wallet metrics file (JSON) + * - `storage_path`: Path to the wallet storage (JSON) + * - `statistics_path`: Path to the wallet statistics file (JSON) * * # Returns * - Opaque wallet handle on success @@ -1595,11 +1643,10 @@ struct FfiCreateWalletOutput wallet_ffi_create_new(const char *config_path, * * # Safety * All string parameters must be valid null-terminated UTF-8 strings. - * `handle` must be a valid wallet handle from `wallet_ffi_create_new` or `wallet_ffi_open`. */ struct WalletHandle *wallet_ffi_open(const char *config_path, const char *storage_path, - const char *metrics_path); + const char *statistics_path); /** * Destroy a wallet handle and free its resources. diff --git a/lez/wallet/configs/debug/wallet_config.json b/lez/wallet/configs/debug/wallet_config.json index 995007fe..c8bd872b 100644 --- a/lez/wallet/configs/debug/wallet_config.json +++ b/lez/wallet/configs/debug/wallet_config.json @@ -1,9 +1,10 @@ { - "sequencers_conn_data": [{ + "sequencers": [{ "sequencer_addr": "http://127.0.0.1:3040" }], "seq_poll_timeout": "30s", "seq_tx_poll_max_blocks": 15, "seq_poll_max_retries": 10, - "seq_block_poll_max_amount": 100 + "seq_block_poll_max_amount": 100, + "calibration_limit": 100 } \ No newline at end of file diff --git a/lez/wallet/src/cli/config.rs b/lez/wallet/src/cli/config.rs index b6d0c523..8c8c4b5c 100644 --- a/lez/wallet/src/cli/config.rs +++ b/lez/wallet/src/cli/config.rs @@ -36,8 +36,8 @@ impl ConfigSubcommand { println!("{config_str}"); } else if let Some(key) = key { match key.as_str() { - "sequencer_addr" => { - println!("{:?}", config.sequencers_conn_data); + "sequencers" => { + println!("{:?}", config.sequencers); } "seq_poll_timeout" => { println!("{:?}", config.seq_poll_timeout); diff --git a/lez/wallet/src/cli/mod.rs b/lez/wallet/src/cli/mod.rs index d01cd860..d436de52 100644 --- a/lez/wallet/src/cli/mod.rs +++ b/lez/wallet/src/cli/mod.rs @@ -107,9 +107,6 @@ pub struct Args { /// Continious run flag. #[arg(short, long)] pub continuous_run: bool, - /// Basic authentication in the format `user` or `user:password`. - #[arg(long)] - pub auth: Option, /// Wallet command. #[command(subcommand)] pub command: Option, @@ -296,11 +293,11 @@ pub async fn execute_subcommand( }; // Kind of a sledgehammer solution, but it is not clear if there is the case to not store - // metrics + // statistics wallet_core - .store_metrics() + .client_rotation() .await - .context("Failed to store metrics")?; + .context("Failed to rotate wallet")?; Ok(subcommand_ret) } diff --git a/lez/wallet/src/config.rs b/lez/wallet/src/config.rs index 80a68756..354b322e 100644 --- a/lez/wallet/src/config.rs +++ b/lez/wallet/src/config.rs @@ -7,6 +7,8 @@ use log::warn; use serde::{Deserialize, Serialize}; use url::Url; +const DEFAULT_CALLIBRATION_LIMIT: usize = 100; + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct SequencerConnectionData { /// Connection data of all known sequencers. @@ -55,7 +57,7 @@ impl Default for MultiSequencerClientConfig { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WalletConfig { /// Connection data of all known sequencers. - pub sequencers_conn_data: Vec, + pub sequencers: Vec, /// Sequencer polling duration for new blocks. #[serde(with = "humantime_serde")] pub seq_poll_timeout: Duration, @@ -65,13 +67,14 @@ pub struct WalletConfig { pub seq_poll_max_retries: u64, /// Max amount of blocks to poll in one request. pub seq_block_poll_max_amount: u64, + /// CURR_TODO: Add default serialization pub multi_sequencer_client_config: MultiSequencerClientConfig, } impl Default for WalletConfig { fn default() -> Self { Self { - sequencers_conn_data: vec![SequencerConnectionData { + sequencers: vec![SequencerConnectionData { sequencer_addr: "http://127.0.0.1:3040".parse().unwrap(), basic_auth: None, }], @@ -124,7 +127,7 @@ impl WalletConfig { pub fn apply_overrides(&mut self, overrides: WalletConfigOverrides) { let Self { - sequencers_conn_data, + sequencers, seq_poll_timeout, seq_tx_poll_max_blocks, seq_poll_max_retries, @@ -133,7 +136,7 @@ impl WalletConfig { } = self; let WalletConfigOverrides { - sequencers_conn_data: o_sequencers_conn_data, + sequencers: o_sequencers, seq_poll_timeout: o_seq_poll_timeout, seq_tx_poll_max_blocks: o_seq_tx_poll_max_blocks, seq_poll_max_retries: o_seq_poll_max_retries, @@ -141,9 +144,9 @@ impl WalletConfig { multi_sequencer_client_config: o_multi_sequencer_client_config, } = overrides; - if let Some(v) = o_sequencers_conn_data { - warn!("Overriding wallet config 'sequencers_conn_data' to {v:?}"); - *sequencers_conn_data = v; + if let Some(v) = o_sequencers { + warn!("Overriding wallet config 'sequencers' to {v:?}"); + *sequencers = v; } if let Some(v) = o_seq_poll_timeout { warn!("Overriding wallet config 'seq_poll_timeout' to {v:?}"); @@ -167,3 +170,7 @@ impl WalletConfig { } } } + +const fn default_calibration_limit() -> usize { + DEFAULT_CALLIBRATION_LIMIT +} diff --git a/lez/wallet/src/helperfunctions.rs b/lez/wallet/src/helperfunctions.rs index 3898d94c..0af99c1f 100644 --- a/lez/wallet/src/helperfunctions.rs +++ b/lez/wallet/src/helperfunctions.rs @@ -66,13 +66,13 @@ pub fn fetch_persistent_storage_path() -> Result { Ok(accs_path) } -/// Fetch path to metrics storage from default home. +/// Fetch path to statistics storage from default home. /// /// File must be created through setup beforehand. -pub fn fetch_metrics_path() -> Result { +pub fn fetch_statistics_path() -> Result { let home = get_home()?; - let metrics_path = home.join("metrics.json"); - Ok(metrics_path) + let statistics_path = home.join("statistics.json"); + Ok(statistics_path) } #[expect(dead_code, reason = "Maybe used later")] diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index 055a0bfa..588f9e0f 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -10,7 +10,6 @@ use std::{ collections::{BTreeMap, HashMap, HashSet}, path::PathBuf, - sync::Arc, }; pub use account_manager::AccountIdentity; @@ -33,13 +32,13 @@ use lee_core::{ use log::info; use sequencer_service_rpc::{RpcClient as _, SequencerClient}; use storage::Storage; -use tokio::{io::AsyncWriteExt as _, sync::RwLock}; +use tokio::io::AsyncWriteExt as _; use url::Url; use crate::{ account::{AccountIdWithPrivacy, Label}, config::WalletConfigOverrides, - multi_client::{Metrics, MetricsUpdate, MultiSequencerClient, extract_metrics_from_path}, + multi_client::{MultiSequencerClient, Statistics, extract_statistics_from_path}, poller::TxPoller, storage::key_chain::{NullifierIndex, SharedAccountEntry}, }; @@ -99,12 +98,8 @@ pub struct WalletCore { storage: Storage, storage_path: PathBuf, - metrics_path: PathBuf, - metrics: HashMap, - /// Wrapping metric updates in Arc to not break interfaces too much. - /// - /// It is assumed, that wallet methods can be accesed via immutable reference. - metric_updates: Arc>>, + statistics_path: PathBuf, + statistics: HashMap, multi_sequencer_client: MultiSequencerClient, } @@ -114,15 +109,15 @@ impl WalletCore { pub async fn from_env() -> Result { let config_path = helperfunctions::fetch_config_path()?; let storage_path = helperfunctions::fetch_persistent_storage_path()?; - let metrics_path = helperfunctions::fetch_metrics_path()?; + let statistics_path = helperfunctions::fetch_statistics_path()?; - Self::new_update_chain(config_path, storage_path, metrics_path, None).await + Self::new_update_chain(config_path, storage_path, statistics_path, None).await } pub async fn new_update_chain( config_path: PathBuf, storage_path: PathBuf, - metrics_path: PathBuf, + statistics_path: PathBuf, config_overrides: Option, ) -> Result { let storage = Storage::from_path(&storage_path) @@ -131,7 +126,7 @@ impl WalletCore { Self::new( config_path, storage_path, - metrics_path, + statistics_path, config_overrides, storage, ) @@ -141,7 +136,7 @@ impl WalletCore { pub async fn new_init_storage( config_path: PathBuf, storage_path: PathBuf, - metrics_path: PathBuf, + statistics_path: PathBuf, config_overrides: Option, password: &str, ) -> Result<(Self, Mnemonic)> { @@ -149,7 +144,7 @@ impl WalletCore { let wallet = Self::new( config_path, storage_path, - metrics_path, + statistics_path, config_overrides, storage, ) @@ -161,7 +156,7 @@ impl WalletCore { async fn new( config_path: PathBuf, storage_path: PathBuf, - metrics_path: PathBuf, + statistics_path: PathBuf, config_overrides: Option, storage: Storage, ) -> Result { @@ -176,11 +171,11 @@ impl WalletCore { config.apply_overrides(config_overrides); } - let mut metrics = extract_metrics_from_path(&metrics_path)?; + let mut statistics = extract_statistics_from_path(&statistics_path)?; let multi_sequencer_client = MultiSequencerClient::new( &config.sequencers_conn_data, - &mut metrics, + &mut statistics, config.multi_sequencer_client_config.clone(), ) .await?; @@ -189,11 +184,10 @@ impl WalletCore { config_path, config_overrides, config, - storage_path, storage, - metrics_path, - metrics, - metric_updates: Arc::new(RwLock::new(vec![])), + storage_path, + statistics_path, + statistics, multi_sequencer_client, }) } @@ -215,12 +209,12 @@ impl WalletCore { #[must_use] pub fn leader_owned(&self) -> SequencerClient { - self.multi_sequencer_client.leader.clone() + self.multi_sequencer_client.leader().clone() } #[must_use] pub fn leader_url(&self) -> Url { - self.multi_sequencer_client.leader_url.clone() + self.multi_sequencer_client.leader_url().clone() } /// Get storage. @@ -259,39 +253,35 @@ impl WalletCore { Ok(()) } - async fn update_metrics(&mut self) -> Result<()> { - let leader_metric = self - .metrics - .get_mut(&self.multi_sequencer_client.leader_url) - .ok_or_else(|| anyhow::anyhow!("Leader URL is not present in metrics"))?; + /// Rotates multi-client and stores metrics. + pub async fn client_rotation(&mut self) -> Result<()> { + let leader_statistic = self + .statistics + .get_mut(self.multi_sequencer_client.leader_url()) + .ok_or_else(|| anyhow::anyhow!("Leader URL is not present in statistics"))?; - { - let metric_updates = self.metric_updates.read().await; - leader_metric.apply_updates(metric_updates.as_ref()); - } + self.multi_sequencer_client + .update_statistics(leader_statistic) + .await; - // Clear updates - { - let mut metric_updates = self.metric_updates.write().await; - metric_updates.clear(); - } + self.multi_sequencer_client + .rotate( + &self.config.sequencers, + &mut self.statistics, + self.config.calibration_limit, + ) + .await?; - Ok(()) - } - - pub async fn store_metrics(&mut self) -> Result<()> { - self.update_metrics().await?; - - let metrics_serialized = serde_json::to_vec_pretty(&self.metrics)?; - let mut file = tokio::fs::File::create(&self.metrics_path) + let statistics_serialized = serde_json::to_vec_pretty(&self.statistics)?; + let mut file = tokio::fs::File::create(&self.statistics_path) .await .context("Failed to create file")?; - file.write_all(&metrics_serialized) + file.write_all(&statistics_serialized) .await .context("Failed to write to file")?; file.sync_all().await.context("Failed to sync file")?; - println!("Stored metrics at {}", self.metrics_path.display()); + println!("Stored statistics at {}", self.statistics_path.display()); Ok(()) } @@ -488,37 +478,26 @@ impl WalletCore { } #[must_use] - pub fn get_metrics(&self, sequencer_url: &Url) -> Option<&Metrics> { - self.metrics.get(sequencer_url) + pub fn get_statistics(&self, sequencer_url: &Url) -> Option<&Statistics> { + self.statistics.get(sequencer_url) } /// Get account balance. pub async fn get_account_balance(&self, acc: AccountId) -> Result { - let call_f = async |client: &SequencerClient| client.get_account_balance(acc).await; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(call_res?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_account_balance(acc).await) + .await?) } /// Get accounts nonces. pub async fn get_accounts_nonces(&self, accs: &[AccountId]) -> Result> { - let call_f = - async |client: &SequencerClient| client.get_accounts_nonces(accs.to_vec()).await; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(call_res?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client.get_accounts_nonces(accs.to_vec()).await + }) + .await?) } pub async fn get_account(&self, account_id: AccountIdWithPrivacy) -> Result { @@ -535,59 +514,35 @@ impl WalletCore { } pub async fn get_last_block_id(&self) -> Result { - let call_f = async |client: &SequencerClient| client.get_last_block_id().await; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(call_res?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_last_block_id().await) + .await?) } pub async fn get_block(&self, block_id: u64) -> Result> { - let call_f = async |client: &SequencerClient| client.get_block(block_id).await; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(call_res?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_block(block_id).await) + .await?) } pub async fn get_transaction( &self, hash: HashType, ) -> Result> { - let call_f = async |client: &SequencerClient| client.get_transaction(hash).await; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(call_res?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_transaction(hash).await) + .await?) } /// Get public account. pub async fn get_account_public(&self, account_id: AccountId) -> Result { - let call_f = async |client: &SequencerClient| client.get_account(account_id).await; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(call_res?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_account(account_id).await) + .await?) } #[must_use] @@ -623,16 +578,10 @@ impl WalletCore { } pub async fn get_program_ids(&self) -> Result> { - let call_f = async |client: &SequencerClient| client.get_program_ids().await; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(call_res?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| client.get_program_ids().await) + .await?) } /// Poll transactions. @@ -647,18 +596,12 @@ impl WalletCore { &self, commitments: Vec, ) -> Result<(Vec>, CommitmentSetDigest)> { - let call_f = - async |client: &SequencerClient| client.get_proofs_and_root(commitments.clone()).await; - - let (proofs_and_root, metrics_update) = - self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(proofs_and_root?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client.get_proofs_and_root(commitments.clone()).await + }) + .await?) } pub fn decode_insert_privacy_preserving_transaction_results( @@ -797,18 +740,14 @@ impl WalletCore { .map(|keys| keys.ssk) .collect(); - let call_f = async |client: &SequencerClient| { - client - .send_transaction(LeeTransaction::PrivacyPreserving(tx.clone())) - .await - }; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } + let call_res = self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client + .send_transaction(LeeTransaction::PrivacyPreserving(tx.clone())) + .await + }) + .await; Ok((call_res?, shared_secrets)) } @@ -869,40 +808,28 @@ impl WalletCore { let tx = lee::public_transaction::PublicTransaction::new(message, witness_set); - let call_f = async |client: &SequencerClient| { - client - .send_transaction(LeeTransaction::Public(tx.clone())) - .await - }; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(call_res?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client + .send_transaction(LeeTransaction::Public(tx.clone())) + .await + }) + .await?) } pub async fn send_program_deployment_transaction(&self, bytecode: Vec) -> Result { let message = lee::program_deployment_transaction::Message::new(bytecode); let transaction = ProgramDeploymentTransaction::new(message); - let call_f = async |client: &SequencerClient| { - client - .send_transaction(LeeTransaction::ProgramDeployment(transaction.clone())) - .await - }; - - let (call_res, metrics_update) = self.multi_sequencer_client.metered_call(call_f).await; - - { - let mut metric_updates_guard = self.metric_updates.write().await; - metric_updates_guard.push(metrics_update); - } - - Ok(call_res?) + Ok(self + .multi_sequencer_client + .metered_call(async |client: &SequencerClient| { + client + .send_transaction(LeeTransaction::ProgramDeployment(transaction.clone())) + .await + }) + .await?) } pub async fn sync_to_latest_block(&mut self) -> Result { @@ -928,7 +855,7 @@ impl WalletCore { println!("Syncing to block {block_id}. Blocks to sync: {num_of_blocks}"); - let poller = self.optimal_poller().clone(); + let poller = self.optimal_poller(); let mut blocks = std::pin::pin!(poller.poll_block_range(last_synced_block.saturating_add(1)..=block_id)); diff --git a/lez/wallet/src/main.rs b/lez/wallet/src/main.rs index 8a66a642..c9cf0820 100644 --- a/lez/wallet/src/main.rs +++ b/lez/wallet/src/main.rs @@ -8,7 +8,7 @@ use clap::{CommandFactory as _, Parser as _}; use wallet::{ WalletCore, cli::{Args, execute_continuous_run, execute_subcommand, read_password_from_stdin}, - helperfunctions::{fetch_config_path, fetch_metrics_path, fetch_persistent_storage_path}, + helperfunctions::{fetch_config_path, fetch_persistent_storage_path, fetch_statistics_path}, }; // TODO #169: We have sample configs for sequencer, but not for wallet @@ -20,7 +20,6 @@ use wallet::{ async fn main() -> Result<()> { let Args { continuous_run, - auth: _auth, command, } = Args::parse(); @@ -29,11 +28,11 @@ async fn main() -> Result<()> { let config_path = fetch_config_path().context("Could not fetch config path")?; let storage_path = fetch_persistent_storage_path().context("Could not fetch persistent storage path")?; - let metrics_path = fetch_metrics_path().context("Could not fetch metrics path")?; + let statistics_path = fetch_statistics_path().context("Could not fetch statistics path")?; if let Some(command) = command { let mut wallet = if storage_path.exists() { - WalletCore::new_update_chain(config_path, storage_path, metrics_path, None).await? + WalletCore::new_update_chain(config_path, storage_path, statistics_path, None).await? } else { // TODO: Maybe move to `WalletCore::from_env()` or similar? @@ -43,7 +42,7 @@ async fn main() -> Result<()> { let (wallet, mnemonic) = WalletCore::new_init_storage( config_path, storage_path, - metrics_path, + statistics_path, None, &password, ) @@ -64,7 +63,7 @@ async fn main() -> Result<()> { Ok(()) } else if continuous_run { let mut wallet = - WalletCore::new_update_chain(config_path, storage_path, metrics_path, None).await?; + WalletCore::new_update_chain(config_path, storage_path, statistics_path, None).await?; execute_continuous_run(&mut wallet).await } else { let help = Args::command().render_long_help(); diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index 421a4194..71a6ce8c 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -7,33 +7,35 @@ reason = "Operated numbers is not big enough to have precision loss" )] -use std::{collections::HashMap, path::Path}; +use std::{collections::HashMap, path::Path, sync::Arc}; use anyhow::{Context as _, Result}; +use lee_core::BlockId; use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder}; use serde::{Deserialize, Serialize}; +use tokio::sync::RwLock; use url::Url; use crate::config::{MultiSequencerClientConfig, SequencerConnectionData}; #[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Metrics { +pub struct Statistics { pub latency_avg: f32, pub latency_var: f32, pub sample_size: usize, - pub latest_block_id: u64, + pub latest_block_id: BlockId, pub errors: u64, } #[derive(Debug, Clone)] -pub struct MetricsUpdate { +pub struct StatisticsUpdate { pub latency: f32, - pub new_latest_block_id: Option, + pub new_latest_block_id: Option, pub is_failed: bool, } -impl Metrics { - pub fn apply_updates(&mut self, updates: &[MetricsUpdate]) { +impl Statistics { + pub fn apply_updates(&mut self, updates: &[StatisticsUpdate]) { let CumulativeUpdates { failure_count, latest_block_id, @@ -77,12 +79,16 @@ pub struct MultiSequencerClient { /// Ordered list of leaders, from best to worst pub leader_list: Vec<(SequencerClient, Url)>, pub multi_sequencer_client_config: MultiSequencerClientConfig, + /// Wrapping statistic updates in Arc to not break interfaces too much. + /// + /// It is assumed, that wallet methods can be accesed via immutable reference. + statistic_updates: Arc>>, } impl MultiSequencerClient { - pub async fn new( + async fn setup( conn_data: &[SequencerConnectionData], - metrics: &mut HashMap, + statistics: &mut HashMap, multi_sequencer_client_config: MultiSequencerClientConfig, ) -> Result { let mut client_list = HashMap::new(); @@ -110,31 +116,30 @@ impl MultiSequencerClient { .context("Failed to create sequencer client")? }; - // If there is no metrics for client, callibrate it - if metrics.contains_key(sequencer_addr) { + // If there is statistics for client, actualize it + if let Some(metric_mut) = statistics.get_mut(sequencer_addr) { let metric_updates = actualize_client(&sequencer_client).await; log::debug!( "Metered call for {sequencer_addr:?}, metric updates is {metric_updates:?}" ); - let metric_mut = metrics.get_mut(sequencer_addr).unwrap(); metric_mut.apply_updates(&[metric_updates]); - // Otherwise actualize client data + // Otherwise calibrate client data + } else if let Some(client_statistics) = + calibrate_client(&sequencer_client, calibration_limit).await + { + statistics.insert(sequencer_addr.clone(), client_statistics); } else { - metrics.insert( - sequencer_addr.clone(), - callibrate_client( - &sequencer_client, - multi_sequencer_client_config.callibration_limit, - ) - .await, - ); + log::warn!("Client {sequencer_addr:?} failed all {calibration_limit} calibration attempts, it may be unhealthy. + \n Consider bumping calibration_limit or remove this client altogether"); } client_list.insert(sequencer_addr.clone(), sequencer_client); } + // Dropping client list, for reasons why, see comment in structure definition. + let leader_list = choose_leaders( &client_list, metrics, @@ -150,36 +155,80 @@ impl MultiSequencerClient { }) } - #[must_use] - pub fn leader_ref(&self) -> &(SequencerClient, Url) { - &self.leader_list.first().expect("Must be at least one leader") + pub async fn new( + conn_data: &[SequencerConnectionData], + statistics: &mut HashMap, + calibration_limit: usize, + ) -> Result { + let (leader_url, leader) = Self::setup(conn_data, statistics, calibration_limit).await?; + + Ok(Self { + leader, + leader_url, + statistic_updates: Arc::new(RwLock::new(vec![])), + }) + } + + /// Re-choose leader, `statistic_updates` must be empty. + pub async fn rotate( + &mut self, + conn_data: &[SequencerConnectionData], + statistics: &mut HashMap, + calibration_limit: usize, + ) -> Result<()> { + let (leader_url, leader) = Self::setup(conn_data, statistics, calibration_limit).await?; + + log::info!("Chosen leader is {leader_url:?}"); + + self.leader = leader; + self.leader_url = leader_url; + + Ok(()) } #[must_use] - pub fn leader_clone(&self) -> (SequencerClient, Url) { - self.leader_list.first().expect("Must be at least one leader").clone() + pub const fn leader(&self) -> &SequencerClient { + &self.leader + } + + #[must_use] + pub const fn leader_url(&self) -> &Url { + &self.leader_url } // Keeping this call abstract, in case if we need to do more than one request pub async fn metered_call Result>( &self, call: I, - ) -> (Result, MetricsUpdate) { - let resp = tokio::join!(call(&(*self.leader_ref()).0), actualize_client(&(*self.leader_ref()).0)); + ) -> Result { + let (resp, statistics_update) = + tokio::join!(call(self.leader()), actualize_client(self.leader())); log::debug!( "Metered call for {:?}, metric updates is {:?}", - self.leader_ref().1, - resp.1 + self.leader_url, + statistics_update ); + { + let mut statistic_updates_guard = self.statistic_updates.write().await; + statistic_updates_guard.push(statistics_update); + } + resp } + + /// Update statistics of a leader, clear statist updates log. + pub async fn update_statistics(&self, leader_statistic: &mut Statistics) { + let mut statistic_updates = self.statistic_updates.write().await; + leader_statistic.apply_updates(statistic_updates.as_ref()); + statistic_updates.clear(); + } } struct CumulativeUpdates { pub failure_count: u64, - pub latest_block_id: Option, + pub latest_block_id: Option, /// Necessary for cumulative average calculation. pub cumulative_latency: f32, /// Necessary for cumulative variance calculation. @@ -188,12 +237,12 @@ struct CumulativeUpdates { } impl CumulativeUpdates { - fn from_metric_updates(metric_updates: &[MetricsUpdate]) -> Self { + fn from_metric_updates(metric_updates: &[StatisticsUpdate]) -> Self { let (failure_count, latest_block_id, cumulative_latency, cumulative_latency_squares) = metric_updates .iter() .fold((0_u64, None, 0_f32, 0_f32), |acc, x| { - let MetricsUpdate { + let StatisticsUpdate { latency, new_latest_block_id, is_failed, @@ -230,34 +279,45 @@ impl CumulativeUpdates { } } -pub fn extract_metrics_from_path(path: &Path) -> Result, anyhow::Error> { +pub fn extract_statistics_from_path( + path: &Path, +) -> Result, anyhow::Error> { match std::fs::File::open(path) { Ok(file) => { let reader = std::io::BufReader::new(file); Ok(serde_json::from_reader(reader)?) } Err(err) if err.kind() == std::io::ErrorKind::NotFound => { - println!("Metrics not found, choosing empty"); + println!("Statistics not found, choosing empty"); Ok(HashMap::new()) } Err(err) => Err(err).context("IO error"), } } -pub async fn callibrate_client(client: &SequencerClient, callibration_limit: usize) -> Metrics { +/// Measuring `get_last_block_id` as it should be the fastest request on sequencer. +async fn measure_request_duration(client: &SequencerClient) -> (u128, Option) { + let now = tokio::time::Instant::now(); + let block_id = client.get_last_block_id().await.ok(); + ( + tokio::time::Instant::now().duration_since(now).as_millis(), + block_id, + ) +} + +pub async fn calibrate_client( + client: &SequencerClient, + calibration_limit: usize, +) -> Option { let mut latencies = vec![]; let mut latest_block_id = 0; let mut errors: u64 = 0; // ToDo: Add some DDoS adaptation - for _ in 0..callibration_limit { - let now = tokio::time::Instant::now(); + for _ in 0..calibration_limit { + let (latency, block_id) = measure_request_duration(client).await; - let block_id = client.get_last_block_id().await; - - let latency = tokio::time::Instant::now().duration_since(now).as_millis(); - - let Ok(block_id) = block_id else { + let Some(block_id) = block_id else { errors = errors.saturating_add(1); continue; }; @@ -266,7 +326,12 @@ pub async fn callibrate_client(client: &SequencerClient, callibration_limit: usi latencies.push(latency); } - // Precision loss if fine there + // There is no point in guard numbers, exclude client if it fails all requests. + if latencies.is_empty() { + return None; + } + + // Precision loss is fine there let sample_size = latencies.len(); #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] let latency_avg = (latencies.iter().sum::() as f32) / (sample_size as f32); @@ -275,24 +340,22 @@ pub async fn callibrate_client(client: &SequencerClient, callibration_limit: usi ((*x as f32) - latency_avg).mul_add((*x as f32) - latency_avg, acc) }) / (sample_size as f32); - Metrics { + Some(Statistics { latency_avg, latency_var, sample_size, latest_block_id, errors, - } + }) } -pub async fn actualize_client(client: &SequencerClient) -> MetricsUpdate { - let now = tokio::time::Instant::now(); - - let block_id = client.get_last_block_id().await.ok(); +pub async fn actualize_client(client: &SequencerClient) -> StatisticsUpdate { + let (latency, block_id) = measure_request_duration(client).await; #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] - let latency = tokio::time::Instant::now().duration_since(now).as_millis() as f32; + let latency = latency as f32; - MetricsUpdate { + StatisticsUpdate { latency, new_latest_block_id: block_id, is_failed: block_id.is_none(), @@ -302,15 +365,13 @@ pub async fn actualize_client(client: &SequencerClient) -> MetricsUpdate { #[must_use] pub fn choose_leaders( client_list: &HashMap, - metrics: &HashMap, + statistics: &HashMap, distribution_limit: usize, ) -> Option> { - let mut client_vec = vec![]; - // Sort out all unmetered clients - client_vec = client_list + let mut client_vec: Vec<_> = client_list .keys() - .filter(|item| metrics.contains_key(*item)) + .filter(|item| statistics.contains_key(*item)) .collect(); if client_vec.is_empty() { @@ -319,8 +380,8 @@ pub fn choose_leaders( // Considering the nature of our requests, the latest_block_id is the dominant characteristic let max_block_id_addr = client_vec.iter().fold(client_vec[0], |acc, x| { - let old_latest_block_id = metrics.get(acc).unwrap().latest_block_id; - let new_latest_block_id = metrics.get(*x).unwrap().latest_block_id; + let old_latest_block_id = statistics.get(acc).unwrap().latest_block_id; + let new_latest_block_id = statistics.get(*x).unwrap().latest_block_id; if new_latest_block_id > old_latest_block_id { *x } else { @@ -328,45 +389,58 @@ pub fn choose_leaders( } }); - let max_block_id = metrics.get(max_block_id_addr).unwrap().latest_block_id; + let max_block_id = statistics.get(max_block_id_addr).unwrap().latest_block_id; // Sort out all clients running late client_vec = client_vec .iter() .filter_map(|x| { - let latest_block_id = metrics.get(*x).unwrap().latest_block_id; + let latest_block_id = statistics.get(*x).unwrap().latest_block_id; (latest_block_id == max_block_id).then_some(*x) }) .collect(); - // Get the clients with lesser or equal to average error count + // Get the clients with lesser or equal to average error ratio #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] - let avg_err_count = (client_vec.iter().fold(0_u64, |acc, x| { - acc.saturating_add(metrics.get(*x).unwrap().errors) - }) as f32) - / (client_vec.len() as f32); + let avg_err_ratio = client_vec.iter().fold(0_f32, |acc, x| { + acc + error_ratio( + statistics.get(*x).unwrap().errors, + statistics.get(*x).unwrap().sample_size, + ) + }) / (client_vec.len() as f32); client_vec.sort_by(|a, b| { - metrics - .get(*a) - .unwrap() - .errors - .cmp(&metrics.get(*b).unwrap().errors) + let err_ratio_a = error_ratio( + statistics.get(*a).unwrap().errors, + statistics.get(*a).unwrap().sample_size, + ); + let err_ratio_b = error_ratio( + statistics.get(*b).unwrap().errors, + statistics.get(*b).unwrap().sample_size, + ); + + err_ratio_a + .partial_cmp(&err_ratio_b) + .expect("Ratios must be a valid numbers") }); - #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] let mut client_vec = client_vec[..(client_vec .iter() - .position(|item| (metrics.get(*item).unwrap().errors as f32) > avg_err_count) + .position(|item| { + error_ratio( + statistics.get(*item).unwrap().errors, + statistics.get(*item).unwrap().sample_size, + ) > avg_err_ratio + }) .unwrap_or(client_vec.len()))] .to_vec(); // Choose clients with least latency and variance client_vec.sort_by(|a, b| { - let left = metrics.get(*a).unwrap(); + let left = statistics.get(*a).unwrap(); let (left_lat, left_var) = (left.latency_avg, left.latency_var); - let right = metrics.get(*b).unwrap(); + let right = statistics.get(*b).unwrap(); let (right_lat, right_var) = (right.latency_avg, right.latency_var); let right_std = right_var.sqrt(); @@ -467,6 +541,15 @@ fn cumulative_var( / (orig_size_f + mod_size_f) } +fn error_ratio(errors: u64, size: usize) -> f32 { + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let errors_f = errors as f32; + #[expect(clippy::as_conversions, reason = "int to float conversion is safe")] + let size_f = size as f32; + + errors_f / (errors_f + size_f) +} + #[cfg(test)] mod tests { use std::collections::HashMap; @@ -475,17 +558,18 @@ mod tests { use url::Url; use crate::multi_client::{ - CumulativeUpdates, Metrics, MetricsUpdate, choose_leaders, cumulative_avg, cumulative_var, + CumulativeUpdates, Statistics, StatisticsUpdate, choose_leaders, cumulative_avg, + cumulative_var, }; - fn update_metrics( - metrics: &mut HashMap, + fn update_statistics( + statistics: &mut HashMap, leader_url: &Url, - metric_updates: &[MetricsUpdate], + metric_updates: &[StatisticsUpdate], ) -> Result<(), anyhow::Error> { - let leader_metric = metrics + let leader_metric = statistics .get_mut(leader_url) - .ok_or_else(|| anyhow::anyhow!("Leader URL is not present in metrics"))?; + .ok_or_else(|| anyhow::anyhow!("Leader URL is not present in statistics"))?; leader_metric.apply_updates(metric_updates); @@ -499,18 +583,18 @@ mod tests { #[test] fn cumulative_updates_test() { - let metrics_updates_vec = vec![ - MetricsUpdate { + let statistics_updates_vec = vec![ + StatisticsUpdate { latency: 100_f32, new_latest_block_id: Some(15), is_failed: false, }, - MetricsUpdate { + StatisticsUpdate { latency: 115_f32, new_latest_block_id: Some(16), is_failed: false, }, - MetricsUpdate { + StatisticsUpdate { latency: 50_f32, new_latest_block_id: None, is_failed: true, @@ -523,7 +607,7 @@ mod tests { cumulative_latency, cumulative_latency_squares, additional_sample_size, - } = CumulativeUpdates::from_metric_updates(&metrics_updates_vec); + } = CumulativeUpdates::from_metric_updates(&statistics_updates_vec); let epsilon = 0.01_f32; @@ -620,18 +704,18 @@ mod tests { #[test] fn metric_updates_correctness() { - let metrics_updates_vec = vec![ - MetricsUpdate { + let statistics_updates_vec = vec![ + StatisticsUpdate { latency: 100_f32, new_latest_block_id: Some(105), is_failed: false, }, - MetricsUpdate { + StatisticsUpdate { latency: 115_f32, new_latest_block_id: Some(106), is_failed: false, }, - MetricsUpdate { + StatisticsUpdate { latency: 50_f32, new_latest_block_id: None, is_failed: true, @@ -640,7 +724,7 @@ mod tests { let addr_leader = Url::parse("https://127.0.0.1:3040").unwrap(); - let leader_metrics = Metrics { + let leader_statistics = Statistics { latency_avg: 100_f32, latency_var: 25_f32, sample_size: 10, @@ -652,15 +736,15 @@ mod tests { let cumulative_latency_squares = 100_f32.mul_add(100_f32, 115_f32 * 115_f32); let avg_manual = cumulative_avg( - leader_metrics.latency_avg, + leader_statistics.latency_avg, cumulative_latency, 10_f32, 2_f32, ); let var_manual = cumulative_var( - leader_metrics.latency_avg, + leader_statistics.latency_avg, avg_manual, - leader_metrics.latency_var, + leader_statistics.latency_var, cumulative_latency, cumulative_latency_squares, 10_f32, @@ -668,11 +752,11 @@ mod tests { ); let mut metric_map = HashMap::new(); - metric_map.insert(addr_leader.clone(), leader_metrics); + metric_map.insert(addr_leader.clone(), leader_statistics); - update_metrics(&mut metric_map, &addr_leader, &metrics_updates_vec).unwrap(); + update_statistics(&mut metric_map, &addr_leader, &statistics_updates_vec).unwrap(); - let Metrics { + let Statistics { latency_avg, latency_var, sample_size, @@ -708,11 +792,11 @@ mod tests { client_list.insert(addr_2.clone(), client_2); client_list.insert(addr_3.clone(), client_3); - let mut metrics = HashMap::new(); + let mut statistics = HashMap::new(); - metrics.insert( + statistics.insert( addr_3, - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -721,9 +805,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_2, - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -732,9 +816,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_1, - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -743,9 +827,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_leader.clone(), - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -754,7 +838,7 @@ mod tests { }, ); - let (leader_url, _) = choose_leader(&client_list, &metrics).unwrap(); + let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); assert_eq!(leader_url, addr_leader); } @@ -778,11 +862,11 @@ mod tests { client_list.insert(addr_2.clone(), client_2); client_list.insert(addr_3.clone(), client_3); - let mut metrics = HashMap::new(); + let mut statistics = HashMap::new(); - metrics.insert( + statistics.insert( addr_3, - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -791,9 +875,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_2, - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -802,9 +886,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_1, - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -813,9 +897,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_leader.clone(), - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -824,7 +908,7 @@ mod tests { }, ); - let (leader_url, _) = choose_leader(&client_list, &metrics).unwrap(); + let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); assert_eq!(leader_url, addr_leader); } @@ -848,11 +932,11 @@ mod tests { client_list.insert(addr_2.clone(), client_2); client_list.insert(addr_3.clone(), client_3); - let mut metrics = HashMap::new(); + let mut statistics = HashMap::new(); - metrics.insert( + statistics.insert( addr_3, - Metrics { + Statistics { latency_avg: 103_f32, latency_var: 10_f32, sample_size: 10, @@ -861,9 +945,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_2, - Metrics { + Statistics { latency_avg: 102_f32, latency_var: 10_f32, sample_size: 10, @@ -872,9 +956,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_1, - Metrics { + Statistics { latency_avg: 101_f32, latency_var: 10_f32, sample_size: 10, @@ -883,9 +967,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_leader.clone(), - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -894,7 +978,7 @@ mod tests { }, ); - let (leader_url, _) = choose_leader(&client_list, &metrics).unwrap(); + let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); assert_eq!(leader_url, addr_leader); } @@ -918,11 +1002,11 @@ mod tests { client_list.insert(addr_2.clone(), client_2); client_list.insert(addr_3.clone(), client_3); - let mut metrics = HashMap::new(); + let mut statistics = HashMap::new(); - metrics.insert( + statistics.insert( addr_3, - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 13_f32, sample_size: 10, @@ -931,9 +1015,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_2, - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 12_f32, sample_size: 10, @@ -942,9 +1026,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_1, - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 11_f32, sample_size: 10, @@ -953,9 +1037,9 @@ mod tests { }, ); - metrics.insert( + statistics.insert( addr_leader.clone(), - Metrics { + Statistics { latency_avg: 100_f32, latency_var: 10_f32, sample_size: 10, @@ -964,7 +1048,7 @@ mod tests { }, ); - let (leader_url, _) = choose_leader(&client_list, &metrics).unwrap(); + let (leader_url, _) = choose_leader(&client_list, &statistics).unwrap(); assert_eq!(leader_url, addr_leader); } diff --git a/test_fixtures/src/config.rs b/test_fixtures/src/config.rs index cacca77d..75310000 100644 --- a/test_fixtures/src/config.rs +++ b/test_fixtures/src/config.rs @@ -194,7 +194,7 @@ pub fn genesis_from_accounts( pub fn wallet_config(sequencer_addr: SocketAddr) -> Result { Ok(WalletConfig { - sequencers_conn_data: vec![SequencerConnectionData { + sequencers: vec![SequencerConnectionData { sequencer_addr: addr_to_url(UrlProtocol::Http, sequencer_addr) .context("Failed to convert sequencer addr to URL")?, basic_auth: None, @@ -205,7 +205,7 @@ pub fn wallet_config(sequencer_addr: SocketAddr) -> Result { seq_block_poll_max_amount: 100, multi_sequencer_client_config: MultiSequencerClientConfig { distribution_limit: 1, - callibration_limit: 1, + callibration_limit: 5, }, }) }