Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
rate limit stats
+41
+23
Cargo.lock
+23
Cargo.lock
···
494
494
checksum = "f46ad14479a25103f283c0f10005961cf086d8dc42205bb44c46ac563475dca6"
495
495
496
496
[[package]]
497
+
name = "clocksource"
498
+
version = "0.8.1"
499
+
source = "registry+https://github.com/rust-lang/crates.io-index"
500
+
checksum = "129026dd5a8a9592d96916258f3a5379589e513ea5e86aeb0bd2530286e44e9e"
501
+
dependencies = [
502
+
"libc",
503
+
"time",
504
+
"winapi",
505
+
]
506
+
507
+
[[package]]
497
508
name = "colorchoice"
498
509
version = "1.0.3"
499
510
source = "registry+https://github.com/rust-lang/crates.io-index"
···
520
531
"metrics-exporter-prometheus",
521
532
"metrics-process",
522
533
"num-format",
534
+
"ratelimit",
523
535
"rocksdb",
524
536
"serde",
525
537
"serde_with",
···
1640
1652
checksum = "6f97cdb2a36ed4183de61b2f824cc45c9f1037f28afe0a322e9fff4c108b5aaa"
1641
1653
dependencies = [
1642
1654
"rand_core",
1655
+
]
1656
+
1657
+
[[package]]
1658
+
name = "ratelimit"
1659
+
version = "0.10.0"
1660
+
source = "registry+https://github.com/rust-lang/crates.io-index"
1661
+
checksum = "36ea961700fd7260e7fa3701c8287d901b2172c51f9c1421fa0f21d7f7e184b7"
1662
+
dependencies = [
1663
+
"clocksource",
1664
+
"parking_lot",
1665
+
"thiserror 1.0.69",
1643
1666
]
1644
1667
1645
1668
[[package]]
+1
constellation/Cargo.toml
+1
constellation/Cargo.toml
···
22
22
metrics-exporter-prometheus = { version = "0.16.1", default-features = false, features = ["http-listener"] }
23
23
metrics-process = "2.4.0"
24
24
num-format = "0.4.4"
25
+
ratelimit = "0.10.0"
25
26
rocksdb = { version = "0.23.0", optional = true }
26
27
serde = { version = "1.0.215", features = ["derive"] }
27
28
serde_with = { version = "3.12.0", features = ["hex"] }
+17
constellation/src/bin/rocks-target-stats.rs
+17
constellation/src/bin/rocks-target-stats.rs
···
20
20
/// where is rocksdb's data
21
21
#[arg(short, long)]
22
22
data: PathBuf,
23
+
/// slow down so we don't kill the firehose consumer, if running concurrently
24
+
#[arg(short, long)]
25
+
limit: Option<u64>,
23
26
}
24
27
25
28
type LinkType = String;
···
69
72
fn main() {
70
73
let args = Args::parse();
71
74
75
+
let limit = args.limit.map(|amount| {
76
+
ratelimit::Ratelimiter::builder(amount, time::Duration::from_secs(1))
77
+
.max_tokens(amount)
78
+
.initial_available(amount)
79
+
.build()
80
+
.unwrap()
81
+
});
82
+
72
83
eprintln!("starting rocksdb...");
73
84
let rocks = RocksStorage::open_readonly(args.data).unwrap();
74
85
eprintln!("rocks ready.");
···
87
98
88
99
let mut i = 0;
89
100
for item in db.iterator_cf(&target_id_cf, IteratorMode::Start) {
101
+
if let Some(ref limiter) = limit {
102
+
if let Err(dur) = limiter.try_wait() {
103
+
std::thread::sleep(dur)
104
+
}
105
+
}
106
+
90
107
if i > 0 && i % REPORT_INTERVAL == 0 {
91
108
let now = time::Instant::now();
92
109
let rate = (REPORT_INTERVAL as f32) / (now.duration_since(t_prev).as_secs_f32());