This repository has no description
feat(ingest): ResolverStats, HydrantStreamErrorFrame, profiling
Lewis: May this revision serve well! <lu5a@proton.me>
+250
-21
+12
Cargo.lock
+12
Cargo.lock
···
338
338
"serde_json",
339
339
"thiserror 2.0.18",
340
340
"tokio",
341
+
"tokio-stream",
341
342
"tokio-util",
342
343
"tracing",
343
344
"tracing-subscriber",
···
3747
3748
checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61"
3748
3749
dependencies = [
3749
3750
"rustls",
3751
+
"tokio",
3752
+
]
3753
+
3754
+
[[package]]
3755
+
name = "tokio-stream"
3756
+
version = "0.1.18"
3757
+
source = "registry+https://github.com/rust-lang/crates.io-index"
3758
+
checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70"
3759
+
dependencies = [
3760
+
"futures-core",
3761
+
"pin-project-lite",
3750
3762
"tokio",
3751
3763
]
3752
3764
+7
Cargo.toml
+7
Cargo.toml
···
40
40
41
41
tokio = { version = "1.52", features = ["macros", "rt-multi-thread", "time", "signal", "io-util", "sync"] }
42
42
tokio-util = "0.7"
43
+
tokio-stream = "0.1"
43
44
tokio-tungstenite = { version = "0.29", features = ["rustls-tls-webpki-roots"] }
44
45
futures = "0.3"
45
46
···
87
88
[profile.bench]
88
89
debug = 1
89
90
strip = false
91
+
92
+
[profile.profiling]
93
+
inherits = "release"
94
+
debug = 1
95
+
strip = false
96
+
lto = "thin"
+5
-3
containerfiles/bobbin.Containerfile
+5
-3
containerfiles/bobbin.Containerfile
···
1
1
FROM docker.io/library/rust:1-alpine3.23 AS builder
2
2
RUN apk add --no-cache build-base musl-dev cmake perl pkgconfig
3
+
ARG BOBBIN_PROFILE=release
3
4
WORKDIR /src
4
5
COPY . ./
5
6
RUN rm -f .cargo/config.toml
6
-
RUN cargo build --release --bin bobbin --package bobbin
7
-
RUN strip target/release/bobbin
7
+
RUN cargo build --profile ${BOBBIN_PROFILE} --bin bobbin --package bobbin
8
+
RUN if [ "${BOBBIN_PROFILE}" = "release" ]; then strip target/${BOBBIN_PROFILE}/bobbin; fi
8
9
9
10
FROM docker.io/library/alpine:3.23
11
+
ARG BOBBIN_PROFILE=release
10
12
RUN apk add --no-cache ca-certificates
11
-
COPY --from=builder /src/target/release/bobbin /usr/local/bin/bobbin
13
+
COPY --from=builder /src/target/${BOBBIN_PROFILE}/bobbin /usr/local/bin/bobbin
12
14
ENV BOBBIN_BIND=0.0.0.0:8090
13
15
EXPOSE 8090
14
16
ENTRYPOINT ["/usr/local/bin/bobbin"]
+1
crates/ingest/Cargo.toml
+1
crates/ingest/Cargo.toml
+7
crates/ingest/src/frame.rs
+7
crates/ingest/src/frame.rs
···
39
39
}
40
40
41
41
#[derive(Clone, Debug, Deserialize)]
42
+
pub struct HydrantStreamErrorFrame {
43
+
pub error: String,
44
+
#[serde(default)]
45
+
pub message: Option<String>,
46
+
}
47
+
48
+
#[derive(Clone, Debug, Deserialize)]
42
49
pub struct RecordFrame {
43
50
pub live: bool,
44
51
pub did: Did<DefaultStr>,
+212
-12
crates/ingest/src/resolver.rs
+212
-12
crates/ingest/src/resolver.rs
···
1
-
use bobbin_runtime::RuntimeHasher;
1
+
use std::sync::Arc;
2
+
use std::sync::atomic::{AtomicU64, Ordering};
3
+
use std::time::Duration;
4
+
5
+
use bobbin_runtime::{Clock, RuntimeHasher};
2
6
use bobbin_slingshot_client::{SlingshotClient, SlingshotError};
3
7
use bobbin_types::edges::{ExtractError, Record};
4
8
use bobbin_types::ids::nsid_static;
···
61
65
}
62
66
}
63
67
68
+
#[derive(Default)]
69
+
pub struct ResolverStats {
70
+
hits: AtomicU64,
71
+
misses_mapped: AtomicU64,
72
+
misses_no_repo_did: AtomicU64,
73
+
misses_unresolvable: AtomicU64,
74
+
misses_transient: AtomicU64,
75
+
misses_no_client: AtomicU64,
76
+
miss_latency_micros_sum: AtomicU64,
77
+
miss_latency_micros_max: AtomicU64,
78
+
}
79
+
80
+
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
81
+
pub struct ResolverStatsSnapshot {
82
+
pub hits: u64,
83
+
pub misses_mapped: u64,
84
+
pub misses_no_repo_did: u64,
85
+
pub misses_unresolvable: u64,
86
+
pub misses_transient: u64,
87
+
pub misses_no_client: u64,
88
+
pub miss_latency_micros_sum: u64,
89
+
pub miss_latency_micros_max: u64,
90
+
}
91
+
92
+
impl ResolverStatsSnapshot {
93
+
pub fn miss_count(&self) -> u64 {
94
+
self.misses_mapped
95
+
+ self.misses_no_repo_did
96
+
+ self.misses_unresolvable
97
+
+ self.misses_transient
98
+
+ self.misses_no_client
99
+
}
100
+
101
+
pub fn total(&self) -> u64 {
102
+
self.hits + self.miss_count()
103
+
}
104
+
105
+
pub fn miss_latency_micros_avg(&self) -> Option<u64> {
106
+
let misses = self.miss_count() - self.misses_no_client;
107
+
(misses > 0).then(|| self.miss_latency_micros_sum / misses)
108
+
}
109
+
}
110
+
111
+
#[derive(Clone, Copy)]
112
+
enum MissKind {
113
+
Mapped,
114
+
NoRepoDid,
115
+
Unresolvable,
116
+
Transient,
117
+
NoClient,
118
+
}
119
+
120
+
impl ResolverStats {
121
+
fn record_hit(&self) {
122
+
self.hits.fetch_add(1, Ordering::Relaxed);
123
+
}
124
+
125
+
fn record_miss(&self, kind: MissKind, latency: Option<Duration>) {
126
+
let counter = match kind {
127
+
MissKind::Mapped => &self.misses_mapped,
128
+
MissKind::NoRepoDid => &self.misses_no_repo_did,
129
+
MissKind::Unresolvable => &self.misses_unresolvable,
130
+
MissKind::Transient => &self.misses_transient,
131
+
MissKind::NoClient => &self.misses_no_client,
132
+
};
133
+
counter.fetch_add(1, Ordering::Relaxed);
134
+
if let Some(latency) = latency {
135
+
let micros = u64::try_from(latency.as_micros()).unwrap_or(u64::MAX);
136
+
self.miss_latency_micros_sum
137
+
.fetch_add(micros, Ordering::Relaxed);
138
+
self.miss_latency_micros_max
139
+
.fetch_max(micros, Ordering::Relaxed);
140
+
}
141
+
}
142
+
143
+
pub fn snapshot(&self) -> ResolverStatsSnapshot {
144
+
ResolverStatsSnapshot {
145
+
hits: self.hits.load(Ordering::Relaxed),
146
+
misses_mapped: self.misses_mapped.load(Ordering::Relaxed),
147
+
misses_no_repo_did: self.misses_no_repo_did.load(Ordering::Relaxed),
148
+
misses_unresolvable: self.misses_unresolvable.load(Ordering::Relaxed),
149
+
misses_transient: self.misses_transient.load(Ordering::Relaxed),
150
+
misses_no_client: self.misses_no_client.load(Ordering::Relaxed),
151
+
miss_latency_micros_sum: self.miss_latency_micros_sum.load(Ordering::Relaxed),
152
+
miss_latency_micros_max: self.miss_latency_micros_max.load(Ordering::Relaxed),
153
+
}
154
+
}
155
+
}
156
+
157
+
struct SlingshotProbe {
158
+
client: SlingshotClient,
159
+
clock: Arc<dyn Clock>,
160
+
}
161
+
64
162
pub struct RepoIdResolver {
65
163
cache: SccMap<RepoRef, CacheEntry, RuntimeHasher>,
66
-
client: Option<SlingshotClient>,
164
+
probe: Option<SlingshotProbe>,
165
+
stats: ResolverStats,
67
166
}
68
167
69
168
impl RepoIdResolver {
70
-
pub fn with_slingshot(client: SlingshotClient, hasher: RuntimeHasher) -> Self {
169
+
pub fn with_slingshot(
170
+
client: SlingshotClient,
171
+
clock: Arc<dyn Clock>,
172
+
hasher: RuntimeHasher,
173
+
) -> Self {
71
174
Self {
72
175
cache: SccMap::with_hasher(hasher),
73
-
client: Some(client),
176
+
probe: Some(SlingshotProbe { client, clock }),
177
+
stats: ResolverStats::default(),
74
178
}
75
179
}
76
180
77
181
pub fn detached(hasher: RuntimeHasher) -> Self {
78
182
Self {
79
183
cache: SccMap::with_hasher(hasher),
80
-
client: None,
184
+
probe: None,
185
+
stats: ResolverStats::default(),
81
186
}
187
+
}
188
+
189
+
pub fn stats(&self) -> ResolverStatsSnapshot {
190
+
self.stats.snapshot()
82
191
}
83
192
84
193
pub async fn observe(
···
112
221
pub async fn resolve(&self, owner: &Did<DefaultStr>, rkey: &Rkey<DefaultStr>) -> Resolution {
113
222
let key = RepoRef::new(owner.clone(), rkey.clone());
114
223
if let Some(entry) = self.cache.get_async(&key).await {
224
+
self.stats.record_hit();
115
225
return entry.get().clone().into_resolution();
116
226
}
117
-
let Some(client) = self.client.as_ref() else {
227
+
let Some(probe) = self.probe.as_ref() else {
228
+
self.stats.record_miss(MissKind::NoClient, None);
118
229
return Resolution::Unresolvable;
119
230
};
231
+
let started = probe.clock.now_instant();
120
232
let nsid: Nsid<DefaultStr> = nsid_static(REPO_COLLECTION);
121
-
let provisional = match client.get_record(owner, &nsid, rkey).await {
233
+
let provisional = match probe.client.get_record(owner, &nsid, rkey).await {
122
234
Ok(body) => match repo_did_from_body(&body.value) {
123
235
Ok(Some(did)) => Resolution::Mapped(did),
124
236
Ok(None) => Resolution::NoRepoDid,
···
156
268
rkey = rkey.as_ref(),
157
269
"slingshot transient failure during repoDID lookup, will retry",
158
270
);
271
+
let elapsed = probe.clock.now_instant().duration_since(started);
272
+
self.stats
273
+
.record_miss(MissKind::Transient, Some(elapsed));
159
274
return Resolution::Unresolvable;
160
275
}
161
276
};
277
+
let elapsed = probe.clock.now_instant().duration_since(started);
278
+
let kind = match &provisional {
279
+
Resolution::Mapped(_) => MissKind::Mapped,
280
+
Resolution::NoRepoDid => MissKind::NoRepoDid,
281
+
Resolution::Unresolvable => MissKind::Unresolvable,
282
+
};
283
+
self.stats.record_miss(kind, Some(elapsed));
162
284
self.fill_provisional(key, provisional.clone()).await;
163
285
provisional
164
286
}
···
185
307
#[cfg(test)]
186
308
mod tests {
187
309
use super::*;
310
+
use bobbin_runtime::SystemClock;
188
311
use jacquard_common::types::did::Did;
189
312
use jacquard_common::types::recordkey::Rkey;
190
313
···
196
319
Rkey::new_owned(s).unwrap()
197
320
}
198
321
322
+
fn test_clock() -> Arc<dyn Clock> {
323
+
Arc::new(SystemClock::new())
324
+
}
325
+
199
326
#[tokio::test]
200
327
async fn observation_with_repo_did_resolves_mapped() {
201
328
let resolver = RepoIdResolver::detached(RuntimeHasher::default());
···
314
441
315
442
let client =
316
443
SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap();
317
-
let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default());
444
+
let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
318
445
319
446
let owner = did("did:plc:nel");
320
447
let key = rkey("abcabcabcabcz");
···
340
467
341
468
let client =
342
469
SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap();
343
-
let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default());
470
+
let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
344
471
345
472
let owner = did("did:plc:nel");
346
473
let key = rkey("abcabcabcabcz");
···
371
498
372
499
let client =
373
500
SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap();
374
-
let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default());
501
+
let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
375
502
376
503
let owner = did("did:plc:nel");
377
504
let key = rkey("abcabcabcabcz");
···
398
525
399
526
let client =
400
527
SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap();
401
-
let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default());
528
+
let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
402
529
403
530
let owner = did("did:plc:nel");
404
531
let key = rkey("abcabcabcabcz");
···
424
551
425
552
let client =
426
553
SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap();
427
-
let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default());
554
+
let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
428
555
429
556
let owner = did("did:plc:nel");
430
557
let key = rkey("abcabcabcabcz");
···
432
559
let second = resolver.resolve(&owner, &key).await;
433
560
assert_eq!(first, Resolution::Unresolvable);
434
561
assert_eq!(second, Resolution::Unresolvable);
562
+
}
563
+
564
+
#[tokio::test]
565
+
async fn slingshot_transient_recorded_separately_from_unresolvable() {
566
+
let server = wiremock::MockServer::start().await;
567
+
wiremock::Mock::given(wiremock::matchers::method("GET"))
568
+
.and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord"))
569
+
.respond_with(wiremock::ResponseTemplate::new(503))
570
+
.expect(2)
571
+
.mount(&server)
572
+
.await;
573
+
574
+
let client =
575
+
SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap();
576
+
let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
577
+
578
+
let owner = did("did:plc:nel");
579
+
let key = rkey("abcabcabcabcz");
580
+
resolver.resolve(&owner, &key).await;
581
+
resolver.resolve(&owner, &key).await;
582
+
583
+
let snap = resolver.stats();
584
+
assert_eq!(
585
+
snap.misses_transient, 2,
586
+
"transport-error retries must count as transient, not as canonical unresolvable",
587
+
);
588
+
assert_eq!(
589
+
snap.misses_unresolvable, 0,
590
+
"canonical unresolvable counter is reserved for cached terminal answers",
591
+
);
592
+
assert!(
593
+
snap.miss_latency_micros_sum > 0,
594
+
"transient misses still have latency contributions",
595
+
);
596
+
assert_eq!(snap.miss_count(), 2);
597
+
}
598
+
599
+
#[tokio::test]
600
+
async fn stats_count_hits_misses_and_latency() {
601
+
let server = wiremock::MockServer::start().await;
602
+
wiremock::Mock::given(wiremock::matchers::method("GET"))
603
+
.and(wiremock::matchers::path("/xrpc/com.atproto.repo.getRecord"))
604
+
.respond_with(wiremock::ResponseTemplate::new(404))
605
+
.mount(&server)
606
+
.await;
607
+
let client =
608
+
SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap();
609
+
let resolver = RepoIdResolver::with_slingshot(client, test_clock(), RuntimeHasher::default());
610
+
611
+
let owner = did("did:plc:nel");
612
+
let key = rkey("abcabcabcabcz");
613
+
resolver.resolve(&owner, &key).await;
614
+
resolver.resolve(&owner, &key).await;
615
+
616
+
let snap = resolver.stats();
617
+
assert_eq!(snap.misses_unresolvable, 1, "first call is the slingshot miss");
618
+
assert_eq!(snap.hits, 1, "second call hits the unresolvable cache");
619
+
assert_eq!(snap.miss_count(), 1);
620
+
assert_eq!(snap.total(), 2);
621
+
assert!(snap.miss_latency_micros_sum > 0, "latency recorded for slingshot miss");
622
+
assert!(snap.miss_latency_micros_avg().unwrap() > 0);
623
+
}
624
+
625
+
#[tokio::test]
626
+
async fn stats_no_client_miss_recorded_without_latency() {
627
+
let resolver = RepoIdResolver::detached(RuntimeHasher::default());
628
+
resolver
629
+
.resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz"))
630
+
.await;
631
+
let snap = resolver.stats();
632
+
assert_eq!(snap.misses_no_client, 1);
633
+
assert_eq!(snap.miss_latency_micros_sum, 0);
634
+
assert_eq!(snap.miss_latency_micros_avg(), None);
435
635
}
436
636
437
637
#[tokio::test]
+1
-1
crates/runtime/src/entropy.rs
+1
-1
crates/runtime/src/entropy.rs
+1
-1
crates/xrpc/tests/aggregation.rs
+1
-1
crates/xrpc/tests/aggregation.rs
···
1188
1188
assert_eq!(
1189
1189
status,
1190
1190
StatusCode::OK,
1191
-
"extractor key must match handler subject; body: {json}",
1191
+
"extractor key must match handler subject, body was {json}",
1192
1192
);
1193
1193
let items = json["items"].as_array().unwrap();
1194
1194
assert_eq!(items.len(), 1, "expected exactly one star edge");
+3
-3
crates/xrpc/tests/extended.rs
+3
-3
crates/xrpc/tests/extended.rs
···
719
719
assert_eq!(
720
720
status,
721
721
StatusCode::OK,
722
-
"extractor key must match handler subject; body: {json}"
722
+
"extractor key must match handler subject, body was {json}"
723
723
);
724
724
let items = json["items"].as_array().unwrap();
725
725
assert_eq!(items.len(), 1, "expected exactly one pipeline edge");
···
758
758
assert_eq!(
759
759
status,
760
760
StatusCode::OK,
761
-
"owner-DID fallback must hydrate; body: {json}"
761
+
"owner-DID fallback must hydrate, body was {json}"
762
762
);
763
763
let items = json["items"].as_array().unwrap();
764
764
assert_eq!(items.len(), 1);
···
770
770
items[0]["value"]["triggerMetadata"]["repo"]
771
771
.get("repoDid")
772
772
.is_none(),
773
-
"fixture must omit repoDid; got {}",
773
+
"fixture must omit repoDid but got {}",
774
774
items[0]["value"]["triggerMetadata"]["repo"]
775
775
);
776
776
}