alice.pds.demo.boltless.dev
This repository has no description
bobbin/xrpc/enrich: use unordered concurrency, deduplicate descriptors and minidoc res, bump minidoc concurrency count
Signed-off-by: dawn <dawn@tangled.org>
author
dawn
date
(Aug 1, 2026, 12:21 PM +0300)
commit
6b4b5877
6b4b587797ec038c4a0393222d7c8c7b4b5f8b3b
parent
733fe419
733fe4199a50b87c390309ea46ba5927585eab81
change-id
zywxnpop
zywxnpopmlpwzynxokllvkzosqvtplrl
+62
-24
+51
-23
bobbin/crates/xrpc/src/enrich.rs
+51
-23
bobbin/crates/xrpc/src/enrich.rs
···
1
-
use std::collections::HashSet;
1
+
use std::collections::{HashMap, HashSet};
2
2
3
3
use axum::{
4
4
Json,
···
29
29
pub const TYPE_MINIDOC: &str = "com.bad-example.identity.miniDoc";
30
30
31
31
const KNOWN_TYPES: [&str; 4] = [TYPE_COUNT, TYPE_DISTINCT_AUTHORS, TYPE_VIEWER, TYPE_MINIDOC];
32
+
const MINIDOC_CONCURRENCY: usize = 32;
32
33
33
34
/// a payload type nsid, with an optional #fragment for lexicon defs. the raw
34
35
/// string is kept because it echoes into the data map as the payload key
···
180
181
State(state): State<AppState>,
181
182
Json(input): Json<EnrichInput>,
182
183
) -> Result<Json<Value>, XrpcError> {
184
+
let mut seen_descriptors = HashSet::new();
185
+
let mut descriptors = Vec::new();
183
186
for descriptor in &input.enrich {
184
187
if !KNOWN_TYPES.contains(&descriptor.ty.as_str()) {
185
188
return Err(descriptor_error(&descriptor.source, "unknown enrich type"));
···
191
194
"viewer payloads require a viewer param",
192
195
));
193
196
}
197
+
if seen_descriptors.insert((&descriptor.source, &descriptor.ty)) {
198
+
descriptors.push(descriptor);
199
+
}
194
200
}
195
201
202
+
let sources = input
203
+
.sources
204
+
.as_ref()
205
+
.map(|sources| {
206
+
sources
207
+
.iter()
208
+
.enumerate()
209
+
.map(|(i, path)| {
210
+
RecordPath::parse(path).map_err(|e| {
211
+
XrpcError::InvalidParams(format!("sources[{i}] {path:?}: {e}"))
212
+
})
213
+
})
214
+
.collect::<Result<Vec<_>, _>>()
215
+
})
216
+
.transpose()?;
217
+
196
218
let inner = run_inner(&state, &input.xrpc, input.params.unwrap_or_default()).await?;
219
+
if descriptors.is_empty() {
220
+
return Ok(Json(json!({ "output": inner, "data": {} })));
221
+
}
197
222
198
223
let mut refs: Vec<SubjectRef> = Vec::new();
199
224
let mut seen: HashSet<SubjectRef> = HashSet::new();
200
-
match &input.sources {
225
+
match &sources {
201
226
Some(sources) => {
202
-
for (i, path) in sources.iter().enumerate() {
203
-
let path = RecordPath::parse(path)
204
-
.map_err(|e| XrpcError::InvalidParams(format!("sources[{i}] {path:?}: {e}")))?;
205
-
for node in walk_path(&path, [&inner]) {
227
+
for path in sources {
228
+
for node in walk_path(path, [&inner]) {
206
229
collect_ref(node, &mut refs, &mut seen);
207
230
}
208
231
}
···
222
245
}
223
246
224
247
let mut data = Map::new();
225
-
let mut minidoc_targets: Vec<(Did<DefaultStr>, LinkSource)> = Vec::new();
226
-
for descriptor in &input.enrich {
248
+
let mut minidoc_targets: HashMap<Did<DefaultStr>, Vec<LinkSource>> = HashMap::new();
249
+
for descriptor in descriptors {
227
250
for reference in &refs {
228
251
let Some(subject) = applicable_subject(descriptor, reference)? else {
229
252
continue;
···
256
279
}
257
280
TYPE_MINIDOC => {
258
281
if let Some(target) = repo_did(&subject) {
259
-
minidoc_targets.push((target, descriptor.source.clone()));
282
+
let sources = minidoc_targets.entry(target).or_default();
283
+
if !sources.contains(&descriptor.source) {
284
+
sources.push(descriptor.source.clone());
285
+
}
260
286
}
261
287
}
262
288
_ => unreachable!("validated up front"),
···
265
291
}
266
292
267
293
let docs = resolve_minidocs(&state, minidoc_targets).await;
268
-
for (target, source, doc) in docs {
269
-
put(
270
-
&mut data,
271
-
target.as_str(),
272
-
source.as_str(),
273
-
TYPE_MINIDOC,
274
-
doc,
275
-
);
294
+
for (target, sources, doc) in docs {
295
+
for source in sources {
296
+
put(
297
+
&mut data,
298
+
target.as_str(),
299
+
source.as_str(),
300
+
TYPE_MINIDOC,
301
+
doc.clone(),
302
+
);
303
+
}
276
304
}
277
305
278
306
Ok(Json(json!({ "output": inner, "data": data })))
···
296
324
/// we drop failures, the client falls back to resolveMiniDoc for misses
297
325
async fn resolve_minidocs(
298
326
state: &AppState,
299
-
targets: Vec<(Did<DefaultStr>, LinkSource)>,
300
-
) -> Vec<(Did<DefaultStr>, LinkSource, Value)> {
327
+
targets: HashMap<Did<DefaultStr>, Vec<LinkSource>>,
328
+
) -> Vec<(Did<DefaultStr>, Vec<LinkSource>, Value)> {
301
329
futures::stream::iter(targets)
302
-
.map(|(did, source)| async move {
330
+
.map(|(did, sources)| async move {
303
331
let doc = state
304
332
.slingshot
305
333
.resolve_mini_doc(&AtIdentifier::Did(did.clone()))
306
334
.await
307
335
.ok()
308
336
.and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok());
309
-
(did, source, doc)
337
+
(did, sources, doc)
310
338
})
311
-
.buffered(crate::FETCH_CONCURRENCY)
312
-
.filter_map(|(did, source, doc)| async move { doc.map(|doc| (did, source, doc)) })
339
+
.buffer_unordered(MINIDOC_CONCURRENCY)
340
+
.filter_map(|(did, sources, doc)| async move { doc.map(|doc| (did, sources, doc)) })
313
341
.collect()
314
342
.await
315
343
}
+11
-1
bobbin/crates/xrpc/tests/enrich.rs
+11
-1
bobbin/crates/xrpc/tests/enrich.rs
···
539
539
"handle": "a.example.com",
540
540
"pds": "https://pds.example.com"
541
541
})))
542
+
.expect(1)
542
543
.mount(&h.server)
543
544
.await;
544
545
Mock::given(method("GET"))
545
546
.and(path("/xrpc/com.bad-example.identity.resolveMiniDoc"))
546
547
.and(query_param("identifier", "did:plc:b"))
547
548
.respond_with(ResponseTemplate::new(404))
549
+
.expect(1)
548
550
.mount(&h.server)
549
551
.await;
550
552
···
553
555
app.oneshot(enrich_request(json!({
554
556
"xrpc": "sh.tangled.graph.listFollows",
555
557
"params": { "subject": owner.as_ref() },
556
-
"enrich": [{ "source": "sh.tangled.graph.follow:.repo", "type": MINIDOC }]
558
+
"enrich": [
559
+
{ "source": "sh.tangled.graph.follow:.repo", "type": MINIDOC },
560
+
{ "source": "sh.tangled.graph.follow:.repo", "type": MINIDOC },
561
+
{ "source": "sh.tangled.feed.star:.repo", "type": MINIDOC }
562
+
]
557
563
})))
558
564
.await
559
565
.unwrap(),
···
563
569
assert_eq!(status, StatusCode::OK, "{body}");
564
570
assert_eq!(
565
571
body["data"]["did:plc:a"]["sh.tangled.graph.follow:.repo"][MINIDOC]["handle"],
572
+
json!("a.example.com")
573
+
);
574
+
assert_eq!(
575
+
body["data"]["did:plc:a"]["sh.tangled.feed.star:.repo"][MINIDOC]["handle"],
566
576
json!("a.example.com")
567
577
);
568
578
// resolution failures are dropped, the client falls back for misses