mirror of
https://git.asonix.dog/asonix/relay.git
synced 2024-09-28 14:51:58 +00:00
Compare commits
49 commits
Author | SHA1 | Date | |
---|---|---|---|
|
752067ffb7 | ||
|
b308e080af | ||
|
6ab37dc06f | ||
|
a23b30cc91 | ||
|
1b58a50d44 | ||
|
308a945283 | ||
|
86cab5d2d9 | ||
|
a70e75665b | ||
|
f1792c8eb3 | ||
|
d918ef1495 | ||
|
2870789e1f | ||
|
cda92e7523 | ||
|
43b03a176c | ||
|
a465d1ae5b | ||
|
4fa7674a35 | ||
|
8c14d613f7 | ||
|
aff2431681 | ||
|
5aa97212b3 | ||
|
97567cf598 | ||
|
4c663f399e | ||
|
8a3256f52a | ||
|
13a2653fe8 | ||
|
8dd9a86d22 | ||
|
5c0c0591dd | ||
|
04ca4e5401 | ||
|
1de1d76506 | ||
|
dd9225bb89 | ||
|
b577730836 | ||
|
21883c168b | ||
|
76a0c79369 | ||
|
6444782db9 | ||
|
14aea3256d | ||
|
f4f2aa2025 | ||
|
615271fe80 | ||
|
4aed601664 | ||
|
bf21f05aca | ||
|
e69f6c6edb | ||
|
1e05eb4fe4 | ||
|
7f09ac3edd | ||
|
4788ad332a | ||
|
1fd82915d3 | ||
|
0472082a97 | ||
|
c8250acce7 | ||
|
b074759eb4 | ||
|
ed399f1531 | ||
|
7e39acdcb0 | ||
|
894d096622 | ||
|
05e31254ba | ||
|
086ca9fbf2 |
41 changed files with 1926 additions and 1544 deletions
421
.drone.yml
421
.drone.yml
|
@ -1,421 +0,0 @@
|
|||
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
|
61
.forgejo/workflows/check.yaml
Normal file
61
.forgejo/workflows/check.yaml
Normal file
|
@ -0,0 +1,61 @@
|
|||
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 }}
|
226
.forgejo/workflows/publish.yaml
Normal file
226
.forgejo/workflows/publish.yaml
Normal file
|
@ -0,0 +1,226 @@
|
|||
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 }}
|
1909
Cargo.lock
generated
1909
Cargo.lock
generated
File diff suppressed because it is too large
Load diff
62
Cargo.toml
62
Cargo.toml
|
@ -1,11 +1,11 @@
|
|||
[package]
|
||||
name = "ap-relay"
|
||||
description = "A simple activitypub relay"
|
||||
version = "0.3.106"
|
||||
version = "0.3.115"
|
||||
authors = ["asonix <asonix@asonix.dog>"]
|
||||
license = "AGPL-3.0"
|
||||
readme = "README.md"
|
||||
repository = "https://git.asonix.dog/asonix/ap-relay"
|
||||
repository = "https://git.asonix.dog/asonix/relay"
|
||||
keywords = ["activitypub", "relay"]
|
||||
edition = "2021"
|
||||
build = "src/build.rs"
|
||||
|
@ -14,6 +14,9 @@ build = "src/build.rs"
|
|||
name = "relay"
|
||||
path = "src/main.rs"
|
||||
|
||||
[profile.release]
|
||||
strip = true
|
||||
|
||||
[features]
|
||||
console = ["dep:console-subscriber"]
|
||||
default = []
|
||||
|
@ -21,47 +24,49 @@ default = []
|
|||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
anyhow = "1.0"
|
||||
actix-web = { version = "4.4.0", default-features = false, features = ["compress-brotli", "compress-gzip", "rustls-0_21"] }
|
||||
actix-web = { version = "4.4.0", default-features = false, features = ["compress-brotli", "compress-gzip", "rustls-0_23"] }
|
||||
actix-webfinger = { version = "0.5.0", default-features = false }
|
||||
activitystreams = "0.7.0-alpha.25"
|
||||
activitystreams-ext = "0.1.0-alpha.3"
|
||||
ammonia = "3.1.0"
|
||||
async-cpupool = "0.2.0"
|
||||
ammonia = "4.0.0"
|
||||
async-cpupool = "0.3.0"
|
||||
bcrypt = "0.15"
|
||||
base64 = "0.21"
|
||||
base64 = "0.22"
|
||||
clap = { version = "4.0.0", features = ["derive"] }
|
||||
config = "0.13.0"
|
||||
console-subscriber = { version = "0.2", optional = true }
|
||||
dashmap = "5.1.0"
|
||||
color-eyre = "0.6.2"
|
||||
config = { version = "0.14.0", default-features = false, features = ["toml", "json", "yaml"] }
|
||||
console-subscriber = { version = "0.4", optional = true }
|
||||
dashmap = "6.0.1"
|
||||
dotenv = "0.15.0"
|
||||
flume = "0.11.0"
|
||||
futures-core = "0.3.30"
|
||||
lru = "0.12.0"
|
||||
metrics = "0.22.0"
|
||||
metrics-exporter-prometheus = { version = "0.13.0", default-features = false, features = [
|
||||
metrics = "0.23.0"
|
||||
metrics-exporter-prometheus = { version = "0.15.0", default-features = false, features = [
|
||||
"http-listener",
|
||||
] }
|
||||
metrics-util = "0.16.0"
|
||||
metrics-util = "0.17.0"
|
||||
mime = "0.3.16"
|
||||
minify-html = "0.15.0"
|
||||
opentelemetry = "0.21"
|
||||
opentelemetry_sdk = { version = "0.21", features = ["rt-tokio"] }
|
||||
opentelemetry-otlp = "0.14"
|
||||
opentelemetry = "0.24"
|
||||
opentelemetry_sdk = { version = "0.24", features = ["rt-tokio"] }
|
||||
opentelemetry-otlp = "0.17"
|
||||
pin-project-lite = "0.2.9"
|
||||
# pinned to metrics-util
|
||||
quanta = "0.12.0"
|
||||
rand = "0.8"
|
||||
reqwest = { version = "0.11", default-features = false, features = ["rustls-tls", "stream"]}
|
||||
reqwest-middleware = "0.2"
|
||||
reqwest-tracing = "0.4.5"
|
||||
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream"]}
|
||||
reqwest-middleware = { version = "0.3", default-features = false, features = ["json"] }
|
||||
reqwest-tracing = "0.5.0"
|
||||
ring = "0.17.5"
|
||||
rsa = { version = "0.9" }
|
||||
rsa = "0.9"
|
||||
rsa-magic-public-key = "0.8.0"
|
||||
rustls = "0.21.0"
|
||||
rustls-pemfile = "1.0.1"
|
||||
rustls = { version = "0.23.0", default-features = false, features = ["ring", "logging", "std", "tls12"] }
|
||||
rustls-channel-resolver = "0.3.0"
|
||||
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 = [
|
||||
"ctrlc_handler",
|
||||
"macros",
|
||||
|
@ -72,7 +77,7 @@ time = { version = "0.3.17", features = ["serde"] }
|
|||
tracing = "0.1"
|
||||
tracing-error = "0.2"
|
||||
tracing-log = "0.2"
|
||||
tracing-opentelemetry = "0.22"
|
||||
tracing-opentelemetry = "0.25"
|
||||
tracing-subscriber = { version = "0.3", features = [
|
||||
"ansi",
|
||||
"env-filter",
|
||||
|
@ -80,20 +85,19 @@ tracing-subscriber = { version = "0.3", features = [
|
|||
] }
|
||||
tokio = { version = "1", features = ["full", "tracing"] }
|
||||
uuid = { version = "1", features = ["v4", "serde"] }
|
||||
streem = "0.2.0"
|
||||
|
||||
[dependencies.background-jobs]
|
||||
version = "0.17.0"
|
||||
version = "0.19.0"
|
||||
default-features = false
|
||||
features = ["error-logging", "metrics", "tokio"]
|
||||
|
||||
[dependencies.http-signature-normalization-actix]
|
||||
version = "0.11.0"
|
||||
version = "0.11.1"
|
||||
default-features = false
|
||||
features = ["server", "ring"]
|
||||
|
||||
[dependencies.http-signature-normalization-reqwest]
|
||||
version = "0.11.0"
|
||||
version = "0.12.0"
|
||||
default-features = false
|
||||
features = ["middleware", "ring"]
|
||||
|
||||
|
@ -101,7 +105,7 @@ features = ["middleware", "ring"]
|
|||
version = "0.7.9"
|
||||
|
||||
[build-dependencies]
|
||||
anyhow = "1.0"
|
||||
color-eyre = "0.6.2"
|
||||
dotenv = "0.15.0"
|
||||
ructe = { version = "0.17.0", features = ["sass", "mime03"] }
|
||||
toml = "0.8.0"
|
||||
|
|
|
@ -1,11 +0,0 @@
|
|||
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"]
|
|
@ -1,25 +0,0 @@
|
|||
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
|
24
docker/forgejo/Dockerfile
Normal file
24
docker/forgejo/Dockerfile
Normal file
|
@ -0,0 +1,24 @@
|
|||
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"]
|
|
@ -2,7 +2,7 @@ version: '3.3'
|
|||
|
||||
services:
|
||||
relay:
|
||||
image: asonix/relay:0.3.85
|
||||
image: asonix/relay:0.3.115
|
||||
ports:
|
||||
- "8079:8079"
|
||||
restart: always
|
||||
|
|
12
flake.lock
12
flake.lock
|
@ -5,11 +5,11 @@
|
|||
"systems": "systems"
|
||||
},
|
||||
"locked": {
|
||||
"lastModified": 1701680307,
|
||||
"narHash": "sha256-kAuep2h5ajznlPMD9rnQyffWG8EM/C73lejGofXvdM8=",
|
||||
"lastModified": 1710146030,
|
||||
"narHash": "sha256-SZ5L6eA7HJ/nmkzGG7/ISclqe6oZdOZTNoesiInkXPQ=",
|
||||
"owner": "numtide",
|
||||
"repo": "flake-utils",
|
||||
"rev": "4022d587cbbfd70fe950c1e2083a02621806a725",
|
||||
"rev": "b1d9ab70662946ef0850d488da1c9019f3a9752a",
|
||||
"type": "github"
|
||||
},
|
||||
"original": {
|
||||
|
@ -20,11 +20,11 @@
|
|||
},
|
||||
"nixpkgs": {
|
||||
"locked": {
|
||||
"lastModified": 1705133751,
|
||||
"narHash": "sha256-rCIsyE80jgiOU78gCWN3A0wE0tR2GI5nH6MlS+HaaSQ=",
|
||||
"lastModified": 1719075281,
|
||||
"narHash": "sha256-CyyxvOwFf12I91PBWz43iGT1kjsf5oi6ax7CrvaMyAo=",
|
||||
"owner": "NixOS",
|
||||
"repo": "nixpkgs",
|
||||
"rev": "9b19f5e77dd906cb52dade0b7bd280339d2a1f3d",
|
||||
"rev": "a71e967ef3694799d0c418c98332f7ff4cc5f6af",
|
||||
"type": "github"
|
||||
},
|
||||
"original": {
|
||||
|
|
|
@ -5,7 +5,7 @@
|
|||
|
||||
rustPlatform.buildRustPackage {
|
||||
pname = "relay";
|
||||
version = "0.3.106";
|
||||
version = "0.3.115";
|
||||
src = ./.;
|
||||
cargoLock.lockFile = ./Cargo.lock;
|
||||
|
||||
|
|
|
@ -5,7 +5,6 @@ use crate::{
|
|||
error::{Error, ErrorKind},
|
||||
extractors::XApiToken,
|
||||
};
|
||||
use actix_web::http::header::Header;
|
||||
use reqwest_middleware::ClientWithMiddleware;
|
||||
use serde::de::DeserializeOwned;
|
||||
|
||||
|
@ -87,13 +86,17 @@ async fn get_results<T: DeserializeOwned>(
|
|||
|
||||
let res = client
|
||||
.get(iri.as_str())
|
||||
.header(XApiToken::name(), x_api_token.to_string())
|
||||
.header(XApiToken::http1_name(), x_api_token.to_string())
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| ErrorKind::SendRequest(iri.to_string(), e.to_string()))?;
|
||||
|
||||
if !res.status().is_success() {
|
||||
return Err(ErrorKind::Status(iri.to_string(), res.status()).into());
|
||||
return Err(ErrorKind::Status(
|
||||
iri.to_string(),
|
||||
crate::http1::status_to_http02(res.status()),
|
||||
)
|
||||
.into());
|
||||
}
|
||||
|
||||
let t = res
|
||||
|
@ -116,7 +119,7 @@ async fn post_domains(
|
|||
|
||||
let res = client
|
||||
.post(iri.as_str())
|
||||
.header(XApiToken::name(), x_api_token.to_string())
|
||||
.header(XApiToken::http1_name(), x_api_token.to_string())
|
||||
.json(&Domains { domains })
|
||||
.send()
|
||||
.await
|
||||
|
|
|
@ -21,7 +21,7 @@ fn git_info() {
|
|||
}
|
||||
}
|
||||
|
||||
fn version_info() -> Result<(), anyhow::Error> {
|
||||
fn version_info() -> color_eyre::Result<()> {
|
||||
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() -> Result<(), anyhow::Error> {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
fn main() -> Result<(), anyhow::Error> {
|
||||
fn main() -> color_eyre::Result<()> {
|
||||
dotenv::dotenv().ok();
|
||||
|
||||
git_info();
|
||||
|
|
|
@ -15,6 +15,10 @@ 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)]
|
||||
|
@ -299,7 +303,14 @@ 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();
|
||||
|
@ -358,6 +369,7 @@ 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>> {
|
||||
|
|
|
@ -12,9 +12,8 @@ use activitystreams::{
|
|||
};
|
||||
use config::Environment;
|
||||
use http_signature_normalization_actix::{digest::ring::Sha256, prelude::VerifyDigest};
|
||||
use rustls::{Certificate, PrivateKey};
|
||||
use rustls::sign::CertifiedKey;
|
||||
use std::{
|
||||
io::BufReader,
|
||||
net::{IpAddr, SocketAddr},
|
||||
path::PathBuf,
|
||||
};
|
||||
|
@ -312,43 +311,34 @@ impl Config {
|
|||
Some((config.addr, config.port).into())
|
||||
}
|
||||
|
||||
pub(crate) fn open_keys(&self) -> Result<Option<(Vec<Certificate>, PrivateKey)>, Error> {
|
||||
pub(crate) async fn open_keys(&self) -> Result<Option<CertifiedKey>, Error> {
|
||||
let tls = if let Some(tls) = &self.tls {
|
||||
tls
|
||||
} else {
|
||||
tracing::warn!("No TLS config present");
|
||||
tracing::info!("No TLS config present");
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let mut certs_reader = BufReader::new(std::fs::File::open(&tls.cert)?);
|
||||
let certs = rustls_pemfile::certs(&mut certs_reader)?;
|
||||
let certs_bytes = tokio::fs::read(&tls.cert).await?;
|
||||
let certs =
|
||||
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 mut key_reader = BufReader::new(std::fs::File::open(&tls.key)?);
|
||||
let key = rustls_pemfile::read_one(&mut key_reader)?;
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
let key_bytes = tokio::fs::read(&tls.key).await?;
|
||||
let key = if let Some(key) = rustls_pemfile::private_key(&mut key_bytes.as_slice())? {
|
||||
key
|
||||
} else {
|
||||
tracing::warn!("Failed to read private key");
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
Ok(Some((certs, key)))
|
||||
let key = rustls::crypto::ring::sign::any_supported_type(&key)?;
|
||||
|
||||
Ok(Some(CertifiedKey::new(certs, key)))
|
||||
}
|
||||
|
||||
pub(crate) fn footer_blurb(&self) -> Option<crate::templates::Html<String>> {
|
||||
|
|
|
@ -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() {
|
||||
self.domains
|
||||
.lock()
|
||||
.unwrap()
|
||||
.insert(authority.to_string(), OffsetDateTime::now_utc());
|
||||
let mut guard = self.domains.lock().unwrap();
|
||||
guard.insert(authority.to_string(), OffsetDateTime::now_utc());
|
||||
metrics::gauge!("relay.last-online.size",)
|
||||
.set(crate::collector::recordable(guard.len()));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -73,7 +73,9 @@ impl State {
|
|||
}
|
||||
|
||||
pub(crate) fn cache(&self, object_id: IriString, actor_id: IriString) {
|
||||
self.object_cache.write().unwrap().put(object_id, actor_id);
|
||||
let mut guard = self.object_cache.write().unwrap();
|
||||
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 {
|
||||
|
|
142
src/db.rs
142
src/db.rs
|
@ -7,7 +7,7 @@ use rsa::{
|
|||
pkcs8::{DecodePrivateKey, EncodePrivateKey},
|
||||
RsaPrivateKey,
|
||||
};
|
||||
use sled::{Batch, Tree};
|
||||
use sled::{transaction::TransactionError, Batch, Transactional, Tree};
|
||||
use std::{
|
||||
collections::{BTreeMap, HashMap},
|
||||
sync::{
|
||||
|
@ -283,10 +283,15 @@ 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| {
|
||||
inner
|
||||
let res = 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?;
|
||||
|
@ -349,6 +354,9 @@ 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
|
||||
|
@ -383,6 +391,9 @@ 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
|
||||
|
@ -417,6 +428,9 @@ 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
|
||||
|
@ -447,6 +461,12 @@ 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
|
||||
|
@ -538,6 +558,14 @@ 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
|
||||
|
@ -550,6 +578,10 @@ 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
|
||||
|
@ -562,6 +594,10 @@ 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
|
||||
|
@ -569,30 +605,64 @@ impl Db {
|
|||
|
||||
pub(crate) async fn add_blocks(&self, domains: Vec<String>) -> Result<(), Error> {
|
||||
self.unblock(move |inner| {
|
||||
for connected in inner.connected_by_domain(&domains) {
|
||||
inner
|
||||
.connected_actor_ids
|
||||
.remove(connected.as_str().as_bytes())?;
|
||||
}
|
||||
let connected_by_domain = inner.connected_by_domain(&domains).collect::<Vec<_>>();
|
||||
|
||||
for authority in &domains {
|
||||
inner
|
||||
.blocked_domains
|
||||
.insert(domain_key(authority), authority.as_bytes())?;
|
||||
inner.allowed_domains.remove(domain_key(authority))?;
|
||||
}
|
||||
let res = (
|
||||
&inner.connected_actor_ids,
|
||||
&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();
|
||||
|
||||
Ok(())
|
||||
for connected in &connected_by_domain {
|
||||
connected_batch.remove(connected.as_str().as_bytes());
|
||||
}
|
||||
|
||||
for authority in &domains {
|
||||
blocked_batch
|
||||
.insert(domain_key(authority).as_bytes(), authority.as_bytes());
|
||||
allowed_batch.remove(domain_key(authority).as_bytes());
|
||||
}
|
||||
|
||||
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 {
|
||||
inner.blocked_domains.remove(domain_key(authority))?;
|
||||
blocked_batch.remove(domain_key(authority).as_bytes());
|
||||
}
|
||||
|
||||
inner.blocked_domains.apply_batch(blocked_batch)?;
|
||||
|
||||
metrics::gauge!("relay.db.blocked-domains.size")
|
||||
.set(crate::collector::recordable(inner.blocked_domains.len()));
|
||||
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
|
@ -600,12 +670,17 @@ 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 {
|
||||
inner
|
||||
.allowed_domains
|
||||
.insert(domain_key(authority), authority.as_bytes())?;
|
||||
allowed_batch.insert(domain_key(authority).as_bytes(), 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
|
||||
|
@ -614,17 +689,32 @@ impl Db {
|
|||
pub(crate) async fn remove_allows(&self, domains: Vec<String>) -> Result<(), Error> {
|
||||
self.unblock(move |inner| {
|
||||
if inner.restricted_mode {
|
||||
for connected in inner.connected_by_domain(&domains) {
|
||||
inner
|
||||
.connected_actor_ids
|
||||
.remove(connected.as_str().as_bytes())?;
|
||||
let connected_by_domain = inner.connected_by_domain(&domains).collect::<Vec<_>>();
|
||||
|
||||
let mut connected_batch = Batch::default();
|
||||
|
||||
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 {
|
||||
inner.allowed_domains.remove(domain_key(authority))?;
|
||||
allowed_batch.remove(domain_key(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
|
||||
|
@ -665,6 +755,10 @@ 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
|
||||
|
|
107
src/error.rs
107
src/error.rs
|
@ -4,54 +4,82 @@ use actix_web::{
|
|||
http::StatusCode,
|
||||
HttpResponse,
|
||||
};
|
||||
use background_jobs::BoxError;
|
||||
use color_eyre::eyre::Error as Report;
|
||||
use http_signature_normalization_reqwest::SignError;
|
||||
use std::{convert::Infallible, fmt::Debug, io};
|
||||
use std::{convert::Infallible, io, sync::Arc};
|
||||
use tokio::task::JoinError;
|
||||
use tracing_error::SpanTrace;
|
||||
|
||||
#[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 {
|
||||
context: String,
|
||||
kind: ErrorKind,
|
||||
kind: ArcKind,
|
||||
display: Box<str>,
|
||||
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 {
|
||||
writeln!(f, "{:?}", self.kind)
|
||||
f.write_str(&self.debug)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Display for Error {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
writeln!(f, "{}", self.kind)?;
|
||||
std::fmt::Display::fmt(&self.context, f)
|
||||
f.write_str(&self.display)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for Error {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
self.kind.source()
|
||||
self.kind().source()
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -60,60 +88,77 @@ 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 {
|
||||
context: SpanTrace::capture().to_string(),
|
||||
kind: error.into(),
|
||||
kind,
|
||||
display: Box::from(display),
|
||||
debug: Box::from(debug),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub(crate) enum ErrorKind {
|
||||
#[error("Error queueing job, {0}")]
|
||||
Queue(anyhow::Error),
|
||||
#[error("Error in extractor")]
|
||||
Extractor(#[from] crate::extractors::ErrorKind),
|
||||
|
||||
#[error("Error in configuration, {0}")]
|
||||
#[error("Error queueing job")]
|
||||
Queue(#[from] BoxError),
|
||||
|
||||
#[error("Error in configuration")]
|
||||
Config(#[from] config::ConfigError),
|
||||
|
||||
#[error("Couldn't parse key, {0}")]
|
||||
#[error("Couldn't parse key")]
|
||||
Pkcs8(#[from] rsa::pkcs8::Error),
|
||||
|
||||
#[error("Couldn't encode public key, {0}")]
|
||||
#[error("Couldn't encode public key")]
|
||||
Spki(#[from] rsa::pkcs8::spki::Error),
|
||||
|
||||
#[error("Couldn't sign request")]
|
||||
SignRequest,
|
||||
|
||||
#[error("Response body from server exceeded limits")]
|
||||
BodyTooLarge,
|
||||
|
||||
#[error("Couldn't make request")]
|
||||
Reqwest(#[from] reqwest::Error),
|
||||
|
||||
#[error("Couldn't build client")]
|
||||
#[error("Couldn't make request")]
|
||||
ReqwestMiddleware(#[from] reqwest_middleware::Error),
|
||||
|
||||
#[error("Couldn't parse IRI, {0}")]
|
||||
#[error("Couldn't parse IRI")]
|
||||
ParseIri(#[from] activitystreams::iri_string::validate::Error),
|
||||
|
||||
#[error("Couldn't normalize IRI, {0}")]
|
||||
#[error("Couldn't normalize IRI")]
|
||||
NormalizeIri(#[from] std::collections::TryReserveError),
|
||||
|
||||
#[error("Couldn't perform IO, {0}")]
|
||||
#[error("Couldn't perform IO")]
|
||||
Io(#[from] io::Error),
|
||||
|
||||
#[error("Couldn't sign string, {0}")]
|
||||
Rsa(rsa::errors::Error),
|
||||
|
||||
#[error("Couldn't use db, {0}")]
|
||||
#[error("Couldn't use db")]
|
||||
Sled(#[from] sled::Error),
|
||||
|
||||
#[error("Couldn't do the json thing, {0}")]
|
||||
#[error("Couldn't do the json thing")]
|
||||
Json(#[from] serde_json::Error),
|
||||
|
||||
#[error("Couldn't sign request, {0}")]
|
||||
#[error("Couldn't sign request")]
|
||||
Sign(#[from] SignError),
|
||||
|
||||
#[error("Couldn't sign digest")]
|
||||
Signature(#[from] rsa::signature::Error),
|
||||
|
||||
#[error("Couldn't prepare TLS private key")]
|
||||
PrepareKey(#[from] rustls::Error),
|
||||
|
||||
#[error("Couldn't verify signature")]
|
||||
VerifySignature,
|
||||
|
||||
|
@ -144,10 +189,10 @@ pub(crate) enum ErrorKind {
|
|||
#[error("Wrong ActivityPub kind, {0}")]
|
||||
Kind(String),
|
||||
|
||||
#[error("Too many CPUs, {0}")]
|
||||
#[error("Too many CPUs")]
|
||||
CpuCount(#[from] std::num::TryFromIntError),
|
||||
|
||||
#[error("{0}")]
|
||||
#[error("Host mismatch")]
|
||||
HostMismatch(#[from] CheckError),
|
||||
|
||||
#[error("Couldn't flush buffer")]
|
||||
|
@ -201,7 +246,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
|
||||
}
|
||||
|
@ -221,7 +266,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()),
|
||||
)
|
||||
|
|
|
@ -1,19 +1,15 @@
|
|||
use actix_web::{
|
||||
dev::Payload,
|
||||
error::ParseError,
|
||||
http::{
|
||||
header::{from_one_raw_str, Header, HeaderName, HeaderValue, TryIntoHeaderValue},
|
||||
StatusCode,
|
||||
},
|
||||
http::header::{from_one_raw_str, Header, HeaderName, HeaderValue, TryIntoHeaderValue},
|
||||
web::Data,
|
||||
FromRequest, HttpMessage, HttpRequest, HttpResponse, ResponseError,
|
||||
FromRequest, HttpMessage, HttpRequest,
|
||||
};
|
||||
use bcrypt::{BcryptError, DEFAULT_COST};
|
||||
use http_signature_normalization_actix::{prelude::InvalidHeaderValue, Canceled, Spawn};
|
||||
use std::{convert::Infallible, str::FromStr, time::Instant};
|
||||
use tracing_error::SpanTrace;
|
||||
|
||||
use crate::{db::Db, future::LocalBoxFuture, spawner::Spawner};
|
||||
use crate::{db::Db, error::Error, future::LocalBoxFuture, spawner::Spawner};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct AdminConfig {
|
||||
|
@ -83,74 +79,42 @@ impl Admin {
|
|||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
#[error("Failed authentication")]
|
||||
pub(crate) struct Error {
|
||||
context: String,
|
||||
#[source]
|
||||
kind: ErrorKind,
|
||||
}
|
||||
|
||||
impl Error {
|
||||
fn invalid() -> Self {
|
||||
Error {
|
||||
context: SpanTrace::capture().to_string(),
|
||||
kind: ErrorKind::Invalid,
|
||||
}
|
||||
Error::from(ErrorKind::Invalid)
|
||||
}
|
||||
|
||||
fn missing_config() -> Self {
|
||||
Error {
|
||||
context: SpanTrace::capture().to_string(),
|
||||
kind: ErrorKind::MissingConfig,
|
||||
}
|
||||
Error::from(ErrorKind::MissingConfig)
|
||||
}
|
||||
|
||||
fn missing_db() -> Self {
|
||||
Error {
|
||||
context: SpanTrace::capture().to_string(),
|
||||
kind: ErrorKind::MissingDb,
|
||||
}
|
||||
Error::from(ErrorKind::MissingDb)
|
||||
}
|
||||
|
||||
fn missing_spawner() -> Self {
|
||||
Error {
|
||||
context: SpanTrace::capture().to_string(),
|
||||
kind: ErrorKind::MissingSpawner,
|
||||
}
|
||||
Error::from(ErrorKind::MissingSpawner)
|
||||
}
|
||||
|
||||
fn bcrypt_verify(e: BcryptError) -> Self {
|
||||
Error {
|
||||
context: SpanTrace::capture().to_string(),
|
||||
kind: ErrorKind::BCryptVerify(e),
|
||||
}
|
||||
Error::from(ErrorKind::BCryptVerify(e))
|
||||
}
|
||||
|
||||
fn bcrypt_hash(e: BcryptError) -> Self {
|
||||
Error {
|
||||
context: SpanTrace::capture().to_string(),
|
||||
kind: ErrorKind::BCryptHash(e),
|
||||
}
|
||||
Error::from(ErrorKind::BCryptHash(e))
|
||||
}
|
||||
|
||||
fn parse_header(e: ParseError) -> Self {
|
||||
Error {
|
||||
context: SpanTrace::capture().to_string(),
|
||||
kind: ErrorKind::ParseHeader(e),
|
||||
}
|
||||
Error::from(ErrorKind::ParseHeader(e))
|
||||
}
|
||||
|
||||
fn canceled(_: Canceled) -> Self {
|
||||
Error {
|
||||
context: SpanTrace::capture().to_string(),
|
||||
kind: ErrorKind::Canceled,
|
||||
}
|
||||
Error::from(ErrorKind::Canceled)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
enum ErrorKind {
|
||||
pub(crate) enum ErrorKind {
|
||||
#[error("Invalid API Token")]
|
||||
Invalid,
|
||||
|
||||
|
@ -176,20 +140,6 @@ 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>>;
|
||||
|
@ -213,6 +163,10 @@ 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 {
|
||||
|
|
18
src/http1.rs
Normal file
18
src/http1.rs
Normal file
|
@ -0,0 +1,18 @@
|
|||
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")
|
||||
}
|
26
src/jobs.rs
26
src/jobs.rs
|
@ -40,7 +40,12 @@ 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,
|
||||
|
@ -48,18 +53,15 @@ pub(crate) fn create_workers(
|
|||
) -> std::io::Result<JobServer> {
|
||||
let deliver_concurrency = config.deliver_concurrency();
|
||||
|
||||
let queue_handle = WorkerConfig::new(
|
||||
MetricsStorage::wrap(Storage::new(TokioTimer)),
|
||||
move |queue_handle| {
|
||||
JobState::new(
|
||||
state.clone(),
|
||||
actors.clone(),
|
||||
JobServer::new(queue_handle),
|
||||
media.clone(),
|
||||
config.clone(),
|
||||
)
|
||||
},
|
||||
)
|
||||
let queue_handle = WorkerConfig::new(storage, move |queue_handle| {
|
||||
JobState::new(
|
||||
state.clone(),
|
||||
actors.clone(),
|
||||
JobServer::new(queue_handle),
|
||||
media.clone(),
|
||||
config.clone(),
|
||||
)
|
||||
})
|
||||
.register::<Deliver>()
|
||||
.register::<DeliverMany>()
|
||||
.register::<QueryNodeinfo>()
|
||||
|
|
|
@ -64,12 +64,13 @@ fn generate_announce(
|
|||
|
||||
impl Job for Announce {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -113,12 +113,13 @@ fn generate_accept_follow(
|
|||
|
||||
impl Job for Follow {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -49,12 +49,13 @@ impl Forward {
|
|||
|
||||
impl Job for Forward {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -35,12 +35,13 @@ impl Reject {
|
|||
|
||||
impl Job for Reject {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -50,12 +50,13 @@ impl Undo {
|
|||
|
||||
impl Job for Undo {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -87,13 +87,14 @@ fn to_contact(contact: AcceptedActors) -> Option<(String, String, IriString, Iri
|
|||
|
||||
impl Job for QueryContact {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -35,7 +35,7 @@ impl Deliver {
|
|||
}
|
||||
|
||||
#[tracing::instrument(name = "Deliver", skip(state))]
|
||||
async fn permform(self, state: JobState) -> Result<(), Error> {
|
||||
async fn perform(self, state: JobState) -> Result<(), Error> {
|
||||
if let Err(e) = state
|
||||
.state
|
||||
.requests
|
||||
|
@ -58,13 +58,14 @@ impl Deliver {
|
|||
|
||||
impl Job for Deliver {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.permform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -47,12 +47,13 @@ impl DeliverMany {
|
|||
|
||||
impl Job for DeliverMany {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -167,13 +167,14 @@ impl QueryInstance {
|
|||
|
||||
impl Job for QueryInstance {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -106,13 +106,14 @@ impl QueryNodeinfo {
|
|||
|
||||
impl Job for QueryNodeinfo {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -155,7 +156,7 @@ struct Link {
|
|||
#[serde(untagged)]
|
||||
enum MaybeSupported<T> {
|
||||
Supported(T),
|
||||
Unsupported(String),
|
||||
Unsupported(#[allow(unused)] String),
|
||||
}
|
||||
|
||||
impl<T> MaybeSupported<T> {
|
||||
|
@ -164,8 +165,8 @@ impl<T> MaybeSupported<T> {
|
|||
}
|
||||
}
|
||||
|
||||
struct SupportedVersion(String);
|
||||
struct SupportedNodeinfo(String);
|
||||
struct SupportedVersion(#[allow(unused)] String);
|
||||
struct SupportedNodeinfo(#[allow(unused)] String);
|
||||
|
||||
static SUPPORTED_VERSIONS: &str = "2.";
|
||||
static SUPPORTED_NODEINFO: &str = "http://nodeinfo.diaspora.software/ns/schema/2.";
|
||||
|
|
|
@ -25,12 +25,13 @@ impl Listeners {
|
|||
|
||||
impl Job for Listeners {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -15,13 +15,14 @@ impl RecordLastOnline {
|
|||
|
||||
impl Job for RecordLastOnline {
|
||||
type State = JobState;
|
||||
type Future = BoxFuture<'static, anyhow::Result<()>>;
|
||||
type Error = 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(async move { self.perform(state).await.map_err(Into::into) })
|
||||
Box::pin(self.perform(state))
|
||||
}
|
||||
}
|
||||
|
|
98
src/main.rs
98
src/main.rs
|
@ -12,7 +12,7 @@ use error::Error;
|
|||
use http_signature_normalization_actix::middleware::VerifySignature;
|
||||
use metrics_exporter_prometheus::PrometheusBuilder;
|
||||
use metrics_util::layers::FanoutBuilder;
|
||||
use opentelemetry::KeyValue;
|
||||
use opentelemetry::{trace::TracerProvider, KeyValue};
|
||||
use opentelemetry_otlp::WithExportConfig;
|
||||
use opentelemetry_sdk::Resource;
|
||||
use reqwest_middleware::ClientWithMiddleware;
|
||||
|
@ -21,7 +21,7 @@ use tokio::task::JoinHandle;
|
|||
use tracing_actix_web::TracingLogger;
|
||||
use tracing_error::ErrorLayer;
|
||||
use tracing_log::LogTracer;
|
||||
use tracing_subscriber::{filter::Targets, fmt::format::FmtSpan, layer::SubscriberExt, Layer};
|
||||
use tracing_subscriber::{filter::Targets, layer::SubscriberExt, Layer};
|
||||
|
||||
mod admin;
|
||||
mod apub;
|
||||
|
@ -33,11 +33,13 @@ 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;
|
||||
|
@ -56,16 +58,15 @@ use self::{
|
|||
fn init_subscriber(
|
||||
software_name: &'static str,
|
||||
opentelemetry_url: Option<&IriString>,
|
||||
) -> Result<(), anyhow::Error> {
|
||||
) -> color_eyre::Result<()> {
|
||||
LogTracer::init()?;
|
||||
color_eyre::install()?;
|
||||
|
||||
let targets: Targets = std::env::var("RUST_LOG")
|
||||
.unwrap_or_else(|_| "warn,actix_web=debug,actix_server=debug,tracing_actix_web=info".into())
|
||||
.unwrap_or_else(|_| "info".into())
|
||||
.parse()?;
|
||||
|
||||
let format_layer = tracing_subscriber::fmt::layer()
|
||||
.with_span_events(FmtSpan::NEW | FmtSpan::CLOSE)
|
||||
.with_filter(targets.clone());
|
||||
let format_layer = tracing_subscriber::fmt::layer().with_filter(targets.clone());
|
||||
|
||||
#[cfg(feature = "console")]
|
||||
let console_layer = ConsoleLayer::builder()
|
||||
|
@ -82,13 +83,11 @@ fn init_subscriber(
|
|||
let subscriber = subscriber.with(console_layer);
|
||||
|
||||
if let Some(url) = opentelemetry_url {
|
||||
let tracer = opentelemetry_otlp::new_pipeline()
|
||||
let tracer_provider = opentelemetry_otlp::new_pipeline()
|
||||
.tracing()
|
||||
.with_trace_config(
|
||||
opentelemetry_sdk::trace::config().with_resource(Resource::new(vec![
|
||||
KeyValue::new("service.name", software_name),
|
||||
])),
|
||||
)
|
||||
.with_trace_config(opentelemetry_sdk::trace::Config::default().with_resource(
|
||||
Resource::new(vec![KeyValue::new("service.name", software_name)]),
|
||||
))
|
||||
.with_exporter(
|
||||
opentelemetry_otlp::new_exporter()
|
||||
.tonic()
|
||||
|
@ -97,7 +96,7 @@ fn init_subscriber(
|
|||
.install_batch(opentelemetry_sdk::runtime::Tokio)?;
|
||||
|
||||
let otel_layer = tracing_opentelemetry::layer()
|
||||
.with_tracer(tracer)
|
||||
.with_tracer(tracer_provider.tracer(software_name))
|
||||
.with_filter(targets);
|
||||
|
||||
let subscriber = subscriber.with(otel_layer);
|
||||
|
@ -142,7 +141,7 @@ fn build_client(
|
|||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), anyhow::Error> {
|
||||
async fn main() -> color_eyre::Result<()> {
|
||||
dotenv::dotenv().ok();
|
||||
|
||||
let config = Config::build()?;
|
||||
|
@ -168,30 +167,30 @@ async fn main() -> Result<(), anyhow::Error> {
|
|||
.add_recorder(recorder)
|
||||
.add_recorder(collector.clone())
|
||||
.build();
|
||||
metrics::set_global_recorder(recorder).map_err(|e| anyhow::anyhow!("{e}"))?;
|
||||
metrics::set_global_recorder(recorder).map_err(|e| color_eyre::eyre::eyre!("{e}"))?;
|
||||
} else {
|
||||
collector.install()?;
|
||||
}
|
||||
|
||||
tracing::warn!("Opening DB");
|
||||
tracing::info!("Opening DB");
|
||||
let db = Db::build(&config)?;
|
||||
|
||||
tracing::warn!("Building caches");
|
||||
tracing::info!("Building caches");
|
||||
let actors = ActorCache::new(db.clone());
|
||||
let media = MediaCache::new(db.clone());
|
||||
|
||||
server_main(db, actors, media, collector, config).await?;
|
||||
|
||||
tracing::warn!("Application exit");
|
||||
tracing::info!("Application exit");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn client_main(config: Config, args: Args) -> JoinHandle<Result<(), anyhow::Error>> {
|
||||
fn client_main(config: Config, args: Args) -> JoinHandle<color_eyre::Result<()>> {
|
||||
tokio::spawn(do_client_main(config, args))
|
||||
}
|
||||
|
||||
async fn do_client_main(config: Config, args: Args) -> Result<(), anyhow::Error> {
|
||||
async fn do_client_main(config: Config, args: Args) -> color_eyre::Result<()> {
|
||||
let client = build_client(
|
||||
&config.user_agent(),
|
||||
config.client_timeout(),
|
||||
|
@ -282,14 +281,14 @@ async fn server_main(
|
|||
media: MediaCache,
|
||||
collector: MemoryCollector,
|
||||
config: Config,
|
||||
) -> Result<(), anyhow::Error> {
|
||||
) -> color_eyre::Result<()> {
|
||||
let client = build_client(
|
||||
&config.user_agent(),
|
||||
config.client_timeout(),
|
||||
config.proxy_config(),
|
||||
)?;
|
||||
|
||||
tracing::warn!("Creating state");
|
||||
tracing::info!("Creating state");
|
||||
|
||||
let (signature_threads, verify_threads) = match config.signature_threads() {
|
||||
0 | 1 => (1, 1),
|
||||
|
@ -309,19 +308,29 @@ async fn server_main(
|
|||
let state = State::build(db.clone(), key_id, sign_spawner.clone(), client).await?;
|
||||
|
||||
if let Some((token, admin_handle)) = config.telegram_info() {
|
||||
tracing::warn!("Creating telegram handler");
|
||||
tracing::info!("Creating telegram handler");
|
||||
telegram::start(admin_handle.to_owned(), db.clone(), token);
|
||||
}
|
||||
|
||||
let keys = config.open_keys()?;
|
||||
let cert_resolver = config
|
||||
.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(state.clone(), actors.clone(), media.clone(), config.clone())
|
||||
.expect("Failed to create job server");
|
||||
let job_server = create_workers(
|
||||
job_store.clone(),
|
||||
state.clone(),
|
||||
actors.clone(),
|
||||
media.clone(),
|
||||
config.clone(),
|
||||
)
|
||||
.expect("Failed to create job server");
|
||||
|
||||
let app = App::new()
|
||||
.app_data(web::Data::new(db.clone()))
|
||||
|
@ -387,27 +396,42 @@ async fn server_main(
|
|||
)
|
||||
});
|
||||
|
||||
if let Some((certs, key)) = keys {
|
||||
tracing::warn!("Binding to {}:{} with TLS", bind_address.0, bind_address.1);
|
||||
if let Some((cert_tx, cert_rx)) = cert_resolver {
|
||||
let handle = tokio::spawn(async move {
|
||||
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_single_cert(certs, key)?;
|
||||
.with_cert_resolver(cert_rx);
|
||||
server
|
||||
.bind_rustls_021(bind_address, server_config)?
|
||||
.bind_rustls_0_23(bind_address, server_config)?
|
||||
.run()
|
||||
.await?;
|
||||
|
||||
handle.abort();
|
||||
let _ = handle.await;
|
||||
} else {
|
||||
tracing::warn!("Binding to {}:{}", bind_address.0, bind_address.1);
|
||||
tracing::info!("Binding to {}:{}", bind_address.0, bind_address.1);
|
||||
server.bind(bind_address)?.run().await?;
|
||||
}
|
||||
|
||||
sign_spawner2.close().await;
|
||||
verify_spawner2.close().await;
|
||||
|
||||
tracing::warn!("Server closed");
|
||||
tracing::info!("Server closed");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
@ -80,7 +80,7 @@ where
|
|||
fn call(&self, req: ServiceRequest) -> Self::Future {
|
||||
let log_on_drop = LogOnDrop {
|
||||
begin: Instant::now(),
|
||||
path: req.path().to_string(),
|
||||
path: format!("{:?}", req.match_pattern()),
|
||||
method: req.method().to_string(),
|
||||
arm: false,
|
||||
};
|
||||
|
|
|
@ -2,6 +2,7 @@ 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;
|
||||
|
@ -24,6 +25,9 @@ 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
|
||||
|
@ -229,7 +233,11 @@ impl Requests {
|
|||
}
|
||||
}
|
||||
|
||||
return Err(ErrorKind::Status(parsed_url.to_string(), status).into());
|
||||
return Err(ErrorKind::Status(
|
||||
parsed_url.to_string(),
|
||||
crate::http1::status_to_http02(status),
|
||||
)
|
||||
.into());
|
||||
}
|
||||
|
||||
// only actually succeed a breaker on 2xx response
|
||||
|
@ -262,7 +270,7 @@ impl Requests {
|
|||
where
|
||||
T: serde::de::DeserializeOwned,
|
||||
{
|
||||
let body = self
|
||||
let stream = self
|
||||
.do_deliver(
|
||||
url,
|
||||
&serde_json::json!({}),
|
||||
|
@ -271,8 +279,9 @@ impl Requests {
|
|||
strategy,
|
||||
)
|
||||
.await?
|
||||
.bytes()
|
||||
.await?;
|
||||
.bytes_stream();
|
||||
|
||||
let body = aggregate(limit_stream(stream, JSON_SIZE_LIMIT)).await?;
|
||||
|
||||
Ok(serde_json::from_slice(&body)?)
|
||||
}
|
||||
|
@ -299,11 +308,12 @@ impl Requests {
|
|||
where
|
||||
T: serde::de::DeserializeOwned,
|
||||
{
|
||||
let body = self
|
||||
let stream = self
|
||||
.do_fetch_response(url, accept, strategy)
|
||||
.await?
|
||||
.bytes()
|
||||
.await?;
|
||||
.bytes_stream();
|
||||
|
||||
let body = aggregate(limit_stream(stream, JSON_SIZE_LIMIT)).await?;
|
||||
|
||||
Ok(serde_json::from_slice(&body)?)
|
||||
}
|
||||
|
|
|
@ -2,10 +2,14 @@ use crate::{
|
|||
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,13 +23,19 @@ pub(crate) async fn route(
|
|||
.fetch_response(&url, BreakerStrategy::Allow404AndBelow)
|
||||
.await?;
|
||||
|
||||
let mut response = HttpResponse::build(res.status());
|
||||
let mut response = HttpResponse::build(crate::http1::status_to_http02(res.status()));
|
||||
|
||||
for (name, value) in res.headers().iter().filter(|(h, _)| *h != "connection") {
|
||||
response.insert_header((name.clone(), value.clone()));
|
||||
response.insert_header((
|
||||
crate::http1::name_to_http02(name),
|
||||
crate::http1::value_to_http02(value),
|
||||
));
|
||||
}
|
||||
|
||||
return Ok(response.body(BodyStream::new(res.bytes_stream())));
|
||||
return Ok(response.body(BodyStream::new(limit_stream(
|
||||
res.bytes_stream(),
|
||||
IMAGE_SIZE_LIMIT,
|
||||
))));
|
||||
}
|
||||
|
||||
Ok(HttpResponse::NotFound().finish())
|
||||
|
|
|
@ -8,7 +8,7 @@ pub(crate) struct Spawner {
|
|||
}
|
||||
|
||||
impl Spawner {
|
||||
pub(crate) fn build(name: &'static str, threads: u16) -> anyhow::Result<Self> {
|
||||
pub(crate) fn build(name: &'static str, threads: u16) -> color_eyre::Result<Self> {
|
||||
let pool = CpuPool::configure()
|
||||
.name(name)
|
||||
.max_threads(threads)
|
||||
|
|
59
src/stream.rs
Normal file
59
src/stream.rs
Normal file
|
@ -0,0 +1,59 @@
|
|||
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())
|
||||
}
|
Loading…
Reference in a new issue