feat(xrpc): issue & pull state, comment count, author filter · ptr.pet/bobbin@8fff662

+198 -13

crates/xrpc/src/lib.rs

··· 20 20 routing::get, 21 21 }; 22 22 use bobbin_edge_index::{ 23 - Coverage, CoverageWatch, CursorParseError, EdgePage, EdgeStore, PageCursor, PageLimit, 24 - PageToken, 23 + Coverage, CoverageWatch, CursorParseError, EdgePage, EdgeStore, IssueStateKind, PageCursor, 24 + PageLimit, PageToken, PullStatusKind, StateIndex, StateKind, 25 25 }; 26 26 use bobbin_knot_proxy::{KnotHost, KnotProxy, KnotProxyError, ProxyResponse, RepoSlug}; 27 27 use bobbin_record_lru::RecordStore; ··· 98 98 pub records: Arc<dyn RecordStore>, 99 99 pub slingshot: SlingshotClient, 100 100 pub edges: Arc<EdgeStore>, 101 + pub issue_states: Arc<StateIndex<IssueStateKind>>, 102 + pub pull_statuses: Arc<StateIndex<PullStatusKind>>, 101 103 pub coverage: Arc<CoverageWatch>, 102 104 pub knots: Arc<KnotProxy>, 103 105 pub search: Arc<dyn SearchReader>, ··· 105 107 } 106 108 107 109 impl AppState { 110 + #[allow(clippy::too_many_arguments)] 108 111 pub fn new( 109 112 records: Arc<dyn RecordStore>, 110 113 slingshot: SlingshotClient, 111 114 edges: Arc<EdgeStore>, 115 + issue_states: Arc<StateIndex<IssueStateKind>>, 116 + pull_statuses: Arc<StateIndex<PullStatusKind>>, 112 117 coverage: Arc<CoverageWatch>, 113 118 knots: Arc<KnotProxy>, 114 119 search: Arc<dyn SearchReader>, ··· 118 123 records, 119 124 slingshot, 120 125 edges, 126 + issue_states, 127 + pull_statuses, 121 128 coverage, 122 129 knots, 123 130 search, ··· 539 546 } 540 547 541 548 #[derive(Debug, Deserialize)] 549 + struct IssueListQuery { 550 + subject: RawAtUriParam, 551 + cursor: Option<String>, 552 + limit: Option<u32>, 553 + author: Option<String>, 554 + } 555 + 556 + impl IssueListQuery { 557 + fn split(self) -> (ListQuery, Option<String>) { 558 + ( 559 + ListQuery { 560 + subject: self.subject, 561 + cursor: self.cursor, 562 + limit: self.limit, 563 + }, 564 + self.author, 565 + ) 566 + } 567 + } 568 + 569 + #[derive(Debug, Deserialize)] 542 570 struct CountQuery { 543 571 subject: RawAtUriParam, 544 572 } ··· 653 681 654 682 #[derive(Serialize)] 655 683 #[serde(rename_all = "camelCase")] 684 + struct StatefulItem<V> { 685 + #[serde(flatten)] 686 + view: RecordView<V>, 687 + #[serde(skip_serializing_if = "Option::is_none")] 688 + state: Option<&'static str>, 689 + #[serde(skip_serializing_if = "Option::is_none")] 690 + state_updated_at: Option<String>, 691 + comment_count: u64, 692 + } 693 + 694 + #[derive(Serialize)] 695 + #[serde(rename_all = "camelCase")] 696 + struct StatefulListResponse<V> { 697 + items: Vec<StatefulItem<V>>, 698 + cursor: Option<String>, 699 + } 700 + 701 + fn format_micros(micros: u64) -> String { 702 + let signed = i64::try_from(micros).ok(); 703 + let rfc = signed 704 + .and_then(chrono::DateTime::<chrono::Utc>::from_timestamp_micros) 705 + .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)); 706 + rfc.unwrap_or_else(|| micros.to_string()) 707 + } 708 + 709 + fn source_did_prefix(source: &AtUri<DefaultStr>) -> Option<&str> { 710 + source 711 + .as_str() 712 + .strip_prefix("at://") 713 + .and_then(|rest| rest.split('/').next()) 714 + .filter(|s| s.starts_with("did:")) 715 + } 716 + 717 + fn enrich_issue_list( 718 + state: &AppState, 719 + list: ListResponse<Issue<DefaultStr>>, 720 + ) -> StatefulListResponse<Issue<DefaultStr>> { 721 + let items = list 722 + .items 723 + .into_iter() 724 + .map(|v| { 725 + let issue_author = source_did_prefix(&v.uri).map(str::to_owned); 726 + let repo_did = v.value.repo.as_str().to_owned(); 727 + enrich_view( 728 + &state.edges, 729 + "sh.tangled.repo.issue.comment", 730 + &state.issue_states, 731 + v, 732 + move |src| accept_state_source(src, issue_author.as_deref(), &repo_did), 733 + ) 734 + }) 735 + .collect(); 736 + StatefulListResponse { 737 + items, 738 + cursor: list.cursor, 739 + } 740 + } 741 + 742 + fn enrich_pull_list( 743 + state: &AppState, 744 + list: ListResponse<Pull<DefaultStr>>, 745 + ) -> StatefulListResponse<Pull<DefaultStr>> { 746 + let items = list 747 + .items 748 + .into_iter() 749 + .map(|v| { 750 + let pull_author = source_did_prefix(&v.uri).map(str::to_owned); 751 + let target_repo = v.value.target.repo.as_str().to_owned(); 752 + enrich_view( 753 + &state.edges, 754 + "sh.tangled.repo.pull.comment", 755 + &state.pull_statuses, 756 + v, 757 + move |src| accept_state_source(src, pull_author.as_deref(), &target_repo), 758 + ) 759 + }) 760 + .collect(); 761 + StatefulListResponse { 762 + items, 763 + cursor: list.cursor, 764 + } 765 + } 766 + 767 + fn accept_state_source( 768 + source: &AtUri<DefaultStr>, 769 + entity_author: Option<&str>, 770 + repo_owner: &str, 771 + ) -> bool { 772 + let Some(src) = source_did_prefix(source) else { 773 + return false; 774 + }; 775 + Some(src) == entity_author || src == repo_owner 776 + } 777 + 778 + fn enrich_view<V, K, F>( 779 + edges: &EdgeStore, 780 + comment_nsid: &'static str, 781 + states: &StateIndex<K>, 782 + view: RecordView<V>, 783 + accept: F, 784 + ) -> StatefulItem<V> 785 + where 786 + K: StateKind, 787 + F: Fn(&AtUri<DefaultStr>) -> bool, 788 + { 789 + let comment_count = edges.count(&EdgeKey::new( 790 + nsid_static(comment_nsid), 791 + SubjectRef::Uri(view.uri.clone()), 792 + )); 793 + let (state, state_updated_at) = states 794 + .latest_by(&view.uri, accept) 795 + .map_or((None, None), |(kind, micros)| { 796 + (Some(kind.wire()), Some(format_micros(micros))) 797 + }); 798 + StatefulItem { 799 + view, 800 + state, 801 + state_updated_at, 802 + comment_count, 803 + } 804 + } 805 + 806 + #[derive(Serialize)] 807 + #[serde(rename_all = "camelCase")] 656 808 struct CountResponse { 657 809 count: u64, 658 810 distinct_authors: u64, ··· 978 1130 .map_err(|e| XrpcError::InvalidParams(format!("limit: {e}"))) 979 1131 } 980 1132 1133 + fn parse_author_prefix(raw: Option<&str>) -> Result<Option<String>, XrpcError> { 1134 + raw.map(|s| { 1135 + Did::<DefaultStr>::new_owned(s) 1136 + .map(|did| format!("at://{}/", did.as_ref())) 1137 + .map_err(|e| XrpcError::InvalidParams(format!("author: {e}"))) 1138 + }) 1139 + .transpose() 1140 + } 1141 + 981 1142 async fn resolve_for_view( 982 1143 state: &AppState, 983 1144 expected_nsid: &str, ··· 1281 1442 R: XrpcResp + HasSubject, 1282 1443 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs, 1283 1444 { 1445 + list_records_with_author::<R, V>(state, q, None).await 1446 + } 1447 + 1448 + async fn list_records_with_author<R, V>( 1449 + state: &AppState, 1450 + q: ListQuery, 1451 + author: Option<&str>, 1452 + ) -> Result<ListResponse<V>, XrpcError> 1453 + where 1454 + R: XrpcResp + HasSubject, 1455 + V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs, 1456 + { 1284 1457 let subject = parse_subject(&q.subject, R::SHAPE)?; 1285 1458 let cursor = parse_cursor(q.cursor.as_deref())?; 1286 1459 let limit = parse_limit(q.limit)?; 1460 + let author_prefix = parse_author_prefix(author)?; 1287 1461 let key = EdgeKey::new(nsid_static(R::NSID), subject); 1288 - let EdgePage { items, next } = state.edges.list(&key, cursor, limit); 1462 + let EdgePage { items, next } = match author_prefix.as_deref() { 1463 + Some(prefix) => state 1464 + .edges 1465 + .list_filtered(&key, cursor, limit, |uri| uri.as_ref().starts_with(prefix)), 1466 + None => state.edges.list(&key, cursor, limit), 1467 + }; 1289 1468 let items = stream::iter(items) 1290 1469 .map(|uri| async move { 1291 1470 let body = resolve_for_view(state, R::NSID, uri).await?; ··· 1400 1579 1401 1580 async fn list_issues( 1402 1581 State(state): State<AppState>, 1403 - XrpcQuery(q): XrpcQuery<ListQuery>, 1404 - ) -> Result<Json<ListResponse<Issue<DefaultStr>>>, XrpcError> { 1405 - list_records::<IssueRecord, _>(&state, q).await.map(Json) 1582 + XrpcQuery(q): XrpcQuery<IssueListQuery>, 1583 + ) -> Result<Json<StatefulListResponse<Issue<DefaultStr>>>, XrpcError> { 1584 + let (base, author) = q.split(); 1585 + let list = list_records_with_author::<IssueRecord, _>(&state, base, author.as_deref()).await?; 1586 + Ok(Json(enrich_issue_list(&state, list))) 1406 1587 } 1407 1588 1408 1589 async fn count_issues( ··· 1414 1595 1415 1596 async fn list_pulls( 1416 1597 State(state): State<AppState>, 1417 - XrpcQuery(q): XrpcQuery<ListQuery>, 1418 - ) -> Result<Json<ListResponse<Pull<DefaultStr>>>, XrpcError> { 1419 - list_records::<PullRecord, _>(&state, q).await.map(Json) 1598 + XrpcQuery(q): XrpcQuery<IssueListQuery>, 1599 + ) -> Result<Json<StatefulListResponse<Pull<DefaultStr>>>, XrpcError> { 1600 + let (base, author) = q.split(); 1601 + let list = list_records_with_author::<PullRecord, _>(&state, base, author.as_deref()).await?; 1602 + Ok(Json(enrich_pull_list(&state, list))) 1420 1603 } 1421 1604 1422 1605 async fn count_pulls( ··· 1756 1939 async fn list_issues_by( 1757 1940 State(state): State<AppState>, 1758 1941 XrpcQuery(q): XrpcQuery<ListQuery>, 1759 - ) -> Result<Json<ListResponse<Issue<DefaultStr>>>, XrpcError> { 1760 - list_mirror::<IssueBy, _>(&state, q).await.map(Json) 1942 + ) -> Result<Json<StatefulListResponse<Issue<DefaultStr>>>, XrpcError> { 1943 + let list = list_mirror::<IssueBy, _>(&state, q).await?; 1944 + Ok(Json(enrich_issue_list(&state, list))) 1761 1945 } 1762 1946 async fn count_issues_by( 1763 1947 State(state): State<AppState>, ··· 1795 1979 async fn list_pulls_by( 1796 1980 State(state): State<AppState>, 1797 1981 XrpcQuery(q): XrpcQuery<ListQuery>, 1798 - ) -> Result<Json<ListResponse<Pull<DefaultStr>>>, XrpcError> { 1799 - list_mirror::<PullBy, _>(&state, q).await.map(Json) 1982 + ) -> Result<Json<StatefulListResponse<Pull<DefaultStr>>>, XrpcError> { 1983 + let list = list_mirror::<PullBy, _>(&state, q).await?; 1984 + Ok(Json(enrich_pull_list(&state, list))) 1800 1985 } 1801 1986 async fn count_pulls_by( 1802 1987 State(state): State<AppState>,

+220 -1

crates/xrpc/tests/aggregation.rs

··· 1 1 use std::sync::Arc; 2 2 3 3 use axum::body::{Body, to_bytes}; 4 - use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, PageToken}; 4 + use bobbin_edge_index::{ 5 + Coverage, CoverageWatch, EdgeStore, HydrantCursor, IssueStateKind, PageToken, PullStatusKind, 6 + StateIndex, 7 + }; 5 8 use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; 6 9 use bobbin_record_lru::{CacheCapacity, LruRecordStore}; 7 10 use bobbin_resolver::RepoIdResolver; ··· 57 60 async fn new() -> Self { 58 61 let server = MockServer::start().await; 59 62 let edges = Arc::new(EdgeStore::new(RuntimeHasher::default())); 63 + let issue_states = Arc::new(StateIndex::new(RuntimeHasher::default())); 64 + let pull_statuses = Arc::new(StateIndex::new(RuntimeHasher::default())); 60 65 let coverage = Arc::new(CoverageWatch::new()); 61 66 let state = AppState::new( 62 67 Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), 63 68 SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), 64 69 edges.clone(), 70 + issue_states.clone(), 71 + pull_statuses.clone(), 65 72 coverage.clone(), 66 73 Arc::new( 67 74 KnotProxy::new( ··· 153 160 "repo": repo_did, 154 161 "title": title, 155 162 "createdAt": "2026-05-01T00:00:00Z" 163 + }) 164 + } 165 + 166 + fn pull_body(repo_did: &str, title: &str) -> Value { 167 + json!({ 168 + "$type": "sh.tangled.repo.pull", 169 + "title": title, 170 + "createdAt": "2026-05-01T00:00:00Z", 171 + "rounds": [], 172 + "target": { 173 + "repo": repo_did, 174 + "branch": "main" 175 + } 156 176 }) 157 177 } 158 178 ··· 1159 1179 assert_eq!(items.len(), 1, "expected exactly one star edge"); 1160 1180 assert_eq!(items[0]["value"]["subject"]["did"], json!(subject_did)); 1161 1181 } 1182 + 1183 + #[tokio::test] 1184 + async fn list_issues_includes_state_comment_count_and_state_updated_at() { 1185 + let h = Harness::new().await; 1186 + let repo = "did:plc:limpet"; 1187 + let subject = format!("at://{repo}"); 1188 + let issue_uri_str = "at://did:plc:nel/sh.tangled.repo.issue/i1".to_owned(); 1189 + h.add_edge("sh.tangled.repo.issue", &subject, &issue_uri_str); 1190 + h.mount( 1191 + "did:plc:nel", 1192 + "sh.tangled.repo.issue", 1193 + "i1", 1194 + issue_body(repo, "hi"), 1195 + ) 1196 + .await; 1197 + h.add_edge( 1198 + "sh.tangled.repo.issue.comment", 1199 + &issue_uri_str, 1200 + "at://did:plc:olaren/sh.tangled.repo.issue.comment/c1", 1201 + ); 1202 + h.add_edge( 1203 + "sh.tangled.repo.issue.comment", 1204 + &issue_uri_str, 1205 + "at://did:plc:teq/sh.tangled.repo.issue.comment/c2", 1206 + ); 1207 + 1208 + let issue_uri = at(&issue_uri_str); 1209 + h.state.issue_states.upsert( 1210 + at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"), 1211 + issue_uri.clone(), 1212 + 1_777_593_600_000_000, 1213 + IssueStateKind::Open, 1214 + ); 1215 + h.state.issue_states.upsert( 1216 + at("at://did:plc:nel/sh.tangled.repo.issue.state/s2"), 1217 + issue_uri.clone(), 1218 + 1_777_593_700_000_000, 1219 + IssueStateKind::Closed, 1220 + ); 1221 + 1222 + let app = router(h.state.clone()); 1223 + let resp = app 1224 + .oneshot(list_request("sh.tangled.repo.listIssues", &subject, &[])) 1225 + .await 1226 + .unwrap(); 1227 + let (status, body) = json_response(resp).await; 1228 + assert_eq!(status, StatusCode::OK); 1229 + let item = &body["items"][0]; 1230 + assert_eq!(item["state"], json!("closed")); 1231 + assert_eq!(item["commentCount"], json!(2)); 1232 + let updated = item["stateUpdatedAt"] 1233 + .as_str() 1234 + .expect("stateUpdatedAt must serialize as RFC3339 string"); 1235 + assert!( 1236 + updated.starts_with("2026-"), 1237 + "expected 2026 timestamp, got {updated}" 1238 + ); 1239 + } 1240 + 1241 + #[tokio::test] 1242 + async fn list_issues_omits_state_when_no_state_record() { 1243 + let h = Harness::new().await; 1244 + let repo = "did:plc:limpet"; 1245 + let subject = format!("at://{repo}"); 1246 + let issue_uri_str = "at://did:plc:nel/sh.tangled.repo.issue/i1".to_owned(); 1247 + h.add_edge("sh.tangled.repo.issue", &subject, &issue_uri_str); 1248 + h.mount( 1249 + "did:plc:nel", 1250 + "sh.tangled.repo.issue", 1251 + "i1", 1252 + issue_body(repo, "no state yet"), 1253 + ) 1254 + .await; 1255 + 1256 + let app = router(h.state.clone()); 1257 + let (_status, body) = 1258 + json_response(app.oneshot(list_request("sh.tangled.repo.listIssues", &subject, &[])).await.unwrap()).await; 1259 + let item = &body["items"][0]; 1260 + assert!(item.get("state").is_none(), "state must be absent"); 1261 + assert!( 1262 + item.get("stateUpdatedAt").is_none(), 1263 + "stateUpdatedAt must be absent" 1264 + ); 1265 + assert_eq!(item["commentCount"], json!(0)); 1266 + } 1267 + 1268 + #[tokio::test] 1269 + async fn list_issues_author_filter_restricts_to_matching_did() { 1270 + let h = Harness::new().await; 1271 + let repo = "did:plc:limpet"; 1272 + let subject = format!("at://{repo}"); 1273 + let owners = [ 1274 + ("did:plc:nel", "n1"), 1275 + ("did:plc:nel", "n2"), 1276 + ("did:plc:olaren", "o1"), 1277 + ("did:plc:olaren", "o2"), 1278 + ]; 1279 + stream::iter(owners) 1280 + .for_each(|(did, rkey)| { 1281 + let h = &h; 1282 + let subject = subject.clone(); 1283 + async move { 1284 + h.add_edge( 1285 + "sh.tangled.repo.issue", 1286 + &subject, 1287 + &format!("at://{did}/sh.tangled.repo.issue/{rkey}"), 1288 + ); 1289 + h.mount( 1290 + did, 1291 + "sh.tangled.repo.issue", 1292 + rkey, 1293 + issue_body(repo, &format!("issue-{rkey}")), 1294 + ) 1295 + .await; 1296 + } 1297 + }) 1298 + .await; 1299 + 1300 + let app = router(h.state.clone()); 1301 + let (status, body) = json_response( 1302 + app.oneshot(list_request( 1303 + "sh.tangled.repo.listIssues", 1304 + &subject, 1305 + &[("author", "did:plc:nel")], 1306 + )) 1307 + .await 1308 + .unwrap(), 1309 + ) 1310 + .await; 1311 + assert_eq!(status, StatusCode::OK); 1312 + let items = body["items"].as_array().expect("items array"); 1313 + assert_eq!(items.len(), 2, "two issues authored by nel"); 1314 + let all_nel = items 1315 + .iter() 1316 + .all(|i| i["uri"].as_str().unwrap().starts_with("at://did:plc:nel/")); 1317 + assert!(all_nel, "every returned uri must be authored by nel"); 1318 + } 1319 + 1320 + #[tokio::test] 1321 + async fn list_issues_invalid_author_returns_400() { 1322 + let h = Harness::new().await; 1323 + let subject = "at://did:plc:limpet".to_owned(); 1324 + let app = router(h.state.clone()); 1325 + let resp = app 1326 + .oneshot(list_request( 1327 + "sh.tangled.repo.listIssues", 1328 + &subject, 1329 + &[("author", "not-a-did")], 1330 + )) 1331 + .await 1332 + .unwrap(); 1333 + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); 1334 + } 1335 + 1336 + #[tokio::test] 1337 + async fn list_pulls_includes_merged_state_and_comment_count() { 1338 + let h = Harness::new().await; 1339 + let repo = "did:plc:limpet"; 1340 + let subject = format!("at://{repo}"); 1341 + let pull_uri_str = "at://did:plc:nel/sh.tangled.repo.pull/p1".to_owned(); 1342 + h.add_edge("sh.tangled.repo.pull", &subject, &pull_uri_str); 1343 + h.mount( 1344 + "did:plc:nel", 1345 + "sh.tangled.repo.pull", 1346 + "p1", 1347 + pull_body(repo, "fix bug"), 1348 + ) 1349 + .await; 1350 + h.add_edge( 1351 + "sh.tangled.repo.pull.comment", 1352 + &pull_uri_str, 1353 + "at://did:plc:teq/sh.tangled.repo.pull.comment/c1", 1354 + ); 1355 + let pull_uri = at(&pull_uri_str); 1356 + h.state.pull_statuses.upsert( 1357 + at("at://did:plc:nel/sh.tangled.repo.pull.status/s1"), 1358 + pull_uri.clone(), 1359 + 1_777_593_600_000_000, 1360 + PullStatusKind::Open, 1361 + ); 1362 + h.state.pull_statuses.upsert( 1363 + at("at://did:plc:nel/sh.tangled.repo.pull.status/s2"), 1364 + pull_uri.clone(), 1365 + 1_777_593_800_000_000, 1366 + PullStatusKind::Merged, 1367 + ); 1368 + 1369 + let app = router(h.state.clone()); 1370 + let (status, body) = json_response( 1371 + app.oneshot(list_request("sh.tangled.repo.listPulls", &subject, &[])) 1372 + .await 1373 + .unwrap(), 1374 + ) 1375 + .await; 1376 + assert_eq!(status, StatusCode::OK); 1377 + let item = &body["items"][0]; 1378 + assert_eq!(item["state"], json!("merged")); 1379 + assert_eq!(item["commentCount"], json!(1)); 1380 + }

+3 -1

crates/xrpc/tests/bulk.rs

··· 1 1 use std::sync::Arc; 2 2 3 3 use axum::body::{Body, to_bytes}; 4 - use bobbin_edge_index::{CoverageWatch, EdgeStore}; 4 + use bobbin_edge_index::{CoverageWatch, EdgeStore, StateIndex}; 5 5 use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; 6 6 use bobbin_record_lru::{CacheCapacity, LruRecordStore}; 7 7 use bobbin_resolver::RepoIdResolver; ··· 32 32 Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), 33 33 SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), 34 34 Arc::new(EdgeStore::new(RuntimeHasher::default())), 35 + Arc::new(StateIndex::new(RuntimeHasher::default())), 36 + Arc::new(StateIndex::new(RuntimeHasher::default())), 35 37 coverage, 36 38 Arc::new( 37 39 KnotProxy::new(

+3 -1

crates/xrpc/tests/cold_start.rs

··· 1 1 use std::sync::Arc; 2 2 3 3 use axum::body::{Body, to_bytes}; 4 - use bobbin_edge_index::{CoverageWatch, EdgeStore}; 4 + use bobbin_edge_index::{CoverageWatch, EdgeStore, StateIndex}; 5 5 use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; 6 6 use bobbin_record_lru::{CacheCapacity, LruRecordStore}; 7 7 use bobbin_resolver::RepoIdResolver; ··· 28 28 Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), 29 29 SlingshotClient::with_default_http(Url::parse(server_uri).unwrap()).unwrap(), 30 30 Arc::new(EdgeStore::new(RuntimeHasher::default())), 31 + Arc::new(StateIndex::new(RuntimeHasher::default())), 32 + Arc::new(StateIndex::new(RuntimeHasher::default())), 31 33 Arc::new(CoverageWatch::new()), 32 34 Arc::new( 33 35 KnotProxy::new(

+3 -1

crates/xrpc/tests/coverage.rs

··· 1 1 use std::sync::Arc; 2 2 3 3 use axum::body::{Body, to_bytes}; 4 - use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor}; 4 + use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, StateIndex}; 5 5 use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; 6 6 use bobbin_record_lru::{CacheCapacity, LruRecordStore}; 7 7 use bobbin_resolver::RepoIdResolver; ··· 28 28 Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), 29 29 SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), 30 30 Arc::new(EdgeStore::new(RuntimeHasher::default())), 31 + Arc::new(StateIndex::new(RuntimeHasher::default())), 32 + Arc::new(StateIndex::new(RuntimeHasher::default())), 31 33 coverage.clone(), 32 34 Arc::new( 33 35 KnotProxy::new(

+3 -1

crates/xrpc/tests/extended.rs

··· 1 1 use std::sync::Arc; 2 2 3 3 use axum::body::{Body, to_bytes}; 4 - use bobbin_edge_index::{CoverageWatch, EdgeStore}; 4 + use bobbin_edge_index::{CoverageWatch, EdgeStore, StateIndex}; 5 5 use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; 6 6 use bobbin_record_lru::{CacheCapacity, LruRecordStore}; 7 7 use bobbin_resolver::RepoIdResolver; ··· 56 56 Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), 57 57 SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), 58 58 edges.clone(), 59 + Arc::new(StateIndex::new(RuntimeHasher::default())), 60 + Arc::new(StateIndex::new(RuntimeHasher::default())), 59 61 coverage, 60 62 Arc::new( 61 63 KnotProxy::new(

+3 -1

crates/xrpc/tests/knot_proxy.rs

··· 2 2 use std::time::Duration; 3 3 4 4 use axum::body::{Body, to_bytes}; 5 - use bobbin_edge_index::{CoverageWatch, EdgeStore}; 5 + use bobbin_edge_index::{CoverageWatch, EdgeStore, StateIndex}; 6 6 use bobbin_knot_proxy::{FailureThreshold, KnotHttpConfig, KnotProxy, KnotProxyConfig}; 7 7 use bobbin_record_lru::{CacheCapacity, LruRecordStore}; 8 8 use bobbin_resolver::RepoIdResolver; ··· 55 55 SlingshotClient::with_default_http(Url::parse(&slingshot_server.uri()).unwrap()) 56 56 .unwrap(), 57 57 Arc::new(EdgeStore::new(RuntimeHasher::default())), 58 + Arc::new(StateIndex::new(RuntimeHasher::default())), 59 + Arc::new(StateIndex::new(RuntimeHasher::default())), 58 60 Arc::new(CoverageWatch::new()), 59 61 Arc::new( 60 62 KnotProxy::new(

+3 -1

crates/xrpc/tests/search.rs

··· 1 1 use std::sync::Arc; 2 2 3 3 use axum::body::{Body, to_bytes}; 4 - use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor}; 4 + use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, StateIndex}; 5 5 use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; 6 6 use bobbin_record_lru::{CacheCapacity, LruRecordStore}; 7 7 use bobbin_resolver::RepoIdResolver; ··· 53 53 Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), 54 54 SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), 55 55 Arc::new(EdgeStore::new(RuntimeHasher::default())), 56 + Arc::new(StateIndex::new(RuntimeHasher::default())), 57 + Arc::new(StateIndex::new(RuntimeHasher::default())), 56 58 coverage.clone(), 57 59 Arc::new( 58 60 KnotProxy::new(

Configure Feed

Configure Feed