Compare commits
19 commits
Author | SHA1 | Date | |
---|---|---|---|
asonix | 97567cf598 | ||
asonix | 4c663f399e | ||
asonix | 8a3256f52a | ||
asonix | 13a2653fe8 | ||
asonix | 8dd9a86d22 | ||
asonix | 5c0c0591dd | ||
asonix | 04ca4e5401 | ||
asonix | 1de1d76506 | ||
asonix | dd9225bb89 | ||
asonix | b577730836 | ||
asonix | 21883c168b | ||
asonix | 76a0c79369 | ||
asonix | 6444782db9 | ||
asonix | 14aea3256d | ||
asonix | f4f2aa2025 | ||
asonix | 615271fe80 | ||
asonix | 4aed601664 | ||
asonix | bf21f05aca | ||
asonix | e69f6c6edb |
|
@ -207,7 +207,6 @@ jobs:
|
||||||
direction: upload
|
direction: upload
|
||||||
token: ${{ secrets.GITHUB_TOKEN }}
|
token: ${{ secrets.GITHUB_TOKEN }}
|
||||||
release-dir: artifacts/
|
release-dir: artifacts/
|
||||||
prerelease: true
|
|
||||||
|
|
||||||
publish-crate:
|
publish-crate:
|
||||||
needs: [build]
|
needs: [build]
|
||||||
|
|
837
Cargo.lock
generated
837
Cargo.lock
generated
File diff suppressed because it is too large
Load diff
22
Cargo.toml
22
Cargo.toml
|
@ -1,7 +1,7 @@
|
||||||
[package]
|
[package]
|
||||||
name = "ap-relay"
|
name = "ap-relay"
|
||||||
description = "A simple activitypub relay"
|
description = "A simple activitypub relay"
|
||||||
version = "0.3.109"
|
version = "0.3.113"
|
||||||
authors = ["asonix <asonix@asonix.dog>"]
|
authors = ["asonix <asonix@asonix.dog>"]
|
||||||
license = "AGPL-3.0"
|
license = "AGPL-3.0"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
|
@ -14,6 +14,9 @@ build = "src/build.rs"
|
||||||
name = "relay"
|
name = "relay"
|
||||||
path = "src/main.rs"
|
path = "src/main.rs"
|
||||||
|
|
||||||
|
[profile.release]
|
||||||
|
strip = true
|
||||||
|
|
||||||
[features]
|
[features]
|
||||||
console = ["dep:console-subscriber"]
|
console = ["dep:console-subscriber"]
|
||||||
default = []
|
default = []
|
||||||
|
@ -25,17 +28,16 @@ actix-web = { version = "4.4.0", default-features = false, features = ["compress
|
||||||
actix-webfinger = { version = "0.5.0", default-features = false }
|
actix-webfinger = { version = "0.5.0", default-features = false }
|
||||||
activitystreams = "0.7.0-alpha.25"
|
activitystreams = "0.7.0-alpha.25"
|
||||||
activitystreams-ext = "0.1.0-alpha.3"
|
activitystreams-ext = "0.1.0-alpha.3"
|
||||||
ammonia = "3.1.0"
|
ammonia = "4.0.0"
|
||||||
async-cpupool = "0.2.0"
|
async-cpupool = "0.2.2"
|
||||||
bcrypt = "0.15"
|
bcrypt = "0.15"
|
||||||
base64 = "0.21"
|
base64 = "0.22"
|
||||||
clap = { version = "4.0.0", features = ["derive"] }
|
clap = { version = "4.0.0", features = ["derive"] }
|
||||||
color-eyre = "0.6.2"
|
color-eyre = "0.6.2"
|
||||||
config = { version = "0.14.0", default-features = false, features = ["toml", "json", "yaml"] }
|
config = { version = "0.14.0", default-features = false, features = ["toml", "json", "yaml"] }
|
||||||
console-subscriber = { version = "0.2", optional = true }
|
console-subscriber = { version = "0.2", optional = true }
|
||||||
dashmap = "5.1.0"
|
dashmap = "5.1.0"
|
||||||
dotenv = "0.15.0"
|
dotenv = "0.15.0"
|
||||||
flume = "0.11.0"
|
|
||||||
lru = "0.12.0"
|
lru = "0.12.0"
|
||||||
metrics = "0.22.0"
|
metrics = "0.22.0"
|
||||||
metrics-exporter-prometheus = { version = "0.13.0", default-features = false, features = [
|
metrics-exporter-prometheus = { version = "0.13.0", default-features = false, features = [
|
||||||
|
@ -44,9 +46,9 @@ metrics-exporter-prometheus = { version = "0.13.0", default-features = false, fe
|
||||||
metrics-util = "0.16.0"
|
metrics-util = "0.16.0"
|
||||||
mime = "0.3.16"
|
mime = "0.3.16"
|
||||||
minify-html = "0.15.0"
|
minify-html = "0.15.0"
|
||||||
opentelemetry = "0.21"
|
opentelemetry = "0.22"
|
||||||
opentelemetry_sdk = { version = "0.21", features = ["rt-tokio"] }
|
opentelemetry_sdk = { version = "0.22", features = ["rt-tokio"] }
|
||||||
opentelemetry-otlp = "0.14"
|
opentelemetry-otlp = "0.15"
|
||||||
pin-project-lite = "0.2.9"
|
pin-project-lite = "0.2.9"
|
||||||
# pinned to metrics-util
|
# pinned to metrics-util
|
||||||
quanta = "0.12.0"
|
quanta = "0.12.0"
|
||||||
|
@ -55,7 +57,7 @@ reqwest = { version = "0.11", default-features = false, features = ["rustls-tls"
|
||||||
reqwest-middleware = "0.2"
|
reqwest-middleware = "0.2"
|
||||||
reqwest-tracing = "0.4.5"
|
reqwest-tracing = "0.4.5"
|
||||||
ring = "0.17.5"
|
ring = "0.17.5"
|
||||||
rsa = { version = "0.9" }
|
rsa = "0.9"
|
||||||
rsa-magic-public-key = "0.8.0"
|
rsa-magic-public-key = "0.8.0"
|
||||||
rustls = "0.22.0"
|
rustls = "0.22.0"
|
||||||
rustls-channel-resolver = "0.2.0"
|
rustls-channel-resolver = "0.2.0"
|
||||||
|
@ -74,7 +76,7 @@ time = { version = "0.3.17", features = ["serde"] }
|
||||||
tracing = "0.1"
|
tracing = "0.1"
|
||||||
tracing-error = "0.2"
|
tracing-error = "0.2"
|
||||||
tracing-log = "0.2"
|
tracing-log = "0.2"
|
||||||
tracing-opentelemetry = "0.22"
|
tracing-opentelemetry = "0.23"
|
||||||
tracing-subscriber = { version = "0.3", features = [
|
tracing-subscriber = { version = "0.3", features = [
|
||||||
"ansi",
|
"ansi",
|
||||||
"env-filter",
|
"env-filter",
|
||||||
|
|
12
flake.lock
12
flake.lock
|
@ -5,11 +5,11 @@
|
||||||
"systems": "systems"
|
"systems": "systems"
|
||||||
},
|
},
|
||||||
"locked": {
|
"locked": {
|
||||||
"lastModified": 1701680307,
|
"lastModified": 1710146030,
|
||||||
"narHash": "sha256-kAuep2h5ajznlPMD9rnQyffWG8EM/C73lejGofXvdM8=",
|
"narHash": "sha256-SZ5L6eA7HJ/nmkzGG7/ISclqe6oZdOZTNoesiInkXPQ=",
|
||||||
"owner": "numtide",
|
"owner": "numtide",
|
||||||
"repo": "flake-utils",
|
"repo": "flake-utils",
|
||||||
"rev": "4022d587cbbfd70fe950c1e2083a02621806a725",
|
"rev": "b1d9ab70662946ef0850d488da1c9019f3a9752a",
|
||||||
"type": "github"
|
"type": "github"
|
||||||
},
|
},
|
||||||
"original": {
|
"original": {
|
||||||
|
@ -20,11 +20,11 @@
|
||||||
},
|
},
|
||||||
"nixpkgs": {
|
"nixpkgs": {
|
||||||
"locked": {
|
"locked": {
|
||||||
"lastModified": 1705133751,
|
"lastModified": 1711163522,
|
||||||
"narHash": "sha256-rCIsyE80jgiOU78gCWN3A0wE0tR2GI5nH6MlS+HaaSQ=",
|
"narHash": "sha256-YN/Ciidm+A0fmJPWlHBGvVkcarYWSC+s3NTPk/P+q3c=",
|
||||||
"owner": "NixOS",
|
"owner": "NixOS",
|
||||||
"repo": "nixpkgs",
|
"repo": "nixpkgs",
|
||||||
"rev": "9b19f5e77dd906cb52dade0b7bd280339d2a1f3d",
|
"rev": "44d0940ea560dee511026a53f0e2e2cde489b4d4",
|
||||||
"type": "github"
|
"type": "github"
|
||||||
},
|
},
|
||||||
"original": {
|
"original": {
|
||||||
|
|
|
@ -5,7 +5,7 @@
|
||||||
|
|
||||||
rustPlatform.buildRustPackage {
|
rustPlatform.buildRustPackage {
|
||||||
pname = "relay";
|
pname = "relay";
|
||||||
version = "0.3.109";
|
version = "0.3.113";
|
||||||
src = ./.;
|
src = ./.;
|
||||||
cargoLock.lockFile = ./Cargo.lock;
|
cargoLock.lockFile = ./Cargo.lock;
|
||||||
|
|
||||||
|
|
|
@ -15,6 +15,10 @@ const MINUTES: u64 = 60 * SECONDS;
|
||||||
const HOURS: u64 = 60 * MINUTES;
|
const HOURS: u64 = 60 * MINUTES;
|
||||||
const DAYS: u64 = 24 * HOURS;
|
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>;
|
type DistributionMap = BTreeMap<Vec<(String, String)>, Summary>;
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
|
@ -299,7 +303,14 @@ impl Inner {
|
||||||
for sample in samples {
|
for sample in samples {
|
||||||
entry.add(*sample);
|
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();
|
let d = self.distributions.read().unwrap().clone();
|
||||||
|
@ -358,6 +369,7 @@ impl MemoryCollector {
|
||||||
) {
|
) {
|
||||||
let mut d = self.inner.descriptions.write().unwrap();
|
let mut d = self.inner.descriptions.write().unwrap();
|
||||||
d.entry(key.as_str().to_owned()).or_insert(description);
|
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<Self>> {
|
||||||
|
|
|
@ -9,10 +9,10 @@ pub(crate) struct LastOnline {
|
||||||
impl LastOnline {
|
impl LastOnline {
|
||||||
pub(crate) fn mark_seen(&self, iri: &IriStr) {
|
pub(crate) fn mark_seen(&self, iri: &IriStr) {
|
||||||
if let Some(authority) = iri.authority_str() {
|
if let Some(authority) = iri.authority_str() {
|
||||||
self.domains
|
let mut guard = self.domains.lock().unwrap();
|
||||||
.lock()
|
guard.insert(authority.to_string(), OffsetDateTime::now_utc());
|
||||||
.unwrap()
|
metrics::gauge!("relay.last-online.size",)
|
||||||
.insert(authority.to_string(), OffsetDateTime::now_utc());
|
.set(crate::collector::recordable(guard.len()));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -73,7 +73,9 @@ impl State {
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn cache(&self, object_id: IriString, actor_id: IriString) {
|
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 {
|
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},
|
pkcs8::{DecodePrivateKey, EncodePrivateKey},
|
||||||
RsaPrivateKey,
|
RsaPrivateKey,
|
||||||
};
|
};
|
||||||
use sled::{Batch, Tree};
|
use sled::{transaction::TransactionError, Batch, Transactional, Tree};
|
||||||
use std::{
|
use std::{
|
||||||
collections::{BTreeMap, HashMap},
|
collections::{BTreeMap, HashMap},
|
||||||
sync::{
|
sync::{
|
||||||
|
@ -283,10 +283,15 @@ impl Db {
|
||||||
pub(crate) async fn check_health(&self) -> Result<(), Error> {
|
pub(crate) async fn check_health(&self) -> Result<(), Error> {
|
||||||
let next = self.inner.healthz_counter.fetch_add(1, Ordering::Relaxed);
|
let next = self.inner.healthz_counter.fetch_add(1, Ordering::Relaxed);
|
||||||
self.unblock(move |inner| {
|
self.unblock(move |inner| {
|
||||||
inner
|
let res = inner
|
||||||
.healthz
|
.healthz
|
||||||
.insert("healthz", &next.to_be_bytes()[..])
|
.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?;
|
.await?;
|
||||||
self.inner.healthz.flush_async().await?;
|
self.inner.healthz.flush_async().await?;
|
||||||
|
@ -349,6 +354,9 @@ impl Db {
|
||||||
.actor_id_info
|
.actor_id_info
|
||||||
.insert(actor_id.as_str().as_bytes(), vec)?;
|
.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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -383,6 +391,9 @@ impl Db {
|
||||||
.actor_id_instance
|
.actor_id_instance
|
||||||
.insert(actor_id.as_str().as_bytes(), vec)?;
|
.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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -417,6 +428,9 @@ impl Db {
|
||||||
.actor_id_contact
|
.actor_id_contact
|
||||||
.insert(actor_id.as_str().as_bytes(), vec)?;
|
.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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -447,6 +461,12 @@ impl Db {
|
||||||
inner
|
inner
|
||||||
.media_url_media_id
|
.media_url_media_id
|
||||||
.insert(url.as_str().as_bytes(), id.as_bytes())?;
|
.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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -538,6 +558,14 @@ impl Db {
|
||||||
inner
|
inner
|
||||||
.actor_id_actor
|
.actor_id_actor
|
||||||
.insert(actor.id.as_str().as_bytes(), vec)?;
|
.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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -550,6 +578,10 @@ impl Db {
|
||||||
.connected_actor_ids
|
.connected_actor_ids
|
||||||
.remove(actor_id.as_str().as_bytes())?;
|
.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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -562,6 +594,10 @@ impl Db {
|
||||||
.connected_actor_ids
|
.connected_actor_ids
|
||||||
.insert(actor_id.as_str().as_bytes(), actor_id.as_str().as_bytes())?;
|
.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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -569,30 +605,64 @@ impl Db {
|
||||||
|
|
||||||
pub(crate) async fn add_blocks(&self, domains: Vec<String>) -> Result<(), Error> {
|
pub(crate) async fn add_blocks(&self, domains: Vec<String>) -> Result<(), Error> {
|
||||||
self.unblock(move |inner| {
|
self.unblock(move |inner| {
|
||||||
for connected in inner.connected_by_domain(&domains) {
|
let connected_by_domain = inner.connected_by_domain(&domains).collect::<Vec<_>>();
|
||||||
inner
|
|
||||||
.connected_actor_ids
|
|
||||||
.remove(connected.as_str().as_bytes())?;
|
|
||||||
}
|
|
||||||
|
|
||||||
for authority in &domains {
|
let res = (
|
||||||
inner
|
&inner.connected_actor_ids,
|
||||||
.blocked_domains
|
&inner.blocked_domains,
|
||||||
.insert(domain_key(authority), authority.as_bytes())?;
|
&inner.allowed_domains,
|
||||||
inner.allowed_domains.remove(domain_key(authority))?;
|
)
|
||||||
}
|
.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
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) async fn remove_blocks(&self, domains: Vec<String>) -> Result<(), Error> {
|
pub(crate) async fn remove_blocks(&self, domains: Vec<String>) -> Result<(), Error> {
|
||||||
self.unblock(move |inner| {
|
self.unblock(move |inner| {
|
||||||
|
let mut blocked_batch = Batch::default();
|
||||||
|
|
||||||
for authority in &domains {
|
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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -600,12 +670,17 @@ impl Db {
|
||||||
|
|
||||||
pub(crate) async fn add_allows(&self, domains: Vec<String>) -> Result<(), Error> {
|
pub(crate) async fn add_allows(&self, domains: Vec<String>) -> Result<(), Error> {
|
||||||
self.unblock(move |inner| {
|
self.unblock(move |inner| {
|
||||||
|
let mut allowed_batch = Batch::default();
|
||||||
|
|
||||||
for authority in &domains {
|
for authority in &domains {
|
||||||
inner
|
allowed_batch.insert(domain_key(authority).as_bytes(), authority.as_bytes());
|
||||||
.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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -614,17 +689,32 @@ impl Db {
|
||||||
pub(crate) async fn remove_allows(&self, domains: Vec<String>) -> Result<(), Error> {
|
pub(crate) async fn remove_allows(&self, domains: Vec<String>) -> Result<(), Error> {
|
||||||
self.unblock(move |inner| {
|
self.unblock(move |inner| {
|
||||||
if inner.restricted_mode {
|
if inner.restricted_mode {
|
||||||
for connected in inner.connected_by_domain(&domains) {
|
let connected_by_domain = inner.connected_by_domain(&domains).collect::<Vec<_>>();
|
||||||
inner
|
|
||||||
.connected_actor_ids
|
let mut connected_batch = Batch::default();
|
||||||
.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 {
|
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(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
@ -665,6 +755,10 @@ impl Db {
|
||||||
inner
|
inner
|
||||||
.settings
|
.settings
|
||||||
.insert("private-key".as_bytes(), pem_pkcs8.as_bytes())?;
|
.insert("private-key".as_bytes(), pem_pkcs8.as_bytes())?;
|
||||||
|
|
||||||
|
metrics::gauge!("relay.db.settings.size")
|
||||||
|
.set(crate::collector::recordable(inner.settings.len()));
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
|
|
26
src/jobs.rs
26
src/jobs.rs
|
@ -40,7 +40,12 @@ fn debug_object(activity: &serde_json::Value) -> &serde_json::Value {
|
||||||
object
|
object
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub(crate) fn build_storage() -> MetricsStorage<Storage<TokioTimer>> {
|
||||||
|
MetricsStorage::wrap(Storage::new(TokioTimer))
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) fn create_workers(
|
pub(crate) fn create_workers(
|
||||||
|
storage: MetricsStorage<Storage<TokioTimer>>,
|
||||||
state: State,
|
state: State,
|
||||||
actors: ActorCache,
|
actors: ActorCache,
|
||||||
media: MediaCache,
|
media: MediaCache,
|
||||||
|
@ -48,18 +53,15 @@ pub(crate) fn create_workers(
|
||||||
) -> std::io::Result<JobServer> {
|
) -> std::io::Result<JobServer> {
|
||||||
let deliver_concurrency = config.deliver_concurrency();
|
let deliver_concurrency = config.deliver_concurrency();
|
||||||
|
|
||||||
let queue_handle = WorkerConfig::new(
|
let queue_handle = WorkerConfig::new(storage, move |queue_handle| {
|
||||||
MetricsStorage::wrap(Storage::new(TokioTimer)),
|
JobState::new(
|
||||||
move |queue_handle| {
|
state.clone(),
|
||||||
JobState::new(
|
actors.clone(),
|
||||||
state.clone(),
|
JobServer::new(queue_handle),
|
||||||
actors.clone(),
|
media.clone(),
|
||||||
JobServer::new(queue_handle),
|
config.clone(),
|
||||||
media.clone(),
|
)
|
||||||
config.clone(),
|
})
|
||||||
)
|
|
||||||
},
|
|
||||||
)
|
|
||||||
.register::<Deliver>()
|
.register::<Deliver>()
|
||||||
.register::<DeliverMany>()
|
.register::<DeliverMany>()
|
||||||
.register::<QueryNodeinfo>()
|
.register::<QueryNodeinfo>()
|
||||||
|
|
12
src/main.rs
12
src/main.rs
|
@ -321,10 +321,16 @@ async fn server_main(
|
||||||
let sign_spawner2 = sign_spawner.clone();
|
let sign_spawner2 = sign_spawner.clone();
|
||||||
let verify_spawner2 = verify_spawner.clone();
|
let verify_spawner2 = verify_spawner.clone();
|
||||||
let config2 = config.clone();
|
let config2 = config.clone();
|
||||||
|
let job_store = jobs::build_storage();
|
||||||
let server = HttpServer::new(move || {
|
let server = HttpServer::new(move || {
|
||||||
let job_server =
|
let job_server = create_workers(
|
||||||
create_workers(state.clone(), actors.clone(), media.clone(), config.clone())
|
job_store.clone(),
|
||||||
.expect("Failed to create job server");
|
state.clone(),
|
||||||
|
actors.clone(),
|
||||||
|
media.clone(),
|
||||||
|
config.clone(),
|
||||||
|
)
|
||||||
|
.expect("Failed to create job server");
|
||||||
|
|
||||||
let app = App::new()
|
let app = App::new()
|
||||||
.app_data(web::Data::new(db.clone()))
|
.app_data(web::Data::new(db.clone()))
|
||||||
|
|
|
@ -80,7 +80,7 @@ where
|
||||||
fn call(&self, req: ServiceRequest) -> Self::Future {
|
fn call(&self, req: ServiceRequest) -> Self::Future {
|
||||||
let log_on_drop = LogOnDrop {
|
let log_on_drop = LogOnDrop {
|
||||||
begin: Instant::now(),
|
begin: Instant::now(),
|
||||||
path: req.path().to_string(),
|
path: format!("{:?}", req.match_pattern()),
|
||||||
method: req.method().to_string(),
|
method: req.method().to_string(),
|
||||||
arm: false,
|
arm: false,
|
||||||
};
|
};
|
||||||
|
|
Loading…
Reference in a new issue