.cargo/config

@@ -0,0 +1,2 @@
+[build]
+# rustflags = ["--cfg", "tokio_unstable"]

.cargo/config.toml

@@ -1,2 +0,0 @@
-[build]
-rustflags = ["--cfg", "tokio_unstable"]

.drone.yml

@@ -0,0 +1,421 @@
+kind: pipeline
+type: docker
+name: clippy
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: clippy
+  image: asonix/rust-builder:latest-linux-amd64
+  pull: always
+  commands:
+  - rustup component add clippy
+  - cargo clippy --no-deps -- -D warnings
+
+trigger:
+  event:
+  - push
+  - pull_request
+  - tag
+
+---
+
+kind: pipeline
+type: docker
+name: tests
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: tests
+  image: asonix/rust-builder:latest-linux-amd64
+  pull: always
+  commands:
+  - cargo test
+
+trigger:
+  event:
+  - push
+  - pull_request
+  - tag
+
+---
+
+kind: pipeline
+type: docker
+name: check-amd64
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: check
+  image: asonix/rust-builder:latest-linux-amd64
+  pull: always
+  commands:
+  - cargo check --target=$TARGET
+
+trigger:
+  event:
+  - push
+  - pull_request
+
+---
+
+kind: pipeline
+type: docker
+name: build-amd64
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: build
+  image: asonix/rust-builder:latest-linux-amd64
+  pull: always
+  commands:
+  - cargo build --target=$TARGET --release
+  - $TOOL-strip target/$TARGET/release/relay
+  - cp target/$TARGET/release/relay .
+  - cp relay relay-linux-amd64
+
+- name: push
+  image: plugins/docker:20
+  settings:
+    username: asonix
+    password:
+      from_secret: dockerhub_token
+    repo: asonix/relay
+    dockerfile: docker/drone/Dockerfile
+    auto_tag: true
+    auto_tag_suffix: linux-amd64
+    build_args:
+    - REPO_ARCH=amd64
+
+- name: publish
+  image: plugins/gitea-release:1
+  settings:
+    api_key:
+      from_secret: gitea_token
+    base_url: https://git.asonix.dog
+    files:
+    - relay-linux-amd64
+
+depends_on:
+- clippy
+- tests
+
+trigger:
+  event:
+  - tag
+
+---
+
+kind: pipeline
+type: docker
+name: check-arm64v8
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: check
+  image: asonix/rust-builder:latest-linux-arm64v8
+  pull: always
+  commands:
+  - cargo check --target=$TARGET
+
+trigger:
+  event:
+  - push
+  - pull_request
+
+---
+
+kind: pipeline
+type: docker
+name: build-arm64v8
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: build
+  image: asonix/rust-builder:latest-linux-arm64v8
+  pull: always
+  commands:
+  - cargo build --target=$TARGET --release
+  - $TOOL-strip target/$TARGET/release/relay
+  - cp target/$TARGET/release/relay .
+  - cp relay relay-linux-arm64v8
+
+- name: push
+  image: plugins/docker:20
+  settings:
+    username: asonix
+    password:
+      from_secret: dockerhub_token
+    repo: asonix/relay
+    dockerfile: docker/drone/Dockerfile
+    auto_tag: true
+    auto_tag_suffix: linux-arm64v8
+    build_args:
+    - REPO_ARCH=arm64v8
+
+- name: publish
+  image: plugins/gitea-release:1
+  settings:
+    api_key:
+      from_secret: gitea_token
+    base_url: https://git.asonix.dog
+    files:
+    - relay-linux-arm64v8
+
+depends_on:
+- clippy
+- tests
+
+trigger:
+  event:
+  - tag
+
+---
+
+kind: pipeline
+type: docker
+name: check-arm32v7
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: check
+  image: asonix/rust-builder:latest-linux-arm32v7
+  pull: always
+  commands:
+  - cargo check --target=$TARGET
+
+trigger:
+  event:
+  - push
+  - pull_request
+
+---
+
+kind: pipeline
+type: docker
+name: build-arm32v7
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: build
+  image: asonix/rust-builder:latest-linux-arm32v7
+  pull: always
+  commands:
+  - cargo build --target=$TARGET --release
+  - $TOOL-strip target/$TARGET/release/relay
+  - cp target/$TARGET/release/relay .
+  - cp relay relay-linux-arm32v7
+
+- name: push
+  image: plugins/docker:20
+  settings:
+    username: asonix
+    password:
+      from_secret: dockerhub_token
+    repo: asonix/relay
+    dockerfile: docker/drone/Dockerfile
+    auto_tag: true
+    auto_tag_suffix: linux-arm32v7
+    build_args:
+    - REPO_ARCH=arm32v7
+
+- name: publish
+  image: plugins/gitea-release:1
+  settings:
+    api_key:
+      from_secret: gitea_token
+    base_url: https://git.asonix.dog
+    files:
+    - relay-linux-arm32v7
+
+depends_on:
+- clippy
+- tests
+
+trigger:
+  event:
+  - tag
+
+---
+
+kind: pipeline
+type: docker
+name: manifest
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: manifest
+  image: plugins/manifest:1
+  settings:
+    username: asonix
+    password:
+      from_secret: dockerhub_token
+    dump: true
+    auto_tag: true
+    ignore_missing: true
+    spec: docker/drone/manifest.tmpl
+
+
+depends_on:
+- build-amd64
+- build-arm64v8
+- build-arm32v7
+
+trigger:
+  event:
+  - tag
+
+---
+
+kind: pipeline
+type: docker
+name: publish-crate
+
+platform:
+  arch: amd64
+
+clone:
+  disable: true
+
+steps:
+- name: clone
+  image: alpine/git:latest
+  user: root
+  commands:
+  - git clone $DRONE_GIT_HTTP_URL .
+  - git checkout $DRONE_COMMIT
+  - chown -R 991:991 .
+
+- name: publish
+  image: asonix/rust-builder:latest-linux-amd64
+  pull: always
+  environment:
+    CRATES_IO_TOKEN:
+      from_secret: crates_io_token
+  commands:
+  - cargo publish --token $CRATES_IO_TOKEN
+
+depends_on:
+- build-amd64
+- build-arm64v8
+- build-arm32v7
+
+trigger:
+  event:
+  - tag

.forgejo/workflows/check.yaml

@@ -1,61 +0,0 @@
-on:
-  push:
-    branches:
-      - '*'
-  pull_request:
-    branches:
-      - main
-
-jobs:
-  clippy:
-    runs-on: docker
-    container:
-      image: docker.io/asonix/actions-base-image:0.1
-    steps:
-      -
-        name: Checkout relay
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Clippy
-        run: |
-          cargo clippy --no-default-features -- -D warnings
-
-  tests:
-    runs-on: docker
-    container:
-      image: docker.io/asonix/actions-base-image:0.1
-    steps:
-      -
-        name: Checkout relay
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Test
-        run: cargo test
-
-  check:
-    strategy:
-      fail-fast: false
-      matrix:
-        target:
-          - x86_64-unknown-linux-musl
-          - armv7-unknown-linux-musleabihf
-          - aarch64-unknown-linux-musl
-    runs-on: docker
-    container:
-      image: docker.io/asonix/actions-base-image:0.1
-    steps:
-      -
-        name: Checkout relay
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Debug builds
-        run: cargo zigbuild --target ${{ matrix.target }}

.forgejo/workflows/publish.yaml

@@ -1,226 +0,0 @@
-on:
-  push:
-    tags:
-      - 'v*.*.*'
-
-env:
-  REGISTRY_IMAGE: asonix/relay
-
-jobs:
-  clippy:
-    runs-on: base-image
-    container:
-      image: docker.io/asonix/actions-base-image:0.1
-    steps:
-      -
-        name: Checkout relay
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Clippy
-        run: |
-          # cargo clippy --no-default-features -- -D warnings
-          cargo clippy --no-default-features
-
-  tests:
-    runs-on: docker
-    container:
-      image: docker.io/asonix/actions-base-image:0.1
-    steps:
-      -
-        name: Checkout relay
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Test
-        run: cargo test
-
-  build:
-    needs:
-      - clippy
-      - tests
-    runs-on: docker
-    container:
-      image: docker.io/asonix/actions-base-image:0.1
-    strategy:
-      fail-fast: false
-      matrix:
-        info:
-          - target: x86_64-unknown-linux-musl
-            artifact: linux-amd64
-            platform: linux/amd64
-          - target: armv7-unknown-linux-musleabihf
-            artifact: linux-arm32v7
-            platform: linux/arm/v7
-          - target: aarch64-unknown-linux-musl
-            artifact: linux-arm64v8
-            platform: linux/arm64
-    steps:
-      -
-        name: Checkout relay
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Prepare Platform
-        run: |
-          platform=${{ matrix.info.platform }}
-          echo "PLATFORM_PAIR=${platform//\//-}" >> $GITHUB_ENV
-        shell: bash
-      -
-        name: Docker meta
-        id: meta
-        uses: https://github.com/docker/metadata-action@v5
-        with:
-          images: ${{ env.REGISTRY_IMAGE }}
-          flavor: |
-            latest=auto
-            suffix=-${{ matrix.info.artifact }}
-          tags: |
-            type=raw,value=latest,enable={{ is_default_branch }}
-            type=semver,pattern={{version}}
-            type=semver,pattern={{major}}.{{minor}}
-            type=semver,pattern={{major}}
-      -
-        name: Set up QEMU
-        uses: https://github.com/docker/setup-qemu-action@v3
-      -
-        name: Set up Docker Buildx
-        uses: https://github.com/docker/setup-buildx-action@v3
-      -
-        name: Docker login
-        uses: https://github.com/docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
-      -
-        name: Compile relay
-        run: cargo zigbuild --target ${{ matrix.info.target }} --release
-      -
-        name: Prepare artifacts
-        run: |
-          mkdir artifacts
-          cp target/${{ matrix.info.target }}/release/relay artifacts/relay-${{ matrix.info.artifact }}
-      -
-        uses: https://github.com/actions/upload-artifact@v3
-        with:
-          name: binaries
-          path: artifacts/
-      -
-        name: Prepare binary
-        run: |
-          cp target/${{ matrix.info.target }}/release/relay docker/forgejo/relay
-      -
-        name: Build and push ${{ matrix.info.platform }} docker image
-        id: build
-        uses: docker/build-push-action@v5
-        with:
-          context: ./docker/forgejo
-          platforms: ${{ matrix.info.platform }}
-          tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
-          outputs: type=image,name=${{ env.REGISTRY_IMAGE }},name-canonical=true,push=true
-      -
-        name: Export digest
-        run: |
-          mkdir -p /tmp/digests
-          digest="${{ steps.build.outputs.digest }}"
-          touch "/tmp/digests/${digest#sha256:}"
-          echo "Created /tmp/digests/${digest#sha256:}"
-        shell: bash
-      -
-        name: Upload ${{ matrix.info.platform }} digest
-        uses: https://github.com/actions/upload-artifact@v3
-        with:
-          name: digests
-          path: /tmp/digests/*
-          if-no-files-found: error
-          retention-days: 1
-
-  publish-docker:
-    runs-on: docker
-    container:
-      image: docker.io/asonix/actions-base-image:0.1
-    needs: [build]
-    steps:
-      -
-        name: Download digests
-        uses: https://github.com/actions/download-artifact@v3
-        with:
-          name: digests
-          path: /tmp/digests
-          pattern: digests-*
-          merge-multiple: true
-      -
-        name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
-      -
-        name: Docker login
-        uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
-      -
-        name: Docker meta
-        id: meta
-        uses: https://github.com/docker/metadata-action@v5
-        with:
-          images: ${{ env.REGISTRY_IMAGE }}
-          flavor: |
-            latest=auto
-          tags: |
-            type=raw,value=latest,enable={{ is_default_branch }}
-            type=semver,pattern={{version}}
-            type=semver,pattern={{major}}.{{minor}}
-            type=semver,pattern={{major}}
-      -
-        name: Create manifest list and push
-        working-directory: /tmp/digests
-        run: |
-          tags=$(jq -cr '.tags | map("-t " + .) | join(" ")' <<< "${DOCKER_METADATA_OUTPUT_JSON}")
-          images=$(printf "${{ env.REGISTRY_IMAGE }}@sha256:%s " *)
-          echo "Running 'docker buildx imagetools create ${tags[@]} ${images[@]}'"
-          docker buildx imagetools create ${tags[@]} ${images[@]}
-        shell: bash
-      -
-        name: Inspect Image
-        run: |
-          docker buildx imagetools inspect ${{ env.REGISTRY_IMAGE }}:${{ steps.meta.outputs.version }}
-
-  publish-forgejo:
-    needs: [build]
-    runs-on: docker
-    container:
-      image: docker.io/asonix/actions-base-image:0.1
-    steps:
-      - uses: https://github.com/actions/download-artifact@v3
-        with:
-          name: binaries
-          path: artifacts/
-          merge-multiple: true
-      - uses: actions/forgejo-release@v1
-        with:
-          direction: upload
-          token: ${{ secrets.GITHUB_TOKEN }}
-          release-dir: artifacts/
-
-  publish-crate:
-    needs: [build]
-    runs-on: docker
-    container:
-      image: docker.io/asonix/actions-base-image:0.1
-    steps:
-      -
-        name: Checkout relay
-        uses: https://github.com/actions/checkout@v4
-      -
-        name: Cargo Cache
-        uses: https://git.asonix.dog/asonix/actions/cache-rust-dependencies@main
-      -
-        name: Publish Crate
-        run: cargo publish --token ${{ secrets.CRATES_IO_TOKEN }}

Cargo.toml

@@ -1,11 +1,11 @@
 [package]
 name = "ap-relay"
 description = "A simple activitypub relay"
-version = "0.3.116"
+version = "0.3.88"
 authors = ["asonix <asonix@asonix.dog>"]
 license = "AGPL-3.0"
 readme = "README.md"
-repository = "https://git.asonix.dog/asonix/relay"
+repository = "https://git.asonix.dog/asonix/ap-relay"
 keywords = ["activitypub", "relay"]
 edition = "2021"
 build = "src/build.rs"
@@ -14,101 +14,92 @@ build = "src/build.rs"
 name = "relay"
 path = "src/main.rs"
 
-[profile.release]
-strip = true
-
 [features]
-console = ["dep:console-subscriber"]
+console = ["console-subscriber"]
 default = []
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-actix-web = { version = "4.4.0", default-features = false, features = ["compress-brotli", "compress-gzip", "rustls-0_23"] }
+anyhow = "1.0"
-actix-webfinger = { version = "0.5.0", default-features = false }
+actix-rt = "2.7.0"
+actix-web = { version = "4.0.1", default-features = false, features = [
+  "rustls",
+  "compress-brotli",
+  "compress-gzip",
+] }
+actix-webfinger = "0.4.0"
 activitystreams = "0.7.0-alpha.25"
 activitystreams-ext = "0.1.0-alpha.3"
-ammonia = "4.0.0"
+ammonia = "3.1.0"
-async-cpupool = "0.3.0"
+awc = { version = "3.0.0", default-features = false, features = ["rustls"] }
-bcrypt = "0.16"
+bcrypt = "0.15"
-base64 = "0.22"
+base64 = "0.21"
 clap = { version = "4.0.0", features = ["derive"] }
-color-eyre = "0.6.2"
+config = "0.13.0"
-config = { version = "0.14.0", default-features = false, features = ["toml", "json", "yaml"] }
+console-subscriber = { version = "0.1", optional = true }
-console-subscriber = { version = "0.4", optional = true }
+dashmap = "5.1.0"
-dashmap = "6.0.1"
 dotenv = "0.15.0"
-futures-core = "0.3.30"
+futures-util = "0.3.17"
-lru = "0.12.0"
+lru = "0.11.0"
-metrics = "0.23.0"
+metrics = "0.21.0"
-metrics-exporter-prometheus = { version = "0.15.0", default-features = false, features = [
+metrics-exporter-prometheus = { version = "0.12.0", default-features = false, features = [
   "http-listener",
 ] }
-metrics-util = "0.17.0"
+metrics-util = "0.15.0"
 mime = "0.3.16"
-minify-html = "0.15.0"
+minify-html = "0.11.0"
-opentelemetry = "0.27.1"
+opentelemetry = { version = "0.19", features = ["rt-tokio"] }
-opentelemetry_sdk = { version = "0.27", features = ["rt-tokio"] }
+opentelemetry-otlp = "0.12"
-opentelemetry-otlp = { version = "0.27", features = ["grpc-tonic"] }
 pin-project-lite = "0.2.9"
-# pinned to metrics-util
+quanta = "0.11.0"
-quanta = "0.12.0"
 rand = "0.8"
-reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream"]}
+rsa = { version = "0.9", features = ["sha2"] }
-reqwest-middleware = { version = "0.4", default-features = false, features = ["json"] }
-reqwest-tracing = "0.5.0"
-ring = "0.17.5"
-rsa = "0.9"
 rsa-magic-public-key = "0.8.0"
-rustls = { version = "0.23.0", default-features = false, features = ["ring", "logging", "std", "tls12"] }
+rustls = "0.20.7"
-rustls-channel-resolver = "0.3.0"
+rustls-pemfile = "1.0.1"
-rustls-pemfile = "2"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 sled = "0.34.7"
-streem = "0.2.0"
+teloxide = { version = "0.12.0", default-features = false, features = [
-teloxide = { version = "0.13.0", default-features = false, features = [
   "ctrlc_handler",
   "macros",
   "rustls",
 ] }
-thiserror = "2.0"
+thiserror = "1.0"
 time = { version = "0.3.17", features = ["serde"] }
 tracing = "0.1"
+tracing-awc = "0.1.7"
 tracing-error = "0.2"
-tracing-log = "0.2"
+tracing-futures = "0.2"
-tracing-opentelemetry = "0.28"
+tracing-log = "0.1"
+tracing-opentelemetry = "0.19"
 tracing-subscriber = { version = "0.3", features = [
   "ansi",
   "env-filter",
   "fmt",
 ] }
-tokio = { version = "1", features = ["full", "tracing"] }
+tokio = { version = "1", features = ["macros", "sync"] }
 uuid = { version = "1", features = ["v4", "serde"] }
 
 [dependencies.background-jobs]
-version = "0.19.0"
+version = "0.15.0"
 default-features = false
-features = ["error-logging", "metrics", "tokio"]
+features = ["background-jobs-actix", "error-logging"]
 
 [dependencies.http-signature-normalization-actix]
-version = "0.11.1"
+version = "0.8.0"
 default-features = false
-features = ["server", "ring"]
+features = ["client", "server", "sha-2"]
-
-[dependencies.http-signature-normalization-reqwest]
-version = "0.13.0"
-default-features = false
-features = ["middleware", "ring"]
 
 [dependencies.tracing-actix-web]
-version = "0.7.9"
+version = "0.7.5"
 
 [build-dependencies]
-color-eyre = "0.6.2"
+anyhow = "1.0"
 dotenv = "0.15.0"
 ructe = { version = "0.17.0", features = ["sass", "mime03"] }
-toml = "0.8.0"
+toml = "0.7.0"
 
 [profile.dev.package.rsa]
 opt-level = 3

README.md

@@ -106,8 +106,8 @@ LOCAL_BLURB="<p>Welcome to my cool relay where I have cool relay things happenin
 PROMETHEUS_ADDR=0.0.0.0
 PROMETHEUS_PORT=9000
 CLIENT_TIMEOUT=10
+CLIENT_POOL_SIZE=20
 DELIVER_CONCURRENCY=8
-SIGNATURE_THREADS=2
 ```
 
 #### Descriptions
@@ -160,19 +160,14 @@ Optional - Port to bind to for serving the prometheus scrape endpoint
 ##### `CLIENT_TIMEOUT`
 Optional - How long the relay will hold open a connection (in seconds) to a remote server during
 fetches and deliveries. This defaults to 10
+##### `CLIENT_POOL_SIZE`
+Optional - How many connections the relay should maintain per thread. This value will be multiplied
+by the number of cores available to the relay. This defaults to 20, so a 4-core machine will have a
+maximum of 160 simultaneous outbound connections. If you run into problems related to "Too many open
+files", you can either decrease this number or increase the ulimit for your system.
 ##### `DELIVER_CONCURRENCY`
 Optional - How many deliver requests the relay should allow to be in-flight per thread. the default
 is 8
-##### `SIGNATURE_THREADS`
-Optional - Override number of threads used for signing and verifying requests. Default is
-`std::thread::available_parallelism()` (It tries to detect how many cores you have). If it cannot
-detect the correct number of cores, it falls back to 1.
-##### 'PROXY_URL'
-Optional - URL of an HTTP proxy to forward outbound requests through
-##### 'PROXY_USERNAME'
-Optional - username to provide to the HTTP proxy set with `PROXY_URL` through HTTP Basic Auth
-##### 'PROXY_PASSWORD'
-Optional - password to provide to the HTTP proxy set with `PROXY_URL` through HTTP Basic Auth
 
 ### Subscribing
 Mastodon admins can subscribe to this relay by adding the `/inbox` route to their relay settings.

docker/drone/Dockerfile

@@ -0,0 +1,11 @@
+ARG REPO_ARCH
+
+FROM asonix/rust-runner:latest-linux-$REPO_ARCH
+
+COPY relay /usr/local/bin/relay
+
+USER app
+EXPOSE 8080
+VOLUME /mnt
+ENTRYPOINT ["/sbin/tini", "--"]
+CMD ["/usr/local/bin/relay"]

docker/drone/manifest.tmpl

@@ -0,0 +1,25 @@
+image: asonix/relay:{{#if build.tag}}{{trimPrefix "v" build.tag}}{{else}}latest{{/if}}
+{{#if build.tags}}
+tags:
+{{#each build.tags}}
+  - {{this}}
+{{/each}}
+{{/if}}
+manifests:
+  -
+    image: asonix/relay:{{#if build.tag}}{{trimPrefix "v" build.tag}}-{{/if}}linux-amd64
+    platform:
+      architecture: amd64
+      os: linux
+  -
+    image: asonix/relay:{{#if build.tag}}{{trimPrefix "v" build.tag}}-{{/if}}linux-arm64v8
+    platform:
+      architecture: arm64
+      os: linux
+      variant: v8
+  -
+    image: asonix/relay:{{#if build.tag}}{{trimPrefix "v" build.tag}}-{{/if}}linux-arm32v7
+    platform:
+      architecture: arm
+      os: linux
+      variant: v7

docker/forgejo/Dockerfile

@@ -1,24 +0,0 @@
-FROM alpine:3.19
-
-ARG UID=991
-ARG GID=991
-
-ENV \
-  UID=${UID} \
-  GID=${GID}
-
-USER root
-RUN \
-  addgroup -g "${GID}" app && \
-  adduser -D -G app -u "${UID}" -g "" -h /opt/app app && \
-  apk add tini && \
-  chown -R app:app /mnt
-
-COPY relay /usr/local/bin/relay
-
-USER app
-EXPOSE 6669
-EXPOSE 8080
-VOLUME /mnt
-ENTRYPOINT ["/sbin/tini", "--"]
-CMD ["/usr/local/bin/relay"]

docker/prod/docker-compose.yml

@@ -2,7 +2,7 @@ version: '3.3'
 
 services:
   relay:
-    image: asonix/relay:0.3.115
+    image: asonix/relay:0.3.85
     ports:
       - "8079:8079"
     restart: always

flake.lock

@@ -5,11 +5,11 @@
         "systems": "systems"
       },
       "locked": {
-        "lastModified": 1710146030,
+        "lastModified": 1689068808,
-        "narHash": "sha256-SZ5L6eA7HJ/nmkzGG7/ISclqe6oZdOZTNoesiInkXPQ=",
+        "narHash": "sha256-6ixXo3wt24N/melDWjq70UuHQLxGV8jZvooRanIHXw0=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "b1d9ab70662946ef0850d488da1c9019f3a9752a",
+        "rev": "919d646de7be200f3bf08cb76ae1f09402b6f9b4",
         "type": "github"
       },
       "original": {
@@ -20,16 +20,16 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1733550349,
+        "lastModified": 1689679375,
-        "narHash": "sha256-NcGumB4Lr6KSDq+nIqXtNA8QwAQKDSZT7N9OTGWbTrs=",
+        "narHash": "sha256-LHUC52WvyVDi9PwyL1QCpaxYWBqp4ir4iL6zgOkmcb8=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "e2605d0744c2417b09f8bf850dfca42fcf537d34",
+        "rev": "684c17c429c42515bafb3ad775d2a710947f3d67",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "nixos-24.11",
+        "ref": "nixos-unstable",
         "repo": "nixpkgs",
         "type": "github"
       }

flake.nix

@@ -2,7 +2,7 @@
   description = "relay";
 
   inputs = {
-    nixpkgs.url = "github:NixOS/nixpkgs/nixos-24.11";
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
     flake-utils.url = "github:numtide/flake-utils";
   };
 

relay.nix

@@ -1,22 +1,24 @@
 { lib
 , nixosTests
+, protobuf
 , rustPlatform
 }:
 
 rustPlatform.buildRustPackage {
   pname = "relay";
-  version = "0.3.116";
+  version = "0.3.88";
   src = ./.;
   cargoLock.lockFile = ./Cargo.lock;
 
-  RUSTFLAGS = "--cfg tokio_unstable";
+  PROTOC = "${protobuf}/bin/protoc";
+  PROTOC_INCLUDE = "${protobuf}/include";
 
   nativeBuildInputs = [ ];
 
   passthru.tests = { inherit (nixosTests) relay; };
 
   meta = with lib; {
-    description = "An ActivityPub relay";
+    description = "A simple image hosting service";
     homepage = "https://git.asonix.dog/asonix/relay";
     license = with licenses; [ agpl3Plus ];
   };

src/admin/client.rs

@@ -3,13 +3,12 @@ use crate::{
     collector::Snapshot,
     config::{AdminUrlKind, Config},
     error::{Error, ErrorKind},
-    extractors::XApiToken,
 };
-use reqwest_middleware::ClientWithMiddleware;
+use awc::Client;
 use serde::de::DeserializeOwned;
 
 pub(crate) async fn allow(
-    client: &ClientWithMiddleware,
+    client: &Client,
     config: &Config,
     domains: Vec<String>,
 ) -> Result<(), Error> {
@@ -17,7 +16,7 @@ pub(crate) async fn allow(
 }
 
 pub(crate) async fn disallow(
-    client: &ClientWithMiddleware,
+    client: &Client,
     config: &Config,
     domains: Vec<String>,
 ) -> Result<(), Error> {
@@ -25,7 +24,7 @@ pub(crate) async fn disallow(
 }
 
 pub(crate) async fn block(
-    client: &ClientWithMiddleware,
+    client: &Client,
     config: &Config,
     domains: Vec<String>,
 ) -> Result<(), Error> {
@@ -33,50 +32,35 @@ pub(crate) async fn block(
 }
 
 pub(crate) async fn unblock(
-    client: &ClientWithMiddleware,
+    client: &Client,
     config: &Config,
     domains: Vec<String>,
 ) -> Result<(), Error> {
     post_domains(client, config, domains, AdminUrlKind::Unblock).await
 }
 
-pub(crate) async fn allowed(
+pub(crate) async fn allowed(client: &Client, config: &Config) -> Result<AllowedDomains, Error> {
-    client: &ClientWithMiddleware,
-    config: &Config,
-) -> Result<AllowedDomains, Error> {
     get_results(client, config, AdminUrlKind::Allowed).await
 }
 
-pub(crate) async fn blocked(
+pub(crate) async fn blocked(client: &Client, config: &Config) -> Result<BlockedDomains, Error> {
-    client: &ClientWithMiddleware,
-    config: &Config,
-) -> Result<BlockedDomains, Error> {
     get_results(client, config, AdminUrlKind::Blocked).await
 }
 
-pub(crate) async fn connected(
+pub(crate) async fn connected(client: &Client, config: &Config) -> Result<ConnectedActors, Error> {
-    client: &ClientWithMiddleware,
-    config: &Config,
-) -> Result<ConnectedActors, Error> {
     get_results(client, config, AdminUrlKind::Connected).await
 }
 
-pub(crate) async fn stats(
+pub(crate) async fn stats(client: &Client, config: &Config) -> Result<Snapshot, Error> {
-    client: &ClientWithMiddleware,
-    config: &Config,
-) -> Result<Snapshot, Error> {
     get_results(client, config, AdminUrlKind::Stats).await
 }
 
-pub(crate) async fn last_seen(
+pub(crate) async fn last_seen(client: &Client, config: &Config) -> Result<LastSeen, Error> {
-    client: &ClientWithMiddleware,
-    config: &Config,
-) -> Result<LastSeen, Error> {
     get_results(client, config, AdminUrlKind::LastSeen).await
 }
 
 async fn get_results<T: DeserializeOwned>(
-    client: &ClientWithMiddleware,
+    client: &Client,
     config: &Config,
     url_kind: AdminUrlKind,
 ) -> Result<T, Error> {
@@ -84,19 +68,15 @@ async fn get_results<T: DeserializeOwned>(
 
     let iri = config.generate_admin_url(url_kind);
 
-    let res = client
+    let mut res = client
         .get(iri.as_str())
-        .header(XApiToken::http1_name(), x_api_token.to_string())
+        .insert_header(x_api_token)
         .send()
         .await
         .map_err(|e| ErrorKind::SendRequest(iri.to_string(), e.to_string()))?;
 
     if !res.status().is_success() {
-        return Err(ErrorKind::Status(
+        return Err(ErrorKind::Status(iri.to_string(), res.status()).into());
-            iri.to_string(),
-            crate::http1::status_to_http02(res.status()),
-        )
-        .into());
     }
 
     let t = res
@@ -108,7 +88,7 @@ async fn get_results<T: DeserializeOwned>(
 }
 
 async fn post_domains(
-    client: &ClientWithMiddleware,
+    client: &Client,
     config: &Config,
     domains: Vec<String>,
     url_kind: AdminUrlKind,
@@ -119,9 +99,8 @@ async fn post_domains(
 
     let res = client
         .post(iri.as_str())
-        .header(XApiToken::http1_name(), x_api_token.to_string())
+        .insert_header(x_api_token)
-        .json(&Domains { domains })
+        .send_json(&Domains { domains })
-        .send()
         .await
         .map_err(|e| ErrorKind::SendRequest(iri.to_string(), e.to_string()))?;
 

src/build.rs

@@ -21,7 +21,7 @@ fn git_info() {
     }
 }
 
-fn version_info() -> color_eyre::Result<()> {
+fn version_info() -> Result<(), anyhow::Error> {
     let cargo_toml = Path::new(&std::env::var("CARGO_MANIFEST_DIR")?).join("Cargo.toml");
 
     let mut file = File::open(cargo_toml)?;
@@ -42,7 +42,7 @@ fn version_info() -> color_eyre::Result<()> {
     Ok(())
 }
 
-fn main() -> color_eyre::Result<()> {
+fn main() -> Result<(), anyhow::Error> {
     dotenv::dotenv().ok();
 
     git_info();

src/collector.rs

@@ -1,4 +1,4 @@
-use metrics::{Key, Metadata, Recorder, SetRecorderError};
+use metrics::{Key, Recorder, SetRecorderError};
 use metrics_util::{
     registry::{AtomicStorage, GenerationalStorage, Recency, Registry},
     MetricKindMask, Summary,
@@ -15,10 +15,6 @@ const MINUTES: u64 = 60 * SECONDS;
 const HOURS: u64 = 60 * MINUTES;
 const DAYS: u64 = 24 * HOURS;
 
-pub(crate) fn recordable(len: usize) -> u32 {
-    ((len as u64) % u64::from(u32::MAX)) as u32
-}
-
 type DistributionMap = BTreeMap<Vec<(String, String)>, Summary>;
 
 #[derive(Clone)]
@@ -293,7 +289,7 @@ impl Inner {
             }
 
             let mut d = self.distributions.write().unwrap();
-            let outer_entry = d.entry(name.clone()).or_default();
+            let outer_entry = d.entry(name.clone()).or_insert_with(BTreeMap::new);
 
             let entry = outer_entry
                 .entry(labels)
@@ -303,14 +299,7 @@ impl Inner {
                 for sample in samples {
                     entry.add(*sample);
                 }
-            });
+            })
-
-            let mut total_len = 0;
-            for dist_map in d.values() {
-                total_len += dist_map.len();
-            }
-
-            metrics::gauge!("relay.collector.distributions.size").set(recordable(total_len));
         }
 
         let d = self.distributions.read().unwrap().clone();
@@ -369,11 +358,10 @@ impl MemoryCollector {
     ) {
         let mut d = self.inner.descriptions.write().unwrap();
         d.entry(key.as_str().to_owned()).or_insert(description);
-        metrics::gauge!("relay.collector.descriptions.size").set(recordable(d.len()));
     }
 
-    pub(crate) fn install(&self) -> Result<(), SetRecorderError<Self>> {
+    pub(crate) fn install(&self) -> Result<(), SetRecorderError> {
-        metrics::set_global_recorder(self.clone())
+        metrics::set_boxed_recorder(Box::new(self.clone()))
     }
 }
 
@@ -405,19 +393,19 @@ impl Recorder for MemoryCollector {
         self.add_description_if_missing(&key, description)
     }
 
-    fn register_counter(&self, key: &Key, _: &Metadata<'_>) -> metrics::Counter {
+    fn register_counter(&self, key: &Key) -> metrics::Counter {
         self.inner
             .registry
             .get_or_create_counter(key, |c| c.clone().into())
     }
 
-    fn register_gauge(&self, key: &Key, _: &Metadata<'_>) -> metrics::Gauge {
+    fn register_gauge(&self, key: &Key) -> metrics::Gauge {
         self.inner
             .registry
             .get_or_create_gauge(key, |c| c.clone().into())
     }
 
-    fn register_histogram(&self, key: &Key, _: &Metadata<'_>) -> metrics::Histogram {
+    fn register_histogram(&self, key: &Key) -> metrics::Histogram {
         self.inner
             .registry
             .get_or_create_histogram(key, |c| c.clone().into())

src/config.rs

@@ -11,9 +11,11 @@ use activitystreams::{
     },
 };
 use config::Environment;
-use http_signature_normalization_actix::{digest::ring::Sha256, prelude::VerifyDigest};
+use http_signature_normalization_actix::prelude::VerifyDigest;
-use rustls::sign::CertifiedKey;
+use rsa::sha2::{Digest, Sha256};
+use rustls::{Certificate, PrivateKey};
 use std::{
+    io::BufReader,
     net::{IpAddr, SocketAddr},
     path::PathBuf,
 };
@@ -45,10 +47,7 @@ pub(crate) struct ParsedConfig {
     prometheus_port: Option<u16>,
     deliver_concurrency: u64,
     client_timeout: u64,
-    proxy_url: Option<IriString>,
+    client_pool_size: usize,
-    proxy_username: Option<String>,
-    proxy_password: Option<String>,
-    signature_threads: Option<usize>,
 }
 
 #[derive(Clone)]
@@ -74,8 +73,7 @@ pub struct Config {
     prometheus_config: Option<PrometheusConfig>,
     deliver_concurrency: u64,
     client_timeout: u64,
-    proxy_config: Option<ProxyConfig>,
+    client_pool_size: usize,
-    signature_threads: Option<usize>,
 }
 
 #[derive(Clone)]
@@ -90,12 +88,6 @@ struct PrometheusConfig {
     port: u16,
 }
 
-#[derive(Clone, Debug)]
-struct ProxyConfig {
-    url: IriString,
-    auth: Option<(String, String)>,
-}
-
 #[derive(Debug)]
 pub enum UrlKind {
     Activity,
@@ -151,8 +143,7 @@ impl std::fmt::Debug for Config {
             .field("prometheus_config", &self.prometheus_config)
             .field("deliver_concurrency", &self.deliver_concurrency)
             .field("client_timeout", &self.client_timeout)
-            .field("proxy_config", &self.proxy_config)
+            .field("client_pool_size", &self.client_pool_size)
-            .field("signature_threads", &self.signature_threads)
             .finish()
     }
 }
@@ -184,10 +175,7 @@ impl Config {
             .set_default("prometheus_port", None as Option<u16>)?
             .set_default("deliver_concurrency", 8u64)?
             .set_default("client_timeout", 10u64)?
-            .set_default("proxy_url", None as Option<&str>)?
+            .set_default("client_pool_size", 20u64)?
-            .set_default("proxy_username", None as Option<&str>)?
-            .set_default("proxy_password", None as Option<&str>)?
-            .set_default("signature_threads", None as Option<u64>)?
             .add_source(Environment::default())
             .build()?;
 
@@ -229,26 +217,6 @@ impl Config {
             (None, None) => None,
         };
 
-        let proxy_config = match (config.proxy_username, config.proxy_password) {
-            (Some(username), Some(password)) => config.proxy_url.map(|url| ProxyConfig {
-                url,
-                auth: Some((username, password)),
-            }),
-            (Some(_), None) => {
-                tracing::warn!(
-                    "PROXY_USERNAME is set but PROXY_PASSWORD is not set, not setting Proxy Auth"
-                );
-                config.proxy_url.map(|url| ProxyConfig { url, auth: None })
-            }
-            (None, Some(_)) => {
-                tracing::warn!(
-                    "PROXY_PASSWORD is set but PROXY_USERNAME is not set, not setting Proxy Auth"
-                );
-                config.proxy_url.map(|url| ProxyConfig { url, auth: None })
-            }
-            (None, None) => config.proxy_url.map(|url| ProxyConfig { url, auth: None }),
-        };
-
         let source_url = match Self::git_hash() {
             Some(hash) => format!(
                 "{}{}{hash}",
@@ -281,22 +249,10 @@ impl Config {
             prometheus_config,
             deliver_concurrency: config.deliver_concurrency,
             client_timeout: config.client_timeout,
-            proxy_config,
+            client_pool_size: config.client_pool_size,
-            signature_threads: config.signature_threads,
         })
     }
 
-    pub(crate) fn signature_threads(&self) -> usize {
-        self.signature_threads
-            .unwrap_or_else(|| {
-                std::thread::available_parallelism()
-                    .map(usize::from)
-                    .map_err(|e| tracing::warn!("Failed to get parallelism, {e}"))
-                    .unwrap_or(1)
-            })
-            .max(1)
-    }
-
     pub(crate) fn client_timeout(&self) -> u64 {
         self.client_timeout
     }
@@ -311,34 +267,43 @@ impl Config {
         Some((config.addr, config.port).into())
     }
 
-    pub(crate) async fn open_keys(&self) -> Result<Option<CertifiedKey>, Error> {
+    pub(crate) fn open_keys(&self) -> Result<Option<(Vec<Certificate>, PrivateKey)>, Error> {
         let tls = if let Some(tls) = &self.tls {
             tls
         } else {
-            tracing::info!("No TLS config present");
+            tracing::warn!("No TLS config present");
             return Ok(None);
         };
 
-        let certs_bytes = tokio::fs::read(&tls.cert).await?;
+        let mut certs_reader = BufReader::new(std::fs::File::open(&tls.cert)?);
-        let certs =
+        let certs = rustls_pemfile::certs(&mut certs_reader)?;
-            rustls_pemfile::certs(&mut certs_bytes.as_slice()).collect::<Result<Vec<_>, _>>()?;
 
         if certs.is_empty() {
             tracing::warn!("No certs read from certificate file");
             return Ok(None);
         }
 
-        let key_bytes = tokio::fs::read(&tls.key).await?;
+        let mut key_reader = BufReader::new(std::fs::File::open(&tls.key)?);
-        let key = if let Some(key) = rustls_pemfile::private_key(&mut key_bytes.as_slice())? {
+        let key = rustls_pemfile::read_one(&mut key_reader)?;
-            key
+
+        let certs = certs.into_iter().map(Certificate).collect();
+
+        let key = if let Some(key) = key {
+            match key {
+                rustls_pemfile::Item::RSAKey(der) => PrivateKey(der),
+                rustls_pemfile::Item::PKCS8Key(der) => PrivateKey(der),
+                rustls_pemfile::Item::ECKey(der) => PrivateKey(der),
+                _ => {
+                    tracing::warn!("Unknown key format: {:?}", key);
+                    return Ok(None);
+                }
+            }
         } else {
             tracing::warn!("Failed to read private key");
             return Ok(None);
         };
 
-        let key = rustls::crypto::ring::sign::any_supported_type(&key)?;
+        Ok(Some((certs, key)))
-
-        Ok(Some(CertifiedKey::new(certs, key)))
     }
 
     pub(crate) fn footer_blurb(&self) -> Option<crate::templates::Html<String>> {
@@ -488,10 +453,8 @@ impl Config {
         )
     }
 
-    pub(crate) fn proxy_config(&self) -> Option<(&IriString, Option<(&str, &str)>)> {
+    pub(crate) fn client_pool_size(&self) -> usize {
-        self.proxy_config.as_ref().map(|ProxyConfig { url, auth }| {
+        self.client_pool_size
-            (url, auth.as_ref().map(|(u, p)| (u.as_str(), p.as_str())))
-        })
     }
 
     pub(crate) fn source_code(&self) -> &IriString {

src/data/actor.rs

@@ -2,7 +2,7 @@ use crate::{
     apub::AcceptedActors,
     db::{Actor, Db},
     error::{Error, ErrorKind},
-    requests::{BreakerStrategy, Requests},
+    requests::Requests,
 };
 use activitystreams::{iri_string::types::IriString, prelude::*};
 use std::time::{Duration, SystemTime};
@@ -71,9 +71,7 @@ impl ActorCache {
         id: &IriString,
         requests: &Requests,
     ) -> Result<Actor, Error> {
-        let accepted_actor = requests
+        let accepted_actor = requests.fetch::<AcceptedActors>(id).await?;
-            .fetch::<AcceptedActors>(id, BreakerStrategy::Require2XX)
-            .await?;
 
         let input_authority = id.authority_components().ok_or(ErrorKind::MissingDomain)?;
         let accepted_actor_id = accepted_actor

src/data/last_online.rs

@@ -9,10 +9,10 @@ pub(crate) struct LastOnline {
 impl LastOnline {
     pub(crate) fn mark_seen(&self, iri: &IriStr) {
         if let Some(authority) = iri.authority_str() {
-            let mut guard = self.domains.lock().unwrap();
+            self.domains
-            guard.insert(authority.to_string(), OffsetDateTime::now_utc());
+                .lock()
-            metrics::gauge!("relay.last-online.size",)
+                .unwrap()
-                .set(crate::collector::recordable(guard.len()));
+                .insert(authority.to_string(), OffsetDateTime::now_utc());
         }
     }
 

src/data/state.rs

@@ -1,15 +1,14 @@
 use crate::{
+    config::{Config, UrlKind},
     data::NodeCache,
     db::Db,
     error::Error,
     requests::{Breakers, Requests},
-    spawner::Spawner,
 };
 use activitystreams::iri_string::types::IriString;
 use actix_web::web;
 use lru::LruCache;
 use rand::thread_rng;
-use reqwest_middleware::ClientWithMiddleware;
 use rsa::{RsaPrivateKey, RsaPublicKey};
 use std::sync::{Arc, RwLock};
 
@@ -17,10 +16,10 @@ use super::LastOnline;
 
 #[derive(Clone)]
 pub struct State {
-    pub(crate) requests: Requests,
     pub(crate) public_key: RsaPublicKey,
+    private_key: RsaPrivateKey,
     object_cache: Arc<RwLock<LruCache<IriString, IriString>>>,
-    pub(crate) node_cache: NodeCache,
+    node_cache: NodeCache,
     breakers: Breakers,
     pub(crate) last_online: Arc<LastOnline>,
     pub(crate) db: Db,
@@ -37,6 +36,22 @@ impl std::fmt::Debug for State {
 }
 
 impl State {
+    pub(crate) fn node_cache(&self) -> NodeCache {
+        self.node_cache.clone()
+    }
+
+    pub(crate) fn requests(&self, config: &Config) -> Requests {
+        Requests::new(
+            config.generate_url(UrlKind::MainKey).to_string(),
+            self.private_key.clone(),
+            config.user_agent(),
+            self.breakers.clone(),
+            self.last_online.clone(),
+            config.client_pool_size(),
+            config.client_timeout(),
+        )
+    }
+
     #[tracing::instrument(
         level = "debug",
         name = "Get inboxes for other domains",
@@ -73,22 +88,11 @@ impl State {
     }
 
     pub(crate) fn cache(&self, object_id: IriString, actor_id: IriString) {
-        let mut guard = self.object_cache.write().unwrap();
+        self.object_cache.write().unwrap().put(object_id, actor_id);
-        guard.put(object_id, actor_id);
-        metrics::gauge!("relay.object-cache.size").set(crate::collector::recordable(guard.len()));
-    }
-
-    pub(crate) fn is_connected(&self, iri: &IriString) -> bool {
-        self.breakers.should_try(iri)
     }
 
     #[tracing::instrument(level = "debug", name = "Building state", skip_all)]
-    pub(crate) async fn build(
+    pub(crate) async fn build(db: Db) -> Result<Self, Error> {
-        db: Db,
-        key_id: String,
-        spawner: Spawner,
-        client: ClientWithMiddleware,
-    ) -> Result<Self, Error> {
         let private_key = if let Ok(Some(key)) = db.private_key().await {
             tracing::debug!("Using existing key");
             key
@@ -107,28 +111,16 @@ impl State {
 
         let public_key = private_key.to_public_key();
 
-        let breakers = Breakers::default();
-        let last_online = Arc::new(LastOnline::empty());
-
-        let requests = Requests::new(
-            key_id,
-            private_key,
-            breakers.clone(),
-            last_online.clone(),
-            spawner,
-            client,
-        );
-
         let state = State {
-            requests,
             public_key,
+            private_key,
             object_cache: Arc::new(RwLock::new(LruCache::new(
                 (1024 * 8).try_into().expect("nonzero"),
             ))),
             node_cache: NodeCache::new(db.clone()),
-            breakers,
+            breakers: Breakers::default(),
             db,
-            last_online,
+            last_online: Arc::new(LastOnline::empty()),
         };
 
         Ok(state)

src/db.rs

@@ -7,7 +7,7 @@ use rsa::{
     pkcs8::{DecodePrivateKey, EncodePrivateKey},
     RsaPrivateKey,
 };
-use sled::{transaction::TransactionError, Batch, Transactional, Tree};
+use sled::{Batch, Tree};
 use std::{
     collections::{BTreeMap, HashMap},
     sync::{
@@ -283,15 +283,10 @@ impl Db {
     pub(crate) async fn check_health(&self) -> Result<(), Error> {
         let next = self.inner.healthz_counter.fetch_add(1, Ordering::Relaxed);
         self.unblock(move |inner| {
-            let res = inner
+            inner
                 .healthz
                 .insert("healthz", &next.to_be_bytes()[..])
-                .map_err(Error::from);
+                .map_err(Error::from)
-
-            metrics::gauge!("relay.db.healthz.size")
-                .set(crate::collector::recordable(inner.healthz.len()));
-
-            res
         })
         .await?;
         self.inner.healthz.flush_async().await?;
@@ -354,9 +349,6 @@ impl Db {
                 .actor_id_info
                 .insert(actor_id.as_str().as_bytes(), vec)?;
 
-            metrics::gauge!("relay.db.actor-id-info.size")
-                .set(crate::collector::recordable(inner.actor_id_info.len()));
-
             Ok(())
         })
         .await
@@ -391,9 +383,6 @@ impl Db {
                 .actor_id_instance
                 .insert(actor_id.as_str().as_bytes(), vec)?;
 
-            metrics::gauge!("relay.db.actor-id-instance.size")
-                .set(crate::collector::recordable(inner.actor_id_instance.len()));
-
             Ok(())
         })
         .await
@@ -428,9 +417,6 @@ impl Db {
                 .actor_id_contact
                 .insert(actor_id.as_str().as_bytes(), vec)?;
 
-            metrics::gauge!("relay.db.actor-id-contact.size")
-                .set(crate::collector::recordable(inner.actor_id_contact.len()));
-
             Ok(())
         })
         .await
@@ -461,12 +447,6 @@ impl Db {
             inner
                 .media_url_media_id
                 .insert(url.as_str().as_bytes(), id.as_bytes())?;
-
-            metrics::gauge!("relay.db.media-id-media-url.size")
-                .set(crate::collector::recordable(inner.media_id_media_url.len()));
-            metrics::gauge!("relay.db.media-url-media-id.size")
-                .set(crate::collector::recordable(inner.media_url_media_id.len()));
-
             Ok(())
         })
         .await
@@ -558,14 +538,6 @@ impl Db {
             inner
                 .actor_id_actor
                 .insert(actor.id.as_str().as_bytes(), vec)?;
-
-            metrics::gauge!("relay.db.public-key-actor-id.size").set(crate::collector::recordable(
-                inner.public_key_id_actor_id.len(),
-            ));
-
-            metrics::gauge!("relay.db.actor-id-actor.size").set(crate::collector::recordable(
-                inner.public_key_id_actor_id.len(),
-            ));
             Ok(())
         })
         .await
@@ -578,10 +550,6 @@ impl Db {
                 .connected_actor_ids
                 .remove(actor_id.as_str().as_bytes())?;
 
-            metrics::gauge!("relay.db.connected-actor-ids.size").set(crate::collector::recordable(
-                inner.connected_actor_ids.len(),
-            ));
-
             Ok(())
         })
         .await
@@ -594,10 +562,6 @@ impl Db {
                 .connected_actor_ids
                 .insert(actor_id.as_str().as_bytes(), actor_id.as_str().as_bytes())?;
 
-            metrics::gauge!("relay.db.connected-actor-ids.size").set(crate::collector::recordable(
-                inner.connected_actor_ids.len(),
-            ));
-
             Ok(())
         })
         .await
@@ -605,64 +569,30 @@ impl Db {
 
     pub(crate) async fn add_blocks(&self, domains: Vec<String>) -> Result<(), Error> {
         self.unblock(move |inner| {
-            let connected_by_domain = inner.connected_by_domain(&domains).collect::<Vec<_>>();
+            for connected in inner.connected_by_domain(&domains) {
-
+                inner
-            let res = (
+                    .connected_actor_ids
-                &inner.connected_actor_ids,
+                    .remove(connected.as_str().as_bytes())?;
-                &inner.blocked_domains,
-                &inner.allowed_domains,
-            )
-                .transaction(|(connected, blocked, allowed)| {
-                    let mut connected_batch = Batch::default();
-                    let mut blocked_batch = Batch::default();
-                    let mut allowed_batch = Batch::default();
-
-                    for connected in &connected_by_domain {
-                        connected_batch.remove(connected.as_str().as_bytes());
             }
 
             for authority in &domains {
-                        blocked_batch
+                inner
-                            .insert(domain_key(authority).as_bytes(), authority.as_bytes());
+                    .blocked_domains
-                        allowed_batch.remove(domain_key(authority).as_bytes());
+                    .insert(domain_key(authority), authority.as_bytes())?;
+                inner.allowed_domains.remove(domain_key(authority))?;
             }
 
-                    connected.apply_batch(&connected_batch)?;
-                    blocked.apply_batch(&blocked_batch)?;
-                    allowed.apply_batch(&allowed_batch)?;
-
             Ok(())
-                });
-
-            metrics::gauge!("relay.db.connected-actor-ids.size").set(crate::collector::recordable(
-                inner.connected_actor_ids.len(),
-            ));
-            metrics::gauge!("relay.db.blocked-domains.size")
-                .set(crate::collector::recordable(inner.blocked_domains.len()));
-            metrics::gauge!("relay.db.allowed-domains.size")
-                .set(crate::collector::recordable(inner.allowed_domains.len()));
-
-            match res {
-                Ok(()) => Ok(()),
-                Err(TransactionError::Abort(e) | TransactionError::Storage(e)) => Err(e.into()),
-            }
         })
         .await
     }
 
     pub(crate) async fn remove_blocks(&self, domains: Vec<String>) -> Result<(), Error> {
         self.unblock(move |inner| {
-            let mut blocked_batch = Batch::default();
-
             for authority in &domains {
-                blocked_batch.remove(domain_key(authority).as_bytes());
+                inner.blocked_domains.remove(domain_key(authority))?;
             }
 
-            inner.blocked_domains.apply_batch(blocked_batch)?;
-
-            metrics::gauge!("relay.db.blocked-domains.size")
-                .set(crate::collector::recordable(inner.blocked_domains.len()));
-
             Ok(())
         })
         .await
@@ -670,17 +600,12 @@ impl Db {
 
     pub(crate) async fn add_allows(&self, domains: Vec<String>) -> Result<(), Error> {
         self.unblock(move |inner| {
-            let mut allowed_batch = Batch::default();
-
             for authority in &domains {
-                allowed_batch.insert(domain_key(authority).as_bytes(), authority.as_bytes());
+                inner
+                    .allowed_domains
+                    .insert(domain_key(authority), authority.as_bytes())?;
             }
 
-            inner.allowed_domains.apply_batch(allowed_batch)?;
-
-            metrics::gauge!("relay.db.allowed-domains.size")
-                .set(crate::collector::recordable(inner.allowed_domains.len()));
-
             Ok(())
         })
         .await
@@ -689,32 +614,17 @@ impl Db {
     pub(crate) async fn remove_allows(&self, domains: Vec<String>) -> Result<(), Error> {
         self.unblock(move |inner| {
             if inner.restricted_mode {
-                let connected_by_domain = inner.connected_by_domain(&domains).collect::<Vec<_>>();
+                for connected in inner.connected_by_domain(&domains) {
-
+                    inner
-                let mut connected_batch = Batch::default();
+                        .connected_actor_ids
-
+                        .remove(connected.as_str().as_bytes())?;
-                for connected in &connected_by_domain {
-                    connected_batch.remove(connected.as_str().as_bytes());
                 }
-
-                inner.connected_actor_ids.apply_batch(connected_batch)?;
-
-                metrics::gauge!("relay.db.connected-actor-ids.size").set(
-                    crate::collector::recordable(inner.connected_actor_ids.len()),
-                );
             }
 
-            let mut allowed_batch = Batch::default();
-
             for authority in &domains {
-                allowed_batch.remove(domain_key(authority).as_bytes());
+                inner.allowed_domains.remove(domain_key(authority))?;
             }
 
-            inner.allowed_domains.apply_batch(allowed_batch)?;
-
-            metrics::gauge!("relay.db.allowed-domains.size")
-                .set(crate::collector::recordable(inner.allowed_domains.len()));
-
             Ok(())
         })
         .await
@@ -755,10 +665,6 @@ impl Db {
             inner
                 .settings
                 .insert("private-key".as_bytes(), pem_pkcs8.as_bytes())?;
-
-            metrics::gauge!("relay.db.settings.size")
-                .set(crate::collector::recordable(inner.settings.len()));
-
             Ok(())
         })
         .await
@@ -844,11 +750,6 @@ mod tests {
     {
         let db =
             Db::build_inner(true, sled::Config::new().temporary(true).open().unwrap()).unwrap();
-
+        actix_rt::System::new().block_on((f)(db));
-        tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .unwrap()
-            .block_on((f)(db));
     }
 }

src/error.rs

@@ -1,85 +1,57 @@
 use activitystreams::checked::CheckError;
+use actix_rt::task::JoinError;
 use actix_web::{
     error::{BlockingError, ResponseError},
     http::StatusCode,
     HttpResponse,
 };
-use background_jobs::BoxError;
+use http_signature_normalization_actix::PrepareSignError;
-use color_eyre::eyre::Error as Report;
+use std::{convert::Infallible, fmt::Debug, io};
-use http_signature_normalization_reqwest::SignError;
+use tracing_error::SpanTrace;
-use std::{convert::Infallible, io, sync::Arc};
-use tokio::task::JoinError;
-
-#[derive(Clone)]
-struct ArcKind {
-    kind: Arc<ErrorKind>,
-}
-
-impl std::fmt::Debug for ArcKind {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        self.kind.fmt(f)
-    }
-}
-
-impl std::fmt::Display for ArcKind {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        self.kind.fmt(f)
-    }
-}
-
-impl std::error::Error for ArcKind {
-    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
-        self.kind.source()
-    }
-}
 
 pub(crate) struct Error {
-    kind: ArcKind,
+    context: String,
-    display: Box<str>,
+    kind: ErrorKind,
-    debug: Box<str>,
 }
 
 impl Error {
-    fn kind(&self) -> &ErrorKind {
-        &self.kind.kind
-    }
-
     pub(crate) fn is_breaker(&self) -> bool {
-        matches!(self.kind(), ErrorKind::Breaker)
+        matches!(self.kind, ErrorKind::Breaker)
     }
 
     pub(crate) fn is_not_found(&self) -> bool {
-        matches!(self.kind(), ErrorKind::Status(_, StatusCode::NOT_FOUND))
+        matches!(self.kind, ErrorKind::Status(_, StatusCode::NOT_FOUND))
     }
 
     pub(crate) fn is_bad_request(&self) -> bool {
-        matches!(self.kind(), ErrorKind::Status(_, StatusCode::BAD_REQUEST))
+        matches!(self.kind, ErrorKind::Status(_, StatusCode::BAD_REQUEST))
     }
 
     pub(crate) fn is_gone(&self) -> bool {
-        matches!(self.kind(), ErrorKind::Status(_, StatusCode::GONE))
+        matches!(self.kind, ErrorKind::Status(_, StatusCode::GONE))
     }
 
     pub(crate) fn is_malformed_json(&self) -> bool {
-        matches!(self.kind(), ErrorKind::Json(_))
+        matches!(self.kind, ErrorKind::Json(_))
     }
 }
 
 impl std::fmt::Debug for Error {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(&self.debug)
+        writeln!(f, "{:?}", self.kind)
     }
 }
 
 impl std::fmt::Display for Error {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(&self.display)
+        writeln!(f, "{}", self.kind)?;
+        std::fmt::Display::fmt(&self.context, f)
     }
 }
 
 impl std::error::Error for Error {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
-        self.kind().source()
+        self.kind.source()
     }
 }
 
@@ -88,82 +60,56 @@ where
     ErrorKind: From<T>,
 {
     fn from(error: T) -> Self {
-        let kind = ArcKind {
-            kind: Arc::new(ErrorKind::from(error)),
-        };
-        let report = Report::new(kind.clone());
-        let display = format!("{report}");
-        let debug = format!("{report:?}");
-
         Error {
-            kind,
+            context: SpanTrace::capture().to_string(),
-            display: Box::from(display),
+            kind: error.into(),
-            debug: Box::from(debug),
         }
     }
 }
 
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum ErrorKind {
-    #[error("Error in extractor")]
+    #[error("Error queueing job, {0}")]
-    Extractor(#[from] crate::extractors::ErrorKind),
+    Queue(anyhow::Error),
 
-    #[error("Error queueing job")]
+    #[error("Error in configuration, {0}")]
-    Queue(#[from] BoxError),
-
-    #[error("Error in configuration")]
     Config(#[from] config::ConfigError),
 
-    #[error("Couldn't parse key")]
+    #[error("Couldn't parse key, {0}")]
     Pkcs8(#[from] rsa::pkcs8::Error),
 
-    #[error("Couldn't encode public key")]
+    #[error("Couldn't encode public key, {0}")]
     Spki(#[from] rsa::pkcs8::spki::Error),
 
-    #[error("Couldn't sign request")]
+    #[error("Couldn't parse IRI, {0}")]
-    SignRequest,
-
-    #[error("Response body from server exceeded limits")]
-    BodyTooLarge,
-
-    #[error("Couldn't make request")]
-    Reqwest(#[from] reqwest::Error),
-
-    #[error("Couldn't make request")]
-    ReqwestMiddleware(#[from] reqwest_middleware::Error),
-
-    #[error("Couldn't parse IRI")]
     ParseIri(#[from] activitystreams::iri_string::validate::Error),
 
-    #[error("Couldn't normalize IRI")]
+    #[error("Couldn't normalize IRI, {0}")]
     NormalizeIri(#[from] std::collections::TryReserveError),
 
-    #[error("Couldn't perform IO")]
+    #[error("Couldn't perform IO, {0}")]
     Io(#[from] io::Error),
 
     #[error("Couldn't sign string, {0}")]
     Rsa(rsa::errors::Error),
 
-    #[error("Couldn't use db")]
+    #[error("Couldn't use db, {0}")]
     Sled(#[from] sled::Error),
 
-    #[error("Couldn't do the json thing")]
+    #[error("Couldn't do the json thing, {0}")]
     Json(#[from] serde_json::Error),
 
-    #[error("Couldn't sign request")]
+    #[error("Couldn't build signing string, {0}")]
-    Sign(#[from] SignError),
+    PrepareSign(#[from] PrepareSignError),
 
     #[error("Couldn't sign digest")]
     Signature(#[from] rsa::signature::Error),
 
-    #[error("Couldn't prepare TLS private key")]
+    #[error("Couldn't read signature")]
-    PrepareKey(#[from] rustls::Error),
+    ReadSignature(rsa::signature::Error),
 
     #[error("Couldn't verify signature")]
-    VerifySignature,
+    VerifySignature(rsa::signature::Error),
-
-    #[error("Failed to encode key der")]
-    DerEncode,
 
     #[error("Couldn't parse the signature header")]
     HeaderValidation(#[from] actix_web::http::header::InvalidHeaderValue),
@@ -189,10 +135,10 @@ pub(crate) enum ErrorKind {
     #[error("Wrong ActivityPub kind, {0}")]
     Kind(String),
 
-    #[error("Too many CPUs")]
+    #[error("Too many CPUs, {0}")]
     CpuCount(#[from] std::num::TryFromIntError),
 
-    #[error("Host mismatch")]
+    #[error("{0}")]
     HostMismatch(#[from] CheckError),
 
     #[error("Couldn't flush buffer")]
@@ -246,7 +192,7 @@ pub(crate) enum ErrorKind {
 
 impl ResponseError for Error {
     fn status_code(&self) -> StatusCode {
-        match self.kind() {
+        match self.kind {
             ErrorKind::NotAllowed(_) | ErrorKind::WrongActor(_) | ErrorKind::BadActor(_, _) => {
                 StatusCode::FORBIDDEN
             }
@@ -266,7 +212,7 @@ impl ResponseError for Error {
             .insert_header(("Content-Type", "application/activity+json"))
             .body(
                 serde_json::to_string(&serde_json::json!({
-                    "error": self.kind().to_string(),
+                    "error": self.kind.to_string(),
                 }))
                 .unwrap_or_else(|_| "{}".to_string()),
             )
@@ -296,15 +242,3 @@ impl From<rsa::errors::Error> for ErrorKind {
         ErrorKind::Rsa(e)
     }
 }
-
-impl From<http_signature_normalization_actix::Canceled> for ErrorKind {
-    fn from(_: http_signature_normalization_actix::Canceled) -> Self {
-        Self::Canceled
-    }
-}
-
-impl From<http_signature_normalization_reqwest::Canceled> for ErrorKind {
-    fn from(_: http_signature_normalization_reqwest::Canceled) -> Self {
-        Self::Canceled
-    }
-}

src/extractors.rs

@@ -1,15 +1,20 @@
 use actix_web::{
     dev::Payload,
-    error::ParseError,
+    error::{BlockingError, ParseError},
-    http::header::{from_one_raw_str, Header, HeaderName, HeaderValue, TryIntoHeaderValue},
+    http::{
+        header::{from_one_raw_str, Header, HeaderName, HeaderValue, TryIntoHeaderValue},
+        StatusCode,
+    },
     web::Data,
-    FromRequest, HttpMessage, HttpRequest,
+    FromRequest, HttpMessage, HttpRequest, HttpResponse, ResponseError,
 };
 use bcrypt::{BcryptError, DEFAULT_COST};
-use http_signature_normalization_actix::{prelude::InvalidHeaderValue, Canceled, Spawn};
+use futures_util::future::LocalBoxFuture;
+use http_signature_normalization_actix::prelude::InvalidHeaderValue;
 use std::{convert::Infallible, str::FromStr, time::Instant};
+use tracing_error::SpanTrace;
 
-use crate::{db::Db, error::Error, future::LocalBoxFuture, spawner::Spawner};
+use crate::db::Db;
 
 #[derive(Clone)]
 pub(crate) struct AdminConfig {
@@ -24,7 +29,7 @@ impl AdminConfig {
     }
 
     fn verify(&self, token: XApiToken) -> Result<bool, Error> {
-        bcrypt::verify(token.0, &self.hashed_api_token).map_err(Error::bcrypt_verify)
+        bcrypt::verify(&token.0, &self.hashed_api_token).map_err(Error::bcrypt_verify)
     }
 }
 
@@ -32,10 +37,10 @@ pub(crate) struct Admin {
     db: Data<Db>,
 }
 
-type PrepareTuple = (Data<Db>, Data<AdminConfig>, Data<Spawner>, XApiToken);
-
 impl Admin {
-    fn prepare_verify(req: &HttpRequest) -> Result<PrepareTuple, Error> {
+    fn prepare_verify(
+        req: &HttpRequest,
+    ) -> Result<(Data<Db>, Data<AdminConfig>, XApiToken), Error> {
         let hashed_api_token = req
             .app_data::<Data<AdminConfig>>()
             .ok_or_else(Error::missing_config)?
@@ -48,23 +53,16 @@ impl Admin {
             .ok_or_else(Error::missing_db)?
             .clone();
 
-        let spawner = req
+        Ok((db, hashed_api_token, x_api_token))
-            .app_data::<Data<Spawner>>()
-            .ok_or_else(Error::missing_spawner)?
-            .clone();
-
-        Ok((db, hashed_api_token, spawner, x_api_token))
     }
 
     #[tracing::instrument(level = "debug", skip_all)]
     async fn verify(
         hashed_api_token: Data<AdminConfig>,
-        spawner: Data<Spawner>,
         x_api_token: XApiToken,
     ) -> Result<(), Error> {
         let span = tracing::Span::current();
-        if spawner
+        if actix_web::web::block(move || span.in_scope(|| hashed_api_token.verify(x_api_token)))
-            .spawn_blocking(move || span.in_scope(|| hashed_api_token.verify(x_api_token)))
             .await
             .map_err(Error::canceled)??
         {
@@ -79,42 +77,67 @@ impl Admin {
     }
 }
 
+#[derive(Debug, thiserror::Error)]
+#[error("Failed authentication")]
+pub(crate) struct Error {
+    context: String,
+    #[source]
+    kind: ErrorKind,
+}
+
 impl Error {
     fn invalid() -> Self {
-        Error::from(ErrorKind::Invalid)
+        Error {
+            context: SpanTrace::capture().to_string(),
+            kind: ErrorKind::Invalid,
+        }
     }
 
     fn missing_config() -> Self {
-        Error::from(ErrorKind::MissingConfig)
+        Error {
+            context: SpanTrace::capture().to_string(),
+            kind: ErrorKind::MissingConfig,
+        }
     }
 
     fn missing_db() -> Self {
-        Error::from(ErrorKind::MissingDb)
+        Error {
+            context: SpanTrace::capture().to_string(),
+            kind: ErrorKind::MissingDb,
         }
-
-    fn missing_spawner() -> Self {
-        Error::from(ErrorKind::MissingSpawner)
     }
 
     fn bcrypt_verify(e: BcryptError) -> Self {
-        Error::from(ErrorKind::BCryptVerify(e))
+        Error {
+            context: SpanTrace::capture().to_string(),
+            kind: ErrorKind::BCryptVerify(e),
+        }
     }
 
     fn bcrypt_hash(e: BcryptError) -> Self {
-        Error::from(ErrorKind::BCryptHash(e))
+        Error {
+            context: SpanTrace::capture().to_string(),
+            kind: ErrorKind::BCryptHash(e),
+        }
     }
 
     fn parse_header(e: ParseError) -> Self {
-        Error::from(ErrorKind::ParseHeader(e))
+        Error {
+            context: SpanTrace::capture().to_string(),
+            kind: ErrorKind::ParseHeader(e),
+        }
     }
 
-    fn canceled(_: Canceled) -> Self {
+    fn canceled(_: BlockingError) -> Self {
-        Error::from(ErrorKind::Canceled)
+        Error {
+            context: SpanTrace::capture().to_string(),
+            kind: ErrorKind::Canceled,
+        }
     }
 }
 
 #[derive(Debug, thiserror::Error)]
-pub(crate) enum ErrorKind {
+enum ErrorKind {
     #[error("Invalid API Token")]
     Invalid,
 
@@ -124,9 +147,6 @@ pub(crate) enum ErrorKind {
     #[error("Missing Db")]
     MissingDb,
 
-    #[error("Missing Spawner")]
-    MissingSpawner,
-
     #[error("Panic in verify")]
     Canceled,
 
@@ -140,6 +160,20 @@ pub(crate) enum ErrorKind {
     ParseHeader(#[source] ParseError),
 }
 
+impl ResponseError for Error {
+    fn status_code(&self) -> StatusCode {
+        match self.kind {
+            ErrorKind::Invalid | ErrorKind::ParseHeader(_) => StatusCode::BAD_REQUEST,
+            _ => StatusCode::INTERNAL_SERVER_ERROR,
+        }
+    }
+
+    fn error_response(&self) -> HttpResponse {
+        HttpResponse::build(self.status_code())
+            .json(serde_json::json!({ "msg": self.kind.to_string() }))
+    }
+}
+
 impl FromRequest for Admin {
     type Error = Error;
     type Future = LocalBoxFuture<'static, Result<Self, Self::Error>>;
@@ -148,10 +182,12 @@ impl FromRequest for Admin {
         let now = Instant::now();
         let res = Self::prepare_verify(req);
         Box::pin(async move {
-            let (db, c, s, t) = res?;
+            let (db, c, t) = res?;
-            Self::verify(c, s, t).await?;
+            Self::verify(c, t).await?;
-            metrics::histogram!("relay.admin.verify")
+            metrics::histogram!(
-                .record(now.elapsed().as_micros() as f64 / 1_000_000_f64);
+                "relay.admin.verify",
+                now.elapsed().as_micros() as f64 / 1_000_000_f64
+            );
             Ok(Admin { db })
         })
     }
@@ -163,10 +199,6 @@ impl XApiToken {
     pub(crate) fn new(token: String) -> Self {
         Self(token)
     }
-
-    pub(crate) const fn http1_name() -> reqwest::header::HeaderName {
-        reqwest::header::HeaderName::from_static("x-api-token")
-    }
 }
 
 impl Header for XApiToken {
@@ -194,9 +226,3 @@ impl FromStr for XApiToken {
         Ok(XApiToken(s.to_string()))
     }
 }
-
-impl std::fmt::Display for XApiToken {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        self.0.fmt(f)
-    }
-}

src/future.rs

@@ -1,4 +0,0 @@
-use std::{future::Future, pin::Pin};
-
-pub(crate) type LocalBoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + 'a>>;
-pub(crate) type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

src/http1.rs

@@ -1,18 +0,0 @@
-pub(crate) fn name_to_http02(
-    name: &reqwest::header::HeaderName,
-) -> actix_web::http::header::HeaderName {
-    actix_web::http::header::HeaderName::from_bytes(name.as_ref())
-        .expect("headername conversions always work")
-}
-
-pub(crate) fn value_to_http02(
-    value: &reqwest::header::HeaderValue,
-) -> actix_web::http::header::HeaderValue {
-    actix_web::http::header::HeaderValue::from_bytes(value.as_bytes())
-        .expect("headervalue conversions always work")
-}
-
-pub(crate) fn status_to_http02(status: reqwest::StatusCode) -> actix_web::http::StatusCode {
-    actix_web::http::StatusCode::from_u16(status.as_u16())
-        .expect("statuscode conversions always work")
-}

src/jobs.rs

@@ -14,15 +14,14 @@ pub(crate) use self::{
 
 use crate::{
     config::Config,
-    data::{ActorCache, MediaCache, State},
+    data::{ActorCache, MediaCache, NodeCache, State},
     error::{Error, ErrorKind},
     jobs::{process_listeners::Listeners, record_last_online::RecordLastOnline},
+    requests::Requests,
 };
 use background_jobs::{
-    memory_storage::{Storage, TokioTimer},
+    memory_storage::{ActixTimer, Storage},
-    metrics::MetricsStorage,
+    Job, QueueHandle, WorkerConfig,
-    tokio::{QueueHandle, WorkerConfig},
-    Job,
 };
 use std::time::Duration;
 
@@ -40,20 +39,15 @@ fn debug_object(activity: &serde_json::Value) -> &serde_json::Value {
     object
 }
 
-pub(crate) fn build_storage() -> MetricsStorage<Storage<TokioTimer>> {
-    MetricsStorage::wrap(Storage::new(TokioTimer))
-}
-
 pub(crate) fn create_workers(
-    storage: MetricsStorage<Storage<TokioTimer>>,
     state: State,
     actors: ActorCache,
     media: MediaCache,
     config: Config,
-) -> std::io::Result<JobServer> {
+) -> JobServer {
     let deliver_concurrency = config.deliver_concurrency();
 
-    let queue_handle = WorkerConfig::new(storage, move |queue_handle| {
+    let queue_handle = WorkerConfig::new(Storage::new(ActixTimer), move |queue_handle| {
         JobState::new(
             state.clone(),
             actors.clone(),
@@ -77,20 +71,22 @@ pub(crate) fn create_workers(
     .set_worker_count("maintenance", 2)
     .set_worker_count("apub", 2)
     .set_worker_count("deliver", deliver_concurrency)
-    .start()?;
+    .start();
 
-    queue_handle.every(Duration::from_secs(60 * 5), Listeners)?;
+    queue_handle.every(Duration::from_secs(60 * 5), Listeners);
-    queue_handle.every(Duration::from_secs(60 * 10), RecordLastOnline)?;
+    queue_handle.every(Duration::from_secs(60 * 10), RecordLastOnline);
 
-    Ok(JobServer::new(queue_handle))
+    JobServer::new(queue_handle)
 }
 
 #[derive(Clone, Debug)]
 pub(crate) struct JobState {
+    requests: Requests,
     state: State,
     actors: ActorCache,
     config: Config,
     media: MediaCache,
+    node_cache: NodeCache,
     job_server: JobServer,
 }
 
@@ -116,10 +112,12 @@ impl JobState {
         config: Config,
     ) -> Self {
         JobState {
-            state,
+            requests: state.requests(&config),
+            node_cache: state.node_cache(),
             actors,
             config,
             media,
+            state,
             job_server,
         }
     }

src/jobs/apub/announce.rs

@@ -2,14 +2,14 @@ use crate::{
     config::{Config, UrlKind},
     db::Actor,
     error::Error,
-    future::BoxFuture,
     jobs::{
         apub::{get_inboxes, prepare_activity},
         DeliverMany, JobState,
     },
 };
 use activitystreams::{activity::Announce as AsAnnounce, iri_string::types::IriString};
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct Announce {
@@ -62,15 +62,14 @@ fn generate_announce(
     )
 }
 
-impl Job for Announce {
+impl ActixJob for Announce {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::apub::Announce";
     const QUEUE: &'static str = "apub";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }

src/jobs/apub/follow.rs

@@ -3,7 +3,6 @@ use crate::{
     config::{Config, UrlKind},
     db::Actor,
     error::{Error, ErrorKind},
-    future::BoxFuture,
     jobs::{apub::prepare_activity, Deliver, JobState, QueryInstance, QueryNodeinfo},
 };
 use activitystreams::{
@@ -11,7 +10,8 @@ use activitystreams::{
     iri_string::types::IriString,
     prelude::*,
 };
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct Follow {
@@ -111,15 +111,14 @@ fn generate_accept_follow(
     )
 }
 
-impl Job for Follow {
+impl ActixJob for Follow {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::apub::Follow";
     const QUEUE: &'static str = "apub";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }

src/jobs/apub/forward.rs

@@ -2,11 +2,11 @@ use crate::{
     apub::AcceptedActivities,
     db::Actor,
     error::{Error, ErrorKind},
-    future::BoxFuture,
     jobs::{apub::get_inboxes, DeliverMany, JobState},
 };
 use activitystreams::prelude::*;
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct Forward {
@@ -47,15 +47,14 @@ impl Forward {
     }
 }
 
-impl Job for Forward {
+impl ActixJob for Forward {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::apub::Forward";
     const QUEUE: &'static str = "apub";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }

src/jobs/apub/reject.rs

@@ -2,10 +2,10 @@ use crate::{
     config::UrlKind,
     db::Actor,
     error::Error,
-    future::BoxFuture,
     jobs::{apub::generate_undo_follow, Deliver, JobState},
 };
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct Reject(pub(crate) Actor);
@@ -33,15 +33,14 @@ impl Reject {
     }
 }
 
-impl Job for Reject {
+impl ActixJob for Reject {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::apub::Reject";
     const QUEUE: &'static str = "apub";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }

src/jobs/apub/undo.rs

@@ -3,11 +3,11 @@ use crate::{
     config::UrlKind,
     db::Actor,
     error::Error,
-    future::BoxFuture,
     jobs::{apub::generate_undo_follow, Deliver, JobState},
 };
 use activitystreams::prelude::BaseExt;
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct Undo {
@@ -48,15 +48,14 @@ impl Undo {
     }
 }
 
-impl Job for Undo {
+impl ActixJob for Undo {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::apub::Undo";
     const QUEUE: &'static str = "apub";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }

src/jobs/contact.rs

@@ -1,12 +1,11 @@
 use crate::{
     apub::AcceptedActors,
     error::{Error, ErrorKind},
-    future::BoxFuture,
     jobs::JobState,
-    requests::BreakerStrategy,
 };
 use activitystreams::{iri_string::types::IriString, object::Image, prelude::*};
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct QueryContact {
@@ -33,7 +32,6 @@ impl QueryContact {
 
     async fn perform(self, state: JobState) -> Result<(), Error> {
         let contact_outdated = state
-            .state
             .node_cache
             .is_contact_outdated(self.actor_id.clone())
             .await;
@@ -43,9 +41,8 @@ impl QueryContact {
         }
 
         let contact = match state
-            .state
             .requests
-            .fetch::<AcceptedActors>(&self.contact_id, BreakerStrategy::Allow404AndBelow)
+            .fetch::<AcceptedActors>(&self.contact_id)
             .await
         {
             Ok(contact) => contact,
@@ -60,7 +57,6 @@ impl QueryContact {
             to_contact(contact).ok_or(ErrorKind::Extract("contact"))?;
 
         state
-            .state
             .node_cache
             .set_contact(self.actor_id, username, display_name, url, avatar)
             .await?;
@@ -85,16 +81,15 @@ fn to_contact(contact: AcceptedActors) -> Option<(String, String, IriString, Iri
     Some((username, display_name, url, avatar))
 }
 
-impl Job for QueryContact {
+impl ActixJob for QueryContact {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::QueryContact";
     const QUEUE: &'static str = "maintenance";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }
 

src/jobs/deliver.rs

@@ -1,11 +1,10 @@
 use crate::{
     error::Error,
-    future::BoxFuture,
     jobs::{debug_object, JobState},
-    requests::BreakerStrategy,
 };
 use activitystreams::iri_string::types::IriString;
-use background_jobs::{Backoff, Job};
+use background_jobs::{ActixJob, Backoff};
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct Deliver {
@@ -35,13 +34,8 @@ impl Deliver {
     }
 
     #[tracing::instrument(name = "Deliver", skip(state))]
-    async fn perform(self, state: JobState) -> Result<(), Error> {
+    async fn permform(self, state: JobState) -> Result<(), Error> {
-        if let Err(e) = state
+        if let Err(e) = state.requests.deliver(&self.to, &self.data).await {
-            .state
-            .requests
-            .deliver(&self.to, &self.data, BreakerStrategy::Allow401AndBelow)
-            .await
-        {
             if e.is_breaker() {
                 tracing::debug!("Not trying due to failed breaker");
                 return Ok(());
@@ -56,16 +50,15 @@ impl Deliver {
     }
 }
 
-impl Job for Deliver {
+impl ActixJob for Deliver {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::Deliver";
     const QUEUE: &'static str = "deliver";
     const BACKOFF: Backoff = Backoff::Exponential(8);
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.permform(state).await.map_err(Into::into) })
     }
 }

src/jobs/deliver_many.rs

@@ -1,10 +1,10 @@
 use crate::{
     error::Error,
-    future::BoxFuture,
     jobs::{debug_object, Deliver, JobState},
 };
 use activitystreams::iri_string::types::IriString;
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use futures_util::future::LocalBoxFuture;
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct DeliverMany {
@@ -45,15 +45,14 @@ impl DeliverMany {
     }
 }
 
-impl Job for DeliverMany {
+impl ActixJob for DeliverMany {
     type State = JobState;
-    type Error = Error;
+    type Future = LocalBoxFuture<'static, Result<(), anyhow::Error>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::DeliverMany";
     const QUEUE: &'static str = "deliver";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }

src/jobs/instance.rs

@@ -1,12 +1,11 @@
 use crate::{
     config::UrlKind,
     error::{Error, ErrorKind},
-    future::BoxFuture,
     jobs::{Boolish, JobState},
-    requests::BreakerStrategy,
 };
 use activitystreams::{iri, iri_string::types::IriString};
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct QueryInstance {
@@ -41,23 +40,15 @@ impl QueryInstance {
             InstanceApiType::Mastodon => {
                 let mastodon_instance_uri = iri!(format!("{scheme}://{authority}/api/v1/instance"));
                 state
-                    .state
                     .requests
-                    .fetch_json::<Instance>(
+                    .fetch_json::<Instance>(&mastodon_instance_uri)
-                        &mastodon_instance_uri,
-                        BreakerStrategy::Allow404AndBelow,
-                    )
                     .await
             }
             InstanceApiType::Misskey => {
                 let msky_meta_uri = iri!(format!("{scheme}://{authority}/api/meta"));
                 state
-                    .state
                     .requests
-                    .fetch_json_msky::<MisskeyMeta>(
+                    .fetch_json_msky::<MisskeyMeta>(&msky_meta_uri)
-                        &msky_meta_uri,
-                        BreakerStrategy::Allow404AndBelow,
-                    )
                     .await
                     .map(|res| res.into())
             }
@@ -67,12 +58,10 @@ impl QueryInstance {
     #[tracing::instrument(name = "Query instance", skip(state))]
     async fn perform(self, state: JobState) -> Result<(), Error> {
         let contact_outdated = state
-            .state
             .node_cache
             .is_contact_outdated(self.actor_id.clone())
             .await;
         let instance_outdated = state
-            .state
             .node_cache
             .is_instance_outdated(self.actor_id.clone())
             .await;
@@ -134,7 +123,6 @@ impl QueryInstance {
             let avatar = state.config.generate_url(UrlKind::Media(uuid));
 
             state
-                .state
                 .node_cache
                 .set_contact(
                     self.actor_id.clone(),
@@ -149,7 +137,6 @@ impl QueryInstance {
         let description = ammonia::clean(&description);
 
         state
-            .state
             .node_cache
             .set_instance(
                 self.actor_id,
@@ -165,16 +152,15 @@ impl QueryInstance {
     }
 }
 
-impl Job for QueryInstance {
+impl ActixJob for QueryInstance {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::QueryInstance";
     const QUEUE: &'static str = "maintenance";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }
 

src/jobs/nodeinfo.rs

@@ -1,18 +1,17 @@
 use crate::{
     error::{Error, ErrorKind},
-    future::BoxFuture,
     jobs::{Boolish, JobState, QueryContact},
-    requests::BreakerStrategy,
 };
 use activitystreams::{iri, iri_string::types::IriString, primitives::OneOrMany};
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use std::{fmt::Debug, future::Future, pin::Pin};
 
 #[derive(Clone, serde::Deserialize, serde::Serialize)]
 pub(crate) struct QueryNodeinfo {
     actor_id: IriString,
 }
 
-impl std::fmt::Debug for QueryNodeinfo {
+impl Debug for QueryNodeinfo {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("QueryNodeinfo")
             .field("actor_id", &self.actor_id.to_string())
@@ -28,7 +27,6 @@ impl QueryNodeinfo {
     #[tracing::instrument(name = "Query node info", skip(state))]
     async fn perform(self, state: JobState) -> Result<(), Error> {
         if !state
-            .state
             .node_cache
             .is_nodeinfo_outdated(self.actor_id.clone())
             .await
@@ -44,9 +42,8 @@ impl QueryNodeinfo {
         let well_known_uri = iri!(format!("{scheme}://{authority}/.well-known/nodeinfo"));
 
         let well_known = match state
-            .state
             .requests
-            .fetch_json::<WellKnown>(&well_known_uri, BreakerStrategy::Allow404AndBelow)
+            .fetch_json::<WellKnown>(&well_known_uri)
             .await
         {
             Ok(well_known) => well_known,
@@ -63,12 +60,7 @@ impl QueryNodeinfo {
             return Ok(());
         };
 
-        let nodeinfo = match state
+        let nodeinfo = match state.requests.fetch_json::<Nodeinfo>(&href).await {
-            .state
-            .requests
-            .fetch_json::<Nodeinfo>(&href, BreakerStrategy::Require2XX)
-            .await
-        {
             Ok(nodeinfo) => nodeinfo,
             Err(e) if e.is_breaker() => {
                 tracing::debug!("Not retrying due to failed breaker");
@@ -78,7 +70,6 @@ impl QueryNodeinfo {
         };
 
         state
-            .state
             .node_cache
             .set_info(
                 self.actor_id.clone(),
@@ -92,7 +83,7 @@ impl QueryNodeinfo {
             .metadata
             .and_then(|meta| meta.into_iter().next().and_then(|meta| meta.staff_accounts))
         {
-            if let Some(contact_id) = accounts.first() {
+            if let Some(contact_id) = accounts.get(0) {
                 state
                     .job_server
                     .queue(QueryContact::new(self.actor_id, contact_id.clone()))
@@ -104,16 +95,15 @@ impl QueryNodeinfo {
     }
 }
 
-impl Job for QueryNodeinfo {
+impl ActixJob for QueryNodeinfo {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::QueryNodeinfo";
     const QUEUE: &'static str = "maintenance";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }
 
@@ -156,7 +146,7 @@ struct Link {
 #[serde(untagged)]
 enum MaybeSupported<T> {
     Supported(T),
-    Unsupported(#[allow(unused)] String),
+    Unsupported(String),
 }
 
 impl<T> MaybeSupported<T> {
@@ -165,8 +155,8 @@ impl<T> MaybeSupported<T> {
     }
 }
 
-struct SupportedVersion(#[allow(unused)] String);
+struct SupportedVersion(String);
-struct SupportedNodeinfo(#[allow(unused)] String);
+struct SupportedNodeinfo(String);
 
 static SUPPORTED_VERSIONS: &str = "2.";
 static SUPPORTED_NODEINFO: &str = "http://nodeinfo.diaspora.software/ns/schema/2.";

src/jobs/process_listeners.rs

@@ -1,9 +1,9 @@
 use crate::{
     error::Error,
-    future::BoxFuture,
     jobs::{instance::QueryInstance, nodeinfo::QueryNodeinfo, JobState},
 };
-use background_jobs::Job;
+use background_jobs::ActixJob;
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub(crate) struct Listeners;
@@ -23,15 +23,14 @@ impl Listeners {
     }
 }
 
-impl Job for Listeners {
+impl ActixJob for Listeners {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::Listeners";
     const QUEUE: &'static str = "maintenance";
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }

src/jobs/record_last_online.rs

@@ -1,5 +1,6 @@
-use crate::{error::Error, future::BoxFuture, jobs::JobState};
+use crate::{error::Error, jobs::JobState};
-use background_jobs::{Backoff, Job};
+use background_jobs::{ActixJob, Backoff};
+use std::{future::Future, pin::Pin};
 
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub(crate) struct RecordLastOnline;
@@ -13,16 +14,15 @@ impl RecordLastOnline {
     }
 }
 
-impl Job for RecordLastOnline {
+impl ActixJob for RecordLastOnline {
     type State = JobState;
-    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>>>>;
-    type Future = BoxFuture<'static, Result<(), Self::Error>>;
 
     const NAME: &'static str = "relay::jobs::RecordLastOnline";
     const QUEUE: &'static str = "maintenance";
     const BACKOFF: Backoff = Backoff::Linear(1);
 
     fn run(self, state: Self::State) -> Self::Future {
-        Box::pin(self.perform(state))
+        Box::pin(async move { self.perform(state).await.map_err(Into::into) })
     }
 }

src/main.rs

@@ -1,27 +1,22 @@
 // need this for ructe
 #![allow(clippy::needless_borrow)]
 
-use std::time::Duration;
-
 use activitystreams::iri_string::types::IriString;
+use actix_rt::task::JoinHandle;
 use actix_web::{middleware::Compress, web, App, HttpServer};
 use collector::MemoryCollector;
 #[cfg(feature = "console")]
 use console_subscriber::ConsoleLayer;
-use error::Error;
 use http_signature_normalization_actix::middleware::VerifySignature;
 use metrics_exporter_prometheus::PrometheusBuilder;
 use metrics_util::layers::FanoutBuilder;
-use opentelemetry::{trace::TracerProvider, KeyValue};
+use opentelemetry::{sdk::Resource, KeyValue};
 use opentelemetry_otlp::WithExportConfig;
-use opentelemetry_sdk::Resource;
-use reqwest_middleware::ClientWithMiddleware;
 use rustls::ServerConfig;
-use tokio::task::JoinHandle;
 use tracing_actix_web::TracingLogger;
 use tracing_error::ErrorLayer;
 use tracing_log::LogTracer;
-use tracing_subscriber::{filter::Targets, layer::SubscriberExt, Layer};
+use tracing_subscriber::{filter::Targets, fmt::format::FmtSpan, layer::SubscriberExt, Layer};
 
 mod admin;
 mod apub;
@@ -32,18 +27,12 @@ mod data;
 mod db;
 mod error;
 mod extractors;
-mod future;
-mod http1;
 mod jobs;
 mod middleware;
 mod requests;
 mod routes;
-mod spawner;
-mod stream;
 mod telegram;
 
-use crate::config::UrlKind;
-
 use self::{
     args::Args,
     config::Config,
@@ -52,21 +41,21 @@ use self::{
     jobs::create_workers,
     middleware::{DebugPayload, MyVerify, RelayResolver, Timings},
     routes::{actor, healthz, inbox, index, nodeinfo, nodeinfo_meta, statics},
-    spawner::Spawner,
 };
 
 fn init_subscriber(
     software_name: &'static str,
     opentelemetry_url: Option<&IriString>,
-) -> color_eyre::Result<()> {
+) -> Result<(), anyhow::Error> {
     LogTracer::init()?;
-    color_eyre::install()?;
 
     let targets: Targets = std::env::var("RUST_LOG")
-        .unwrap_or_else(|_| "info".into())
+        .unwrap_or_else(|_| "warn,actix_web=debug,actix_server=debug,tracing_actix_web=info".into())
         .parse()?;
 
-    let format_layer = tracing_subscriber::fmt::layer().with_filter(targets.clone());
+    let format_layer = tracing_subscriber::fmt::layer()
+        .with_span_events(FmtSpan::NEW | FmtSpan::CLOSE)
+        .with_filter(targets.clone());
 
     #[cfg(feature = "console")]
     let console_layer = ConsoleLayer::builder()
@@ -83,21 +72,21 @@ fn init_subscriber(
     let subscriber = subscriber.with(console_layer);
 
     if let Some(url) = opentelemetry_url {
-        let exporter = opentelemetry_otlp::SpanExporter::builder()
+        let tracer =
-            .with_tonic()
+            opentelemetry_otlp::new_pipeline()
-            .with_endpoint(url.as_str())
+                .tracing()
-            .build()?;
+                .with_trace_config(opentelemetry::sdk::trace::config().with_resource(
-
+                    Resource::new(vec![KeyValue::new("service.name", software_name)]),
-        let tracer_provider = opentelemetry_sdk::trace::TracerProvider::builder()
+                ))
-            .with_resource(Resource::new(vec![KeyValue::new(
+                .with_exporter(
-                "service.name",
+                    opentelemetry_otlp::new_exporter()
-                software_name,
+                        .tonic()
-            )]))
+                        .with_endpoint(url.as_str()),
-            .with_batch_exporter(exporter, opentelemetry_sdk::runtime::Tokio)
+                )
-            .build();
+                .install_batch(opentelemetry::runtime::Tokio)?;
 
         let otel_layer = tracing_opentelemetry::layer()
-            .with_tracer(tracer_provider.tracer(software_name))
+            .with_tracer(tracer)
             .with_filter(targets);
 
         let subscriber = subscriber.with(otel_layer);
@@ -109,40 +98,8 @@ fn init_subscriber(
     Ok(())
 }
 
-fn build_client(
+#[actix_rt::main]
-    user_agent: &str,
+async fn main() -> Result<(), anyhow::Error> {
-    timeout_seconds: u64,
-    proxy: Option<(&IriString, Option<(&str, &str)>)>,
-) -> Result<ClientWithMiddleware, Error> {
-    let builder = reqwest::Client::builder().user_agent(user_agent.to_string());
-
-    let builder = if let Some((url, auth)) = proxy {
-        let proxy = reqwest::Proxy::all(url.as_str())?;
-
-        let proxy = if let Some((username, password)) = auth {
-            proxy.basic_auth(username, password)
-        } else {
-            proxy
-        };
-
-        builder.proxy(proxy)
-    } else {
-        builder
-    };
-
-    let client = builder
-        .timeout(Duration::from_secs(timeout_seconds))
-        .build()?;
-
-    let client_with_middleware = reqwest_middleware::ClientBuilder::new(client)
-        .with(reqwest_tracing::TracingMiddleware::default())
-        .build();
-
-    Ok(client_with_middleware)
-}
-
-#[tokio::main]
-async fn main() -> color_eyre::Result<()> {
     dotenv::dotenv().ok();
 
     let config = Config::build()?;
@@ -152,8 +109,7 @@ async fn main() -> color_eyre::Result<()> {
     let args = Args::new();
 
     if args.any() {
-        client_main(config, args).await??;
+        return client_main(config, args).await?;
-        return Ok(());
     }
 
     let collector = MemoryCollector::new();
@@ -163,40 +119,40 @@ async fn main() -> color_eyre::Result<()> {
             .with_http_listener(bind_addr)
             .build()?;
 
-        tokio::spawn(exporter);
+        actix_rt::spawn(exporter);
         let recorder = FanoutBuilder::default()
             .add_recorder(recorder)
             .add_recorder(collector.clone())
             .build();
-        metrics::set_global_recorder(recorder).map_err(|e| color_eyre::eyre::eyre!("{e}"))?;
+        metrics::set_boxed_recorder(Box::new(recorder))?;
     } else {
         collector.install()?;
     }
 
-    tracing::info!("Opening DB");
+    tracing::warn!("Opening DB");
     let db = Db::build(&config)?;
 
-    tracing::info!("Building caches");
+    tracing::warn!("Building caches");
     let actors = ActorCache::new(db.clone());
     let media = MediaCache::new(db.clone());
 
-    server_main(db, actors, media, collector, config).await?;
+    server_main(db, actors, media, collector, config).await??;
 
-    tracing::info!("Application exit");
+    tracing::warn!("Application exit");
 
     Ok(())
 }
 
-fn client_main(config: Config, args: Args) -> JoinHandle<color_eyre::Result<()>> {
+fn client_main(config: Config, args: Args) -> JoinHandle<Result<(), anyhow::Error>> {
-    tokio::spawn(do_client_main(config, args))
+    actix_rt::spawn(do_client_main(config, args))
 }
 
-async fn do_client_main(config: Config, args: Args) -> color_eyre::Result<()> {
+async fn do_client_main(config: Config, args: Args) -> Result<(), anyhow::Error> {
-    let client = build_client(
+    let client = requests::build_client(
         &config.user_agent(),
+        config.client_pool_size(),
         config.client_timeout(),
-        config.proxy_config(),
+    );
-    )?;
 
     if !args.blocks().is_empty() || !args.allowed().is_empty() {
         if args.undo() {
@@ -274,77 +230,49 @@ async fn do_client_main(config: Config, args: Args) -> color_eyre::Result<()> {
     Ok(())
 }
 
-const VERIFY_RATIO: usize = 7;
+fn server_main(
-
-async fn server_main(
     db: Db,
     actors: ActorCache,
     media: MediaCache,
     collector: MemoryCollector,
     config: Config,
-) -> color_eyre::Result<()> {
+) -> JoinHandle<Result<(), anyhow::Error>> {
-    let client = build_client(
+    actix_rt::spawn(do_server_main(db, actors, media, collector, config))
-        &config.user_agent(),
+}
-        config.client_timeout(),
-        config.proxy_config(),
-    )?;
 
-    tracing::info!("Creating state");
+async fn do_server_main(
-
+    db: Db,
-    let (signature_threads, verify_threads) = match config.signature_threads() {
+    actors: ActorCache,
-        0 | 1 => (1, 1),
+    media: MediaCache,
-        n if n <= VERIFY_RATIO => (n, 1),
+    collector: MemoryCollector,
-        n => {
+    config: Config,
-            let verify_threads = (n / VERIFY_RATIO).max(1);
+) -> Result<(), anyhow::Error> {
-            let signature_threads = n.saturating_sub(verify_threads).max(VERIFY_RATIO);
+    tracing::warn!("Creating state");
-
+    let state = State::build(db.clone()).await?;
-            (signature_threads, verify_threads)
-        }
-    };
-
-    let verify_spawner = Spawner::build("verify-cpu", verify_threads.try_into()?)?;
-    let sign_spawner = Spawner::build("sign-cpu", signature_threads.try_into()?)?;
-
-    let key_id = config.generate_url(UrlKind::MainKey).to_string();
-    let state = State::build(db.clone(), key_id, sign_spawner.clone(), client).await?;
 
     if let Some((token, admin_handle)) = config.telegram_info() {
-        tracing::info!("Creating telegram handler");
+        tracing::warn!("Creating telegram handler");
         telegram::start(admin_handle.to_owned(), db.clone(), token);
     }
 
-    let cert_resolver = config
+    let keys = config.open_keys()?;
-        .open_keys()
-        .await?
-        .map(rustls_channel_resolver::channel::<32>);
 
     let bind_address = config.bind_address();
-    let sign_spawner2 = sign_spawner.clone();
-    let verify_spawner2 = verify_spawner.clone();
-    let config2 = config.clone();
-    let job_store = jobs::build_storage();
     let server = HttpServer::new(move || {
-        let job_server = create_workers(
+        let requests = state.requests(&config);
-            job_store.clone(),
+
-            state.clone(),
+        let job_server =
-            actors.clone(),
+            create_workers(state.clone(), actors.clone(), media.clone(), config.clone());
-            media.clone(),
-            config.clone(),
-        )
-        .expect("Failed to create job server");
 
         let app = App::new()
             .app_data(web::Data::new(db.clone()))
             .app_data(web::Data::new(state.clone()))
-            .app_data(web::Data::new(
+            .app_data(web::Data::new(requests.clone()))
-                state.requests.clone().spawner(verify_spawner.clone()),
-            ))
             .app_data(web::Data::new(actors.clone()))
             .app_data(web::Data::new(config.clone()))
             .app_data(web::Data::new(job_server))
             .app_data(web::Data::new(media.clone()))
-            .app_data(web::Data::new(collector.clone()))
+            .app_data(web::Data::new(collector.clone()));
-            .app_data(web::Data::new(verify_spawner.clone()));
 
         let app = if let Some(data) = config.admin_config() {
             app.app_data(data)
@@ -360,15 +288,10 @@ async fn server_main(
             .service(web::resource("/media/{path}").route(web::get().to(routes::media)))
             .service(
                 web::resource("/inbox")
-                    .wrap(config.digest_middleware().spawner(verify_spawner.clone()))
+                    .wrap(config.digest_middleware())
                     .wrap(VerifySignature::new(
-                        MyVerify(
+                        MyVerify(requests, actors.clone(), state.clone()),
-                            state.requests.clone().spawner(verify_spawner.clone()),
+                        Default::default(),
-                            actors.clone(),
-                            state.clone(),
-                            verify_spawner.clone(),
-                        ),
-                        http_signature_normalization_actix::Config::new(),
                     ))
                     .wrap(DebugPayload(config.debug()))
                     .route(web::post().to(inbox)),
@@ -397,42 +320,24 @@ async fn server_main(
             )
     });
 
-    if let Some((cert_tx, cert_rx)) = cert_resolver {
+    if let Some((certs, key)) = keys {
-        let handle = tokio::spawn(async move {
+        tracing::warn!("Binding to {}:{} with TLS", bind_address.0, bind_address.1);
-            let mut interval = tokio::time::interval(Duration::from_secs(30));
-            interval.tick().await;
-
-            loop {
-                interval.tick().await;
-
-                match config2.open_keys().await {
-                    Ok(Some(key)) => cert_tx.update(key),
-                    Ok(None) => tracing::warn!("Missing TLS keys"),
-                    Err(e) => tracing::error!("Failed to read TLS keys {e}"),
-                }
-            }
-        });
-
-        tracing::info!("Binding to {}:{} with TLS", bind_address.0, bind_address.1);
         let server_config = ServerConfig::builder()
+            .with_safe_default_cipher_suites()
+            .with_safe_default_kx_groups()
+            .with_safe_default_protocol_versions()?
             .with_no_client_auth()
-            .with_cert_resolver(cert_rx);
+            .with_single_cert(certs, key)?;
         server
-            .bind_rustls_0_23(bind_address, server_config)?
+            .bind_rustls(bind_address, server_config)?
             .run()
             .await?;
-
-        handle.abort();
-        let _ = handle.await;
     } else {
-        tracing::info!("Binding to {}:{}", bind_address.0, bind_address.1);
+        tracing::warn!("Binding to {}:{}", bind_address.0, bind_address.1);
         server.bind(bind_address)?.run().await?;
     }
 
-    sign_spawner2.close().await;
+    tracing::warn!("Server closed");
-    verify_spawner2.close().await;
-
-    tracing::info!("Server closed");
 
     Ok(())
 }

src/middleware/payload.rs

@@ -4,11 +4,14 @@ use actix_web::{
     web::BytesMut,
     HttpMessage,
 };
+use futures_util::{
+    future::TryFutureExt,
+    stream::{once, TryStreamExt},
+};
 use std::{
     future::{ready, Ready},
     task::{Context, Poll},
 };
-use streem::IntoStreamer;
 
 #[derive(Clone, Debug)]
 pub(crate) struct DebugPayload(pub bool);
@@ -50,23 +53,19 @@ where
 
     fn call(&self, mut req: ServiceRequest) -> Self::Future {
         if self.0 && req.method() == Method::POST {
-            let mut pl = req.take_payload().into_streamer();
+            let pl = req.take_payload();
-
             req.set_payload(Payload::Stream {
-                payload: Box::pin(streem::try_from_fn(|yielder| async move {
+                payload: Box::pin(once(
-                    let mut buf = BytesMut::new();
+                    pl.try_fold(BytesMut::new(), |mut acc, bytes| async {
-
+                        acc.extend(bytes);
-                    while let Some(bytes) = pl.try_next().await? {
+                        Ok(acc)
-                        buf.extend(bytes);
+                    })
-                    }
+                    .map_ok(|bytes| {
-
+                        let bytes = bytes.freeze();
-                    let bytes = buf.freeze();
                         tracing::info!("{}", String::from_utf8_lossy(&bytes));
-
+                        bytes
-                    yielder.yield_ok(bytes).await;
+                    }),
-
+                )),
-                    Ok(())
-                })),
             });
 
             self.1.call(req)

src/middleware/timings.rs

@@ -40,7 +40,7 @@ impl Drop for LogOnDrop {
     fn drop(&mut self) {
         if self.arm {
             let duration = self.begin.elapsed();
-            metrics::histogram!("relay.request.complete", "path" => self.path.clone(), "method" => self.method.clone()).record(duration);
+            metrics::histogram!("relay.request.complete", duration, "path" => self.path.clone(), "method" => self.method.clone());
         }
     }
 }
@@ -80,7 +80,7 @@ where
     fn call(&self, req: ServiceRequest) -> Self::Future {
         let log_on_drop = LogOnDrop {
             begin: Instant::now(),
-            path: format!("{:?}", req.match_pattern()),
+            path: req.path().to_string(),
             method: req.method().to_string(),
             arm: false,
         };

src/middleware/verifier.rs

@@ -2,17 +2,20 @@ use crate::{
     apub::AcceptedActors,
     data::{ActorCache, State},
     error::{Error, ErrorKind},
-    requests::{BreakerStrategy, Requests},
+    requests::Requests,
-    spawner::Spawner,
 };
 use activitystreams::{base::BaseExt, iri, iri_string::types::IriString};
+use actix_web::web;
 use base64::{engine::general_purpose::STANDARD, Engine};
-use http_signature_normalization_actix::{prelude::*, verify::DeprecatedAlgorithm, Spawn};
+use http_signature_normalization_actix::{prelude::*, verify::DeprecatedAlgorithm};
-use rsa::{pkcs1::EncodeRsaPublicKey, pkcs8::DecodePublicKey, RsaPublicKey};
+use rsa::{
+    pkcs1v15::Signature, pkcs1v15::VerifyingKey, pkcs8::DecodePublicKey, sha2::Sha256,
+    signature::Verifier, RsaPublicKey,
+};
 use std::{future::Future, pin::Pin};
 
 #[derive(Clone, Debug)]
-pub(crate) struct MyVerify(pub Requests, pub ActorCache, pub State, pub Spawner);
+pub(crate) struct MyVerify(pub Requests, pub ActorCache, pub State);
 
 impl MyVerify {
     #[tracing::instrument("Verify request", skip(self, signature, signing_string))]
@@ -52,13 +55,7 @@ impl MyVerify {
                 None => (),
             };
 
-            let res = do_verify(
+            let res = do_verify(&actor.public_key, signature.clone(), signing_string.clone()).await;
-                &self.3,
-                &actor.public_key,
-                signature.clone(),
-                signing_string.clone(),
-            )
-            .await;
 
             if let Err(e) = res {
                 if !was_cached {
@@ -70,11 +67,7 @@ impl MyVerify {
 
             actor_id
         } else {
-            match self
+            match self.0.fetch::<PublicKeyResponse>(&public_key_id).await {
-                .0
-                .fetch::<PublicKeyResponse>(&public_key_id, BreakerStrategy::Require2XX)
-                .await
-            {
                 Ok(res) => res.actor_id().ok_or(ErrorKind::MissingId),
                 Err(e) => {
                     if e.is_gone() {
@@ -92,7 +85,7 @@ impl MyVerify {
         // Now we make sure we fetch an updated actor
         let actor = self.1.get_no_cache(&actor_id, &self.0).await?;
 
-        do_verify(&self.3, &actor.public_key, signature, signing_string).await?;
+        do_verify(&actor.public_key, signature, signing_string).await?;
 
         Ok(true)
     }
@@ -123,29 +116,23 @@ impl PublicKeyResponse {
 
 #[tracing::instrument("Verify signature")]
 async fn do_verify(
-    spawner: &Spawner,
     public_key: &str,
     signature: String,
     signing_string: String,
 ) -> Result<(), Error> {
     let public_key = RsaPublicKey::from_public_key_pem(public_key.trim())?;
-    let public_key_der = public_key
-        .to_pkcs1_der()
-        .map_err(|_| ErrorKind::DerEncode)?;
-    let public_key = ring::signature::UnparsedPublicKey::new(
-        &ring::signature::RSA_PKCS1_2048_8192_SHA256,
-        public_key_der,
-    );
 
     let span = tracing::Span::current();
-    spawner
+    web::block(move || {
-        .spawn_blocking(move || {
         span.in_scope(|| {
             let decoded = STANDARD.decode(signature)?;
+            let signature =
+                Signature::try_from(decoded.as_slice()).map_err(ErrorKind::ReadSignature)?;
 
-                public_key
+            let verifying_key = VerifyingKey::<Sha256>::new(public_key);
-                    .verify(signing_string.as_bytes(), decoded.as_slice())
+            verifying_key
-                    .map_err(|_| ErrorKind::VerifySignature)?;
+                .verify(signing_string.as_bytes(), &signature)
+                .map_err(ErrorKind::VerifySignature)?;
 
             Ok(()) as Result<(), Error>
         })

src/middleware/webfinger.rs

@@ -1,10 +1,10 @@
 use crate::{
     config::{Config, UrlKind},
     data::State,
-    future::LocalBoxFuture,
 };
 use actix_web::web::Data;
 use actix_webfinger::{Resolver, Webfinger};
+use futures_util::future::LocalBoxFuture;
 use rsa_magic_public_key::AsMagicPublicKey;
 
 pub(crate) struct RelayResolver;

src/requests.rs

@@ -1,43 +1,31 @@
 use crate::{
     data::LastOnline,
     error::{Error, ErrorKind},
-    spawner::Spawner,
-    stream::{aggregate, limit_stream},
 };
 use activitystreams::iri_string::types::IriString;
 use actix_web::http::header::Date;
+use awc::{error::SendRequestError, Client, ClientResponse, Connector};
 use base64::{engine::general_purpose::STANDARD, Engine};
 use dashmap::DashMap;
-use http_signature_normalization_reqwest::{digest::ring::Sha256, prelude::*};
+use http_signature_normalization_actix::prelude::*;
-use reqwest_middleware::ClientWithMiddleware;
+use rand::thread_rng;
-use ring::{
+use rsa::{
-    rand::SystemRandom,
+    pkcs1v15::SigningKey,
-    signature::{RsaKeyPair, RSA_PKCS1_SHA256},
+    sha2::{Digest, Sha256},
+    signature::{RandomizedSigner, SignatureEncoding},
+    RsaPrivateKey,
 };
-use rsa::{pkcs1::EncodeRsaPrivateKey, RsaPrivateKey};
 use std::{
     sync::Arc,
     time::{Duration, SystemTime},
 };
+use tracing_awc::Tracing;
 
 const ONE_SECOND: u64 = 1;
 const ONE_MINUTE: u64 = 60 * ONE_SECOND;
 const ONE_HOUR: u64 = 60 * ONE_MINUTE;
 const ONE_DAY: u64 = 24 * ONE_HOUR;
 
-// 20 KB
-const JSON_SIZE_LIMIT: usize = 20 * 1024;
-
-#[derive(Debug)]
-pub(crate) enum BreakerStrategy {
-    // Requires a successful response
-    Require2XX,
-    // Allows HTTP 2xx-401
-    Allow401AndBelow,
-    // Allows HTTP 2xx-404
-    Allow404AndBelow,
-}
-
 #[derive(Clone)]
 pub(crate) struct Breakers {
     inner: Arc<DashMap<String, Breaker>>,
@@ -50,7 +38,7 @@ impl std::fmt::Debug for Breakers {
 }
 
 impl Breakers {
-    pub(crate) fn should_try(&self, url: &IriString) -> bool {
+    fn should_try(&self, url: &IriString) -> bool {
         if let Some(authority) = url.authority_str() {
             if let Some(breaker) = self.inner.get(authority) {
                 breaker.should_try()
@@ -152,11 +140,12 @@ impl Default for Breaker {
 
 #[derive(Clone)]
 pub(crate) struct Requests {
-    client: ClientWithMiddleware,
+    pool_size: usize,
+    client: Client,
     key_id: String,
-    private_key: Arc<RsaKeyPair>,
+    user_agent: String,
-    rng: SystemRandom,
+    private_key: RsaPrivateKey,
-    config: Config<Spawner>,
+    config: Config,
     breakers: Breakers,
     last_online: Arc<LastOnline>,
 }
@@ -164,42 +153,58 @@ pub(crate) struct Requests {
 impl std::fmt::Debug for Requests {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Requests")
+            .field("pool_size", &self.pool_size)
             .field("key_id", &self.key_id)
+            .field("user_agent", &self.user_agent)
             .field("config", &self.config)
             .field("breakers", &self.breakers)
             .finish()
     }
 }
 
+thread_local! {
+    static CLIENT: std::cell::OnceCell<Client> = std::cell::OnceCell::new();
+}
+
+pub(crate) fn build_client(user_agent: &str, pool_size: usize, timeout_seconds: u64) -> Client {
+    CLIENT.with(|client| {
+        client
+            .get_or_init(|| {
+                let connector = Connector::new().limit(pool_size);
+
+                Client::builder()
+                    .connector(connector)
+                    .wrap(Tracing)
+                    .add_default_header(("User-Agent", user_agent.to_string()))
+                    .timeout(Duration::from_secs(timeout_seconds))
+                    .finish()
+            })
+            .clone()
+    })
+}
+
 impl Requests {
-    #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
         key_id: String,
         private_key: RsaPrivateKey,
+        user_agent: String,
         breakers: Breakers,
         last_online: Arc<LastOnline>,
-        spawner: Spawner,
+        pool_size: usize,
-        client: ClientWithMiddleware,
+        timeout_seconds: u64,
     ) -> Self {
-        let private_key_der = private_key.to_pkcs1_der().expect("Can encode der");
-        let private_key = ring::signature::RsaKeyPair::from_der(private_key_der.as_bytes())
-            .expect("Key is valid");
         Requests {
-            client,
+            pool_size,
+            client: build_client(&user_agent, pool_size, timeout_seconds),
             key_id,
-            private_key: Arc::new(private_key),
+            user_agent,
-            rng: SystemRandom::new(),
+            private_key,
-            config: Config::new_with_spawner(spawner).mastodon_compat(),
+            config: Config::default().mastodon_compat(),
             breakers,
             last_online,
         }
     }
 
-    pub(crate) fn spawner(mut self, spawner: Spawner) -> Self {
-        self.config = self.config.set_spawner(spawner);
-        self
-    }
-
     pub(crate) fn reset_breaker(&self, iri: &IriString) {
         self.breakers.succeed(iri);
     }
@@ -207,132 +212,97 @@ impl Requests {
     async fn check_response(
         &self,
         parsed_url: &IriString,
-        strategy: BreakerStrategy,
+        res: Result<ClientResponse, SendRequestError>,
-        res: Result<reqwest::Response, reqwest_middleware::Error>,
+    ) -> Result<ClientResponse, Error> {
-    ) -> Result<reqwest::Response, Error> {
         if res.is_err() {
             self.breakers.fail(&parsed_url);
         }
 
-        let res = res?;
+        let mut res =
+            res.map_err(|e| ErrorKind::SendRequest(parsed_url.to_string(), e.to_string()))?;
 
-        let status = res.status();
+        if res.status().is_server_error() {
-
-        let success = match strategy {
-            BreakerStrategy::Require2XX => status.is_success(),
-            BreakerStrategy::Allow401AndBelow => (200..=401).contains(&status.as_u16()),
-            BreakerStrategy::Allow404AndBelow => (200..=404).contains(&status.as_u16()),
-        };
-
-        if !success {
             self.breakers.fail(&parsed_url);
 
-            if let Ok(s) = res.text().await {
+            if let Ok(bytes) = res.body().await {
+                if let Ok(s) = String::from_utf8(bytes.as_ref().to_vec()) {
                     if !s.is_empty() {
-                    tracing::debug!("Response from {parsed_url}, {s}");
+                        tracing::warn!("Response from {parsed_url}, {s}");
+                    }
                 }
             }
 
-            return Err(ErrorKind::Status(
+            return Err(ErrorKind::Status(parsed_url.to_string(), res.status()).into());
-                parsed_url.to_string(),
-                crate::http1::status_to_http02(status),
-            )
-            .into());
         }
 
-        // only actually succeed a breaker on 2xx response
-        if status.is_success() {
         self.last_online.mark_seen(&parsed_url);
         self.breakers.succeed(&parsed_url);
-        }
 
         Ok(res)
     }
 
     #[tracing::instrument(name = "Fetch Json", skip(self), fields(signing_string))]
-    pub(crate) async fn fetch_json<T>(
+    pub(crate) async fn fetch_json<T>(&self, url: &IriString) -> Result<T, Error>
-        &self,
-        url: &IriString,
-        strategy: BreakerStrategy,
-    ) -> Result<T, Error>
     where
         T: serde::de::DeserializeOwned,
     {
-        self.do_fetch(url, "application/json", strategy).await
+        self.do_fetch(url, "application/json").await
     }
 
     #[tracing::instrument(name = "Fetch Json", skip(self), fields(signing_string))]
-    pub(crate) async fn fetch_json_msky<T>(
+    pub(crate) async fn fetch_json_msky<T>(&self, url: &IriString) -> Result<T, Error>
-        &self,
-        url: &IriString,
-        strategy: BreakerStrategy,
-    ) -> Result<T, Error>
     where
         T: serde::de::DeserializeOwned,
     {
-        let stream = self
+        let mut res = self
             .do_deliver(
                 url,
                 &serde_json::json!({}),
                 "application/json",
                 "application/json",
-                strategy,
             )
-            .await?
+            .await?;
-            .bytes_stream();
 
-        let body = aggregate(limit_stream(stream, JSON_SIZE_LIMIT)).await?;
+        let body = res
+            .body()
+            .await
+            .map_err(|e| ErrorKind::ReceiveResponse(url.to_string(), e.to_string()))?;
 
-        Ok(serde_json::from_slice(&body)?)
+        Ok(serde_json::from_slice(body.as_ref())?)
     }
 
     #[tracing::instrument(name = "Fetch Activity+Json", skip(self), fields(signing_string))]
-    pub(crate) async fn fetch<T>(
+    pub(crate) async fn fetch<T>(&self, url: &IriString) -> Result<T, Error>
-        &self,
-        url: &IriString,
-        strategy: BreakerStrategy,
-    ) -> Result<T, Error>
     where
         T: serde::de::DeserializeOwned,
     {
-        self.do_fetch(url, "application/activity+json", strategy)
+        self.do_fetch(url, "application/activity+json").await
-            .await
     }
 
-    async fn do_fetch<T>(
+    async fn do_fetch<T>(&self, url: &IriString, accept: &str) -> Result<T, Error>
-        &self,
-        url: &IriString,
-        accept: &str,
-        strategy: BreakerStrategy,
-    ) -> Result<T, Error>
     where
         T: serde::de::DeserializeOwned,
     {
-        let stream = self
+        let mut res = self.do_fetch_response(url, accept).await?;
-            .do_fetch_response(url, accept, strategy)
-            .await?
-            .bytes_stream();
 
-        let body = aggregate(limit_stream(stream, JSON_SIZE_LIMIT)).await?;
+        let body = res
+            .body()
+            .await
+            .map_err(|e| ErrorKind::ReceiveResponse(url.to_string(), e.to_string()))?;
 
-        Ok(serde_json::from_slice(&body)?)
+        Ok(serde_json::from_slice(body.as_ref())?)
     }
 
     #[tracing::instrument(name = "Fetch response", skip(self), fields(signing_string))]
-    pub(crate) async fn fetch_response(
+    pub(crate) async fn fetch_response(&self, url: &IriString) -> Result<ClientResponse, Error> {
-        &self,
+        self.do_fetch_response(url, "*/*").await
-        url: &IriString,
-        strategy: BreakerStrategy,
-    ) -> Result<reqwest::Response, Error> {
-        self.do_fetch_response(url, "*/*", strategy).await
     }
 
     pub(crate) async fn do_fetch_response(
         &self,
         url: &IriString,
         accept: &str,
-        strategy: BreakerStrategy,
+    ) -> Result<ClientResponse, Error> {
-    ) -> Result<reqwest::Response, Error> {
         if !self.breakers.should_try(url) {
             return Err(ErrorKind::Breaker.into());
         }
@@ -340,20 +310,25 @@ impl Requests {
         let signer = self.signer();
         let span = tracing::Span::current();
 
-        let request = self
+        let res = self
             .client
             .get(url.as_str())
-            .header("Accept", accept)
+            .insert_header(("Accept", accept))
-            .header("Date", Date(SystemTime::now().into()).to_string())
+            .insert_header(Date(SystemTime::now().into()))
-            .signature(&self.config, self.key_id.clone(), move |signing_string| {
+            .no_decompress()
+            .signature(
+                self.config.clone(),
+                self.key_id.clone(),
+                move |signing_string| {
                     span.record("signing_string", signing_string);
                     span.in_scope(|| signer.sign(signing_string))
-            })
+                },
-            .await?;
+            )
+            .await?
+            .send()
+            .await;
 
-        let res = self.client.execute(request).await;
+        let res = self.check_response(url, res).await?;
-
-        let res = self.check_response(url, strategy, res).await?;
 
         Ok(res)
     }
@@ -363,12 +338,7 @@ impl Requests {
         skip_all,
         fields(inbox = inbox.to_string().as_str(), signing_string)
     )]
-    pub(crate) async fn deliver<T>(
+    pub(crate) async fn deliver<T>(&self, inbox: &IriString, item: &T) -> Result<(), Error>
-        &self,
-        inbox: &IriString,
-        item: &T,
-        strategy: BreakerStrategy,
-    ) -> Result<(), Error>
     where
         T: serde::ser::Serialize + std::fmt::Debug,
     {
@@ -377,7 +347,6 @@ impl Requests {
             item,
             "application/activity+json",
             "application/activity+json",
-            strategy,
         )
         .await?;
         Ok(())
@@ -389,8 +358,7 @@ impl Requests {
         item: &T,
         content_type: &str,
         accept: &str,
-        strategy: BreakerStrategy,
+    ) -> Result<ClientResponse, Error>
-    ) -> Result<reqwest::Response, Error>
     where
         T: serde::ser::Serialize + std::fmt::Debug,
     {
@@ -402,12 +370,12 @@ impl Requests {
         let span = tracing::Span::current();
         let item_string = serde_json::to_string(item)?;
 
-        let request = self
+        let (req, body) = self
             .client
             .post(inbox.as_str())
-            .header("Accept", accept)
+            .insert_header(("Accept", accept))
-            .header("Content-Type", content_type)
+            .insert_header(("Content-Type", content_type))
-            .header("Date", Date(SystemTime::now().into()).to_string())
+            .insert_header(Date(SystemTime::now().into()))
             .signature_with_digest(
                 self.config.clone(),
                 self.key_id.clone(),
@@ -418,11 +386,12 @@ impl Requests {
                     span.in_scope(|| signer.sign(signing_string))
                 },
             )
-            .await?;
+            .await?
+            .split();
 
-        let res = self.client.execute(request).await;
+        let res = req.send_body(body).await;
 
-        let res = self.check_response(inbox, strategy, res).await?;
+        let res = self.check_response(inbox, res).await?;
 
         Ok(res)
     }
@@ -430,29 +399,19 @@ impl Requests {
     fn signer(&self) -> Signer {
         Signer {
             private_key: self.private_key.clone(),
-            rng: self.rng.clone(),
         }
     }
 }
 
 struct Signer {
-    private_key: Arc<RsaKeyPair>,
+    private_key: RsaPrivateKey,
-    rng: SystemRandom,
 }
 
 impl Signer {
     fn sign(&self, signing_string: &str) -> Result<String, Error> {
-        let mut signature = vec![0; self.private_key.public().modulus_len()];
+        let signing_key = SigningKey::<Sha256>::new(self.private_key.clone());
-
+        let signature =
-        self.private_key
+            signing_key.try_sign_with_rng(&mut thread_rng(), signing_string.as_bytes())?;
-            .sign(
+        Ok(STANDARD.encode(signature.to_bytes().as_ref()))
-                &RSA_PKCS1_SHA256,
-                &self.rng,
-                signing_string.as_bytes(),
-                &mut signature,
-            )
-            .map_err(|_| ErrorKind::SignRequest)?;
-
-        Ok(STANDARD.encode(&signature))
     }
 }

src/routes/index.rs

@@ -14,11 +14,10 @@ const MINIFY_CONFIG: minify_html::Cfg = minify_html::Cfg {
     keep_html_and_head_opening_tags: false,
     keep_spaces_between_attributes: true,
     keep_comments: false,
-    keep_input_type_text_attr: true,
-    keep_ssi_comments: false,
-    preserve_brace_template_syntax: false,
-    preserve_chevron_percent_template_syntax: false,
     minify_css: true,
+    minify_css_level_1: true,
+    minify_css_level_2: false,
+    minify_css_level_3: false,
     minify_js: true,
     remove_bangs: true,
     remove_processing_instructions: true,
@@ -37,16 +36,12 @@ pub(crate) async fn route(
     state: web::Data<State>,
     config: web::Data<Config>,
 ) -> Result<HttpResponse, Error> {
-    let all_nodes = state.node_cache.nodes().await?;
+    let all_nodes = state.node_cache().nodes().await?;
 
     let mut nodes = Vec::new();
     let mut local = Vec::new();
 
     for node in all_nodes {
-        if !state.is_connected(&node.base) {
-            continue;
-        }
-
         if node
             .base
             .authority_str()

src/routes/media.rs

@@ -1,15 +1,7 @@
-use crate::{
+use crate::{data::MediaCache, error::Error, requests::Requests};
-    data::MediaCache,
-    error::Error,
-    requests::{BreakerStrategy, Requests},
-    stream::limit_stream,
-};
 use actix_web::{body::BodyStream, web, HttpResponse};
 use uuid::Uuid;
 
-// 16 MB
-const IMAGE_SIZE_LIMIT: usize = 16 * 1024 * 1024;
-
 #[tracing::instrument(name = "Media", skip(media, requests))]
 pub(crate) async fn route(
     media: web::Data<MediaCache>,
@@ -19,23 +11,15 @@ pub(crate) async fn route(
     let uuid = uuid.into_inner();
 
     if let Some(url) = media.get_url(uuid).await? {
-        let res = requests
+        let res = requests.fetch_response(&url).await?;
-            .fetch_response(&url, BreakerStrategy::Allow404AndBelow)
-            .await?;
 
-        let mut response = HttpResponse::build(crate::http1::status_to_http02(res.status()));
+        let mut response = HttpResponse::build(res.status());
 
         for (name, value) in res.headers().iter().filter(|(h, _)| *h != "connection") {
-            response.insert_header((
+            response.insert_header((name.clone(), value.clone()));
-                crate::http1::name_to_http02(name),
-                crate::http1::value_to_http02(value),
-            ));
         }
 
-        return Ok(response.body(BodyStream::new(limit_stream(
+        return Ok(response.body(BodyStream::new(res)));
-            res.bytes_stream(),
-            IMAGE_SIZE_LIMIT,
-        ))));
     }
 
     Ok(HttpResponse::NotFound().finish())

src/spawner.rs

@@ -1,92 +0,0 @@
-use async_cpupool::CpuPool;
-use http_signature_normalization_actix::{Canceled, Spawn};
-use std::time::Duration;
-
-#[derive(Clone)]
-pub(crate) struct Spawner {
-    pool: CpuPool,
-}
-
-impl Spawner {
-    pub(crate) fn build(name: &'static str, threads: u16) -> color_eyre::Result<Self> {
-        let pool = CpuPool::configure()
-            .name(name)
-            .max_threads(threads)
-            .build()?;
-
-        Ok(Spawner { pool })
-    }
-
-    pub(crate) async fn close(self) {
-        self.pool.close().await;
-    }
-}
-
-impl std::fmt::Debug for Spawner {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Spawner").finish()
-    }
-}
-
-async fn timer<Fut>(fut: Fut) -> Fut::Output
-where
-    Fut: std::future::Future,
-{
-    let id = uuid::Uuid::new_v4();
-
-    metrics::counter!("relay.spawner.wait-timer.start").increment(1);
-
-    let mut interval = tokio::time::interval(Duration::from_secs(5));
-
-    // pass the first tick (instant)
-    interval.tick().await;
-
-    let mut fut = std::pin::pin!(fut);
-
-    let mut counter = 0;
-    loop {
-        tokio::select! {
-            out = &mut fut => {
-                metrics::counter!("relay.spawner.wait-timer.end").increment(1);
-                return out;
-            }
-            _ = interval.tick() => {
-                counter += 1;
-                metrics::counter!("relay.spawner.wait-timer.pending").increment(1);
-                tracing::warn!("Blocking operation {id} is taking a long time, {} seconds", counter * 5);
-            }
-        }
-    }
-}
-
-impl Spawn for Spawner {
-    type Future<T> = std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, Canceled>>>>;
-
-    fn spawn_blocking<Func, Out>(&self, func: Func) -> Self::Future<Out>
-    where
-        Func: FnOnce() -> Out + Send + 'static,
-        Out: Send + 'static,
-    {
-        let pool = self.pool.clone();
-
-        Box::pin(async move { timer(pool.spawn(func)).await.map_err(|_| Canceled) })
-    }
-}
-
-impl http_signature_normalization_reqwest::Spawn for Spawner {
-    type Future<T> = std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, http_signature_normalization_reqwest::Canceled>> + Send>> where T: Send;
-
-    fn spawn_blocking<Func, Out>(&self, func: Func) -> Self::Future<Out>
-    where
-        Func: FnOnce() -> Out + Send + 'static,
-        Out: Send + 'static,
-    {
-        let pool = self.pool.clone();
-
-        Box::pin(async move {
-            timer(pool.spawn(func))
-                .await
-                .map_err(|_| http_signature_normalization_reqwest::Canceled)
-        })
-    }
-}

src/stream.rs

@@ -1,59 +0,0 @@
-use crate::error::{Error, ErrorKind};
-use actix_web::web::{Bytes, BytesMut};
-use futures_core::Stream;
-use streem::IntoStreamer;
-
-pub(crate) fn limit_stream<'a, S>(
-    input: S,
-    limit: usize,
-) -> impl Stream<Item = Result<Bytes, Error>> + Send + 'a
-where
-    S: Stream<Item = reqwest::Result<Bytes>> + Send + 'a,
-{
-    streem::try_from_fn(move |yielder| async move {
-        let stream = std::pin::pin!(input);
-        let mut stream = stream.into_streamer();
-
-        let mut count = 0;
-
-        while let Some(bytes) = stream.try_next().await? {
-            count += bytes.len();
-
-            if count > limit {
-                return Err(ErrorKind::BodyTooLarge.into());
-            }
-
-            yielder.yield_ok(bytes).await;
-        }
-
-        Ok(())
-    })
-}
-
-pub(crate) async fn aggregate<S>(input: S) -> Result<Bytes, Error>
-where
-    S: Stream<Item = Result<Bytes, Error>>,
-{
-    let stream = std::pin::pin!(input);
-    let mut streamer = stream.into_streamer();
-
-    let mut buf = Vec::new();
-
-    while let Some(bytes) = streamer.try_next().await? {
-        buf.push(bytes);
-    }
-
-    if buf.len() == 1 {
-        return Ok(buf.pop().expect("buf has exactly one element"));
-    }
-
-    let total_size: usize = buf.iter().map(|b| b.len()).sum();
-
-    let mut bytes_mut = BytesMut::with_capacity(total_size);
-
-    for bytes in &buf {
-        bytes_mut.extend_from_slice(&bytes);
-    }
-
-    Ok(bytes_mut.freeze())
-}

src/telegram.rs

@@ -46,7 +46,7 @@ pub(crate) fn start(admin_handle: String, db: Db, token: &str) {
     let bot = Bot::new(token);
     let admin_handle = Arc::new(admin_handle);
 
-    tokio::spawn(async move {
+    actix_rt::spawn(async move {
         let command_handler = teloxide::filter_command::<Command, _>().endpoint(
             move |bot: Bot, msg: Message, cmd: Command| {
                 let admin_handle = admin_handle.clone();
@@ -75,8 +75,7 @@ pub(crate) fn start(admin_handle: String, db: Db, token: &str) {
 
 fn is_admin(admin_handle: &str, message: &Message) -> bool {
     message
-        .from
+        .from()
-        .as_ref()
         .and_then(|user| user.username.as_deref())
         .map(|username| username == admin_handle)
         .unwrap_or(false)