very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust
fjall
at-protocol
atproto
indexer
[firehose,lib,api] add is_source_running, shortcircuit requestCrawl if source is already running to prevent task restart
+15
-5
+5
src/api/xrpc/request_crawl.rs
+5
src/api/xrpc/request_crawl.rs
···
24
24
});
25
25
}
26
26
27
+
// if this host is already running, everything is already ok
28
+
if hydrant.firehose.is_source_running(&url) {
29
+
return Ok(StatusCode::OK);
30
+
}
31
+
27
32
// enforce daily new pds limit on unknown hosts
28
33
if !hydrant.firehose.is_source_known(&url) {
29
34
let (allowed, to_persist) = hydrant.state.pds_daily_limit.try_increment();
+10
-5
src/control/firehose.rs
+10
-5
src/control/firehose.rs
···
151
151
*self.state.firehose_enabled.borrow()
152
152
}
153
153
154
-
/// returns `true` if this URL is already a known firehose source — either currently
155
-
/// running or persisted (e.g. the host is offline but was previously added).
154
+
/// returns `true` if this URL is already a known firehose source.
155
+
/// either currently running or persisted (e.g. the host is offline but was previously added).
156
156
pub fn is_source_known(&self, url: &Url) -> bool {
157
157
self.known_sources.contains_sync(url)
158
158
}
159
159
160
+
/// return `true` if this source has a running firehose task (eg. its not offline).
161
+
pub fn is_source_running(&self, url: &Url) -> bool {
162
+
self.tasks.contains_sync(url)
163
+
}
164
+
160
165
/// list all currently active firehose sources.
161
166
pub async fn list_sources(&self) -> Vec<FirehoseSourceInfo> {
162
-
let mut out = Vec::new();
167
+
let mut out = Vec::with_capacity(self.tasks.capacity());
163
168
self.tasks
164
-
.any_async(|url, handle| {
169
+
.iter_async(|url, handle| {
165
170
out.push(FirehoseSourceInfo {
166
171
url: url.clone(),
167
172
is_pds: handle.is_pds,
168
173
});
169
-
false
174
+
true
170
175
})
171
176
.await;
172
177
out