BERJAYA

+264 -16

4 changed files

Diff Round #1

Expand all

bobbin

crates

resolver

src

lib.rs

xrpc

src

filter.rs

lib.rs

tests

aggregation.rs

bobbin/crates/resolver/src/lib.rs

··· 226 226 .map(|e| e.get().clone()) 227 227 } 228 228 229 + pub fn owner_by_repo_did(&self, repo_did: &Did<DefaultStr>) -> Option<Did<DefaultStr>> { 230 + self.by_repo_did 231 + .read_sync(repo_did, |_, ident| ident.owner.clone()) 232 + } 233 + 229 234 pub async fn observe( 230 235 &self, 231 236 owner: Did<DefaultStr>, ··· 569 574 assert_eq!( 570 575 got, 571 576 Some(RepoIdent::new(did("did:plc:nel"), rkey("abcabcabcabcz"))), 577 + ); 578 + assert_eq!( 579 + resolver.owner_by_repo_did(&did("did:plc:limpet")), 580 + Some(did("did:plc:nel")), 572 581 ); 573 582 } 574 583

+25 -7

bobbin/crates/xrpc/src/filter.rs

··· 54 54 55 55 impl ListFilter for IssueFilter { 56 56 fn predicate(&self, state: &AppState, subject: &SubjectRef) -> FilterPredicate { 57 + let repo_did = subject.as_did().cloned(); 58 + let repo_owner = repo_did 59 + .as_ref() 60 + .and_then(|repo_did| state.resolver.owner_by_repo_did(repo_did)); 57 61 compose_state_filter::<IssueStateKind>( 58 62 self.author.clone(), 59 63 self.state.map(Into::into), 64 + state.edges.clone(), 60 65 state.issue_states.clone(), 61 - subject.as_did().cloned(), 66 + repo_did, 67 + repo_owner, 62 68 ) 63 69 } 64 70 ··· 93 99 94 100 impl ListFilter for PullFilter { 95 101 fn predicate(&self, state: &AppState, subject: &SubjectRef) -> FilterPredicate { 102 + let repo_did = subject.as_did().cloned(); 103 + let repo_owner = repo_did 104 + .as_ref() 105 + .and_then(|repo_did| state.resolver.owner_by_repo_did(repo_did)); 96 106 compose_state_filter::<PullStatusKind>( 97 107 self.author.clone(), 98 108 self.status.map(Into::into), 109 + state.edges.clone(), 99 110 state.pull_statuses.clone(), 100 - subject.as_did().cloned(), 111 + repo_did, 112 + repo_owner, 101 113 ) 102 114 } 103 115 ··· 109 121 fn compose_state_filter<K>( 110 122 author: Option<Did<DefaultStr>>, 111 123 want: Option<K>, 124 + edges: Arc<bobbin_edge_index::EdgeStore>, 112 125 index: Arc<StateIndex<K>>, 126 + repo_did: Option<Did<DefaultStr>>, 113 127 repo_owner: Option<Did<DefaultStr>>, 114 128 ) -> FilterPredicate 115 129 where ··· 124 138 let Some(want) = want else { 125 139 return true; 126 140 }; 127 - let Some(repo_owner) = repo_owner.as_ref() else { 128 - return false; 129 - }; 130 141 let entity_author = source_authority_did(uri); 131 - let accept = 132 - |src: &AtUri<DefaultStr>| accept_state_source(src, entity_author.as_ref(), repo_owner); 142 + let accept = |src: &AtUri<DefaultStr>| { 143 + accept_state_source( 144 + &edges, 145 + src, 146 + entity_author.as_ref(), 147 + repo_did.as_ref(), 148 + repo_owner.as_ref(), 149 + ) 150 + }; 133 151 let effective = index 134 152 .latest_by(uri, accept) 135 153 .map(|(k, _)| k)

+59 -5

bobbin/crates/xrpc/src/lib.rs

··· 805 805 ) -> StatefulItem<Issue<DefaultStr>> { 806 806 let issue_author = source_authority_did(&view.uri); 807 807 let repo_did = view.value.repo.clone(); 808 + let repo_owner = state.resolver.owner_by_repo_did(&repo_did); 808 809 enrich_view( 809 810 &state.edges, 810 811 nsid_static("sh.tangled.feed.comment"), 811 812 &state.issue_states, 812 813 view, 813 - move |src| accept_state_source(src, issue_author.as_ref(), &repo_did), 814 + move |src| { 815 + accept_state_source( 816 + &state.edges, 817 + src, 818 + issue_author.as_ref(), 819 + Some(&repo_did), 820 + repo_owner.as_ref(), 821 + ) 822 + }, 814 823 ) 815 824 } 816 825 ··· 819 828 view: RecordView<Pull<DefaultStr>>, 820 829 ) -> StatefulItem<Pull<DefaultStr>> { 821 830 let pull_author = source_authority_did(&view.uri); 822 - let target_repo = view.value.target.repo.clone(); 831 + let target_repo_did = view.value.target.repo.clone(); 832 + let repo_owner = state.resolver.owner_by_repo_did(&target_repo_did); 823 833 enrich_view( 824 834 &state.edges, 825 835 nsid_static("sh.tangled.feed.comment"), 826 836 &state.pull_statuses, 827 837 view, 828 - move |src| accept_state_source(src, pull_author.as_ref(), &target_repo), 838 + move |src| { 839 + accept_state_source( 840 + &state.edges, 841 + src, 842 + pull_author.as_ref(), 843 + Some(&target_repo_did), 844 + repo_owner.as_ref(), 845 + ) 846 + }, 829 847 ) 830 848 } 831 849 832 850 pub(crate) fn accept_state_source( 851 + edges: &EdgeStore, 833 852 source: &AtUri<DefaultStr>, 834 853 entity_author: Option<&Did<DefaultStr>>, 835 - repo_owner: &Did<DefaultStr>, 854 + repo_did: Option<&Did<DefaultStr>>, 855 + repo_owner: Option<&Did<DefaultStr>>, 836 856 ) -> bool { 837 857 let Some(src) = source_authority_did(source) else { 838 858 return false; 839 859 }; 840 - Some(&src) == entity_author || &src == repo_owner 860 + Some(&src) == entity_author 861 + || Some(&src) == repo_owner 862 + || repo_did.is_some_and(|repo_did| is_repo_collaborator(edges, repo_did, repo_owner, &src)) 863 + } 864 + 865 + fn is_repo_collaborator( 866 + edges: &EdgeStore, 867 + repo_did: &Did<DefaultStr>, 868 + repo_owner: Option<&Did<DefaultStr>>, 869 + actor: &Did<DefaultStr>, 870 + ) -> bool { 871 + let primary = EdgeKey::new( 872 + nsid_static("sh.tangled.repo.collaborator"), 873 + SubjectRef::Did(repo_did.clone()), 874 + ); 875 + let primary_sources = edges.sources_for(&primary); 876 + 877 + if bobbin_types::knot_acl::collaborator_source(repo_did, actor) 878 + .is_some_and(|source| primary_sources.contains(&source)) 879 + { 880 + return true; 881 + } 882 + 883 + let Some(repo_owner) = repo_owner else { 884 + return false; 885 + }; 886 + let by_actor = EdgeKey::new( 887 + nsid_static("sh.tangled.repo.collaborator.by"), 888 + SubjectRef::Did(actor.clone()), 889 + ); 890 + let actor_sources = edges.sources_for(&by_actor); 891 + primary_sources.into_iter().any(|source| { 892 + actor_sources.contains(&source) 893 + && source_authority_did(&source).as_ref() == Some(repo_owner) 894 + }) 841 895 } 842 896 843 897 fn enrich_view<V, K, F>(

+171 -4

bobbin/crates/xrpc/tests/aggregation.rs

··· 1806 1806 async fn list_issues_state_filter_ignores_third_party_state_source() { 1807 1807 let h = Harness::new().await; 1808 1808 let repo = did("did:plc:limpet"); 1809 + h.state 1810 + .resolver 1811 + .observe( 1812 + did("did:plc:limpet-owner"), 1813 + rkey("repo1"), 1814 + Some(repo.clone()), 1815 + ) 1816 + .await; 1809 1817 let subject = at(&format!("at://{}", repo.as_ref())); 1810 1818 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); 1811 1819 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); ··· 1816 1824 issue_body(&repo, "open issue"), 1817 1825 ) 1818 1826 .await; 1827 + let spoofed_collaborator = at("at://did:plc:nautilus/sh.tangled.repo.collaborator/spoof"); 1828 + h.add_edge( 1829 + &nsid("sh.tangled.repo.collaborator"), 1830 + &subject, 1831 + &spoofed_collaborator, 1832 + ); 1833 + h.add_edge( 1834 + &nsid("sh.tangled.repo.collaborator.by"), 1835 + &at("at://did:plc:nautilus"), 1836 + &spoofed_collaborator, 1837 + ); 1819 1838 h.state.issue_states.upsert( 1820 1839 at("at://did:plc:nautilus/sh.tangled.repo.issue.state/spoof"), 1821 1840 issue_uri.clone(), ··· 1892 1911 #[tokio::test] 1893 1912 async fn list_issues_state_filter_accepts_repo_owner_state_source() { 1894 1913 let h = Harness::new().await; 1895 - let repo_owner = did("did:plc:limpet"); 1896 - let subject = at(&format!("at://{}", repo_owner.as_ref())); 1914 + let repo_owner = did("did:plc:limpet-owner"); 1915 + let repo_did = did("did:plc:limpet-repo"); 1916 + h.state 1917 + .resolver 1918 + .observe(repo_owner.clone(), rkey("repo1"), Some(repo_did.clone())) 1919 + .await; 1920 + let subject = at(&format!("at://{}", repo_did.as_ref())); 1897 1921 let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); 1898 1922 h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); 1899 1923 h.mount( 1900 1924 &did("did:plc:nel"), 1901 1925 &nsid("sh.tangled.repo.issue"), 1902 1926 &rkey("i1"), 1903 - issue_body(&repo_owner, "owner closed"), 1927 + issue_body(&repo_did, "owner closed"), 1904 1928 ) 1905 1929 .await; 1906 1930 h.state.issue_states.upsert( 1907 - at("at://did:plc:limpet/sh.tangled.repo.issue.state/legit"), 1931 + at("at://did:plc:limpet-owner/sh.tangled.repo.issue.state/legit"), 1908 1932 issue_uri.clone(), 1909 1933 1_777_593_800_000_000, 1910 1934 IssueStateKind::Closed, ··· 1929 1953 "repo-owner state record must satisfy state=closed" 1930 1954 ); 1931 1955 assert_eq!(items[0]["state"], json!("closed")); 1956 + } 1957 + 1958 + #[tokio::test] 1959 + async fn list_pulls_status_filter_accepts_repo_owner_status_source() { 1960 + let h = Harness::new().await; 1961 + let repo_owner = did("did:plc:limpet-owner"); 1962 + let repo_did = did("did:plc:limpet-repo"); 1963 + h.state 1964 + .resolver 1965 + .observe(repo_owner.clone(), rkey("repo1"), Some(repo_did.clone())) 1966 + .await; 1967 + let subject = at(&format!("at://{}", repo_did.as_ref())); 1968 + let pull_uri = at("at://did:plc:nel/sh.tangled.repo.pull/p1"); 1969 + h.add_edge(&nsid("sh.tangled.repo.pull"), &subject, &pull_uri); 1970 + h.mount( 1971 + &did("did:plc:nel"), 1972 + &nsid("sh.tangled.repo.pull"), 1973 + &rkey("p1"), 1974 + pull_body(&repo_did, "owner merged"), 1975 + ) 1976 + .await; 1977 + h.state.pull_statuses.upsert( 1978 + at("at://did:plc:limpet-owner/sh.tangled.repo.pull.status/legit"), 1979 + pull_uri.clone(), 1980 + 1_777_593_800_000_000, 1981 + PullStatusKind::Merged, 1982 + ); 1983 + 1984 + let app = router(h.state.clone()); 1985 + let (status, body) = json_response( 1986 + app.oneshot(list_request( 1987 + "sh.tangled.repo.listPulls", 1988 + subject.as_ref(), 1989 + &[("status", "merged")], 1990 + )) 1991 + .await 1992 + .unwrap(), 1993 + ) 1994 + .await; 1995 + assert_eq!(status, StatusCode::OK); 1996 + let items = body["items"].as_array().expect("items array"); 1997 + assert_eq!(items.len(), 1, "repo-owner status must satisfy merged"); 1998 + assert_eq!(items[0]["state"], json!("merged")); 1999 + } 2000 + 2001 + #[tokio::test] 2002 + async fn list_issues_state_filter_accepts_knot_collaborator_state_source() { 2003 + let h = Harness::new().await; 2004 + let repo_owner = did("did:plc:dfl62fgb7wtjj3fcbb72naae"); 2005 + let repo_did = did("did:plc:hnv6iwmznhtn3onxkjyytvir"); 2006 + let collaborator = did("did:plc:3rwz3xfw2crswgifqgc3g7zh"); 2007 + h.state 2008 + .resolver 2009 + .observe(repo_owner, rkey("ark"), Some(repo_did.clone())) 2010 + .await; 2011 + let (collaborator_source, collaborator_edges) = 2012 + bobbin_types::knot_acl::collaborator_upsert(&repo_did, &collaborator, 1) 2013 + .expect("valid collaborator acl"); 2014 + h.edges 2015 + .upsert_source(&collaborator_source, collaborator_edges); 2016 + 2017 + let subject = at(&format!("at://{}", repo_did.as_ref())); 2018 + let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); 2019 + h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); 2020 + h.mount( 2021 + &did("did:plc:nel"), 2022 + &nsid("sh.tangled.repo.issue"), 2023 + &rkey("i1"), 2024 + issue_body(&repo_did, "collaborator closed"), 2025 + ) 2026 + .await; 2027 + h.state.issue_states.upsert( 2028 + at("at://did:plc:3rwz3xfw2crswgifqgc3g7zh/sh.tangled.repo.issue.state/s1"), 2029 + issue_uri, 2030 + 1_777_593_800_000_000, 2031 + IssueStateKind::Closed, 2032 + ); 2033 + 2034 + let (status, body) = json_response( 2035 + router(h.state.clone()) 2036 + .oneshot(list_request( 2037 + "sh.tangled.repo.listIssues", 2038 + subject.as_ref(), 2039 + &[("state", "closed")], 2040 + )) 2041 + .await 2042 + .unwrap(), 2043 + ) 2044 + .await; 2045 + assert_eq!(status, StatusCode::OK); 2046 + let items = body["items"].as_array().expect("items array"); 2047 + assert_eq!(items.len(), 1, "collaborator state must satisfy closed"); 2048 + assert_eq!(items[0]["state"], json!("closed")); 2049 + } 2050 + 2051 + #[tokio::test] 2052 + async fn list_pulls_status_filter_accepts_knot_collaborator_status_source() { 2053 + let h = Harness::new().await; 2054 + let repo_owner = did("did:plc:dfl62fgb7wtjj3fcbb72naae"); 2055 + let repo_did = did("did:plc:hnv6iwmznhtn3onxkjyytvir"); 2056 + let collaborator = did("did:plc:3rwz3xfw2crswgifqgc3g7zh"); 2057 + h.state 2058 + .resolver 2059 + .observe(repo_owner, rkey("ark"), Some(repo_did.clone())) 2060 + .await; 2061 + let (collaborator_source, collaborator_edges) = 2062 + bobbin_types::knot_acl::collaborator_upsert(&repo_did, &collaborator, 1) 2063 + .expect("valid collaborator acl"); 2064 + h.edges 2065 + .upsert_source(&collaborator_source, collaborator_edges); 2066 + 2067 + let subject = at(&format!("at://{}", repo_did.as_ref())); 2068 + let pull_uri = at("at://did:plc:nel/sh.tangled.repo.pull/p1"); 2069 + h.add_edge(&nsid("sh.tangled.repo.pull"), &subject, &pull_uri); 2070 + h.mount( 2071 + &did("did:plc:nel"), 2072 + &nsid("sh.tangled.repo.pull"), 2073 + &rkey("p1"), 2074 + pull_body(&repo_did, "collaborator merged"), 2075 + ) 2076 + .await; 2077 + h.state.pull_statuses.upsert( 2078 + at("at://did:plc:3rwz3xfw2crswgifqgc3g7zh/sh.tangled.repo.pull.status/s1"), 2079 + pull_uri, 2080 + 1_777_593_800_000_000, 2081 + PullStatusKind::Merged, 2082 + ); 2083 + 2084 + let (status, body) = json_response( 2085 + router(h.state.clone()) 2086 + .oneshot(list_request( 2087 + "sh.tangled.repo.listPulls", 2088 + subject.as_ref(), 2089 + &[("status", "merged")], 2090 + )) 2091 + .await 2092 + .unwrap(), 2093 + ) 2094 + .await; 2095 + assert_eq!(status, StatusCode::OK); 2096 + let items = body["items"].as_array().expect("items array"); 2097 + assert_eq!(items.len(), 1, "collaborator status must satisfy merged"); 2098 + assert_eq!(items[0]["state"], json!("merged")); 1932 2099 } 1933 2100 1934 2101 #[tokio::test]

History

2 rounds 11 comments

okami.mom submitted #1 7w

Interdiff

1 commit

Expand

028c9440

bobbin/xrpc: accept owner- and collaborator-authored issue and pull state

Expand 9 comments

okami.mom Edited 7w ago

so i was able to actually test with a hydrant instance (dawn pointed me to (hyd.bob.oyster.cafe)), the PR now considers collaborators and ensures once were collabarators but no longer arent counted either, however when pointed at tangled.org/core, bobbin is still slightly off the count of open issues/pulls that appear on https://tangled.org/tangled.org/core/

production bobbin: issues 687 open, 17 closed
patched bobbin: issues 314 open, 390 closed
as of aug 8 3am est tangled.org 305 open, 399 closed

i think either the hydrant is missing repos, or the appview has zombie records or something that dont exist anymore

cc: @oyster.cafe

oyster.cafe 7w ago

there is indeed a real possibility of appview having zombie records

thank you! i will look through this

okami.mom 6w ago

@oyster.cafe boop

oyster.cafe 6w ago

oh yeah sorry mama mia completely forgor

oyster.cafe 5w ago

ok so

bobbin/crates/resolver/src/lib.rs:229 ok hang on repoDid on sh.tangled.repo is an unrestricted format: did and claim_repo goes straight to observe with no check that the repoDID's doc points back at the publisher, and the by_repo_did update is LWW? I think that's a squatting/overwriting opportunity right. could you please put in a fix in observe to like refuse rebinding an existing by_repo_did entry to a different owner? Actually or maybe port something like verifyOwnership from the golang appview/ingester_repo ?

oyster.cafe 5w ago

bobbin/crates/xrpc/src/lib.rs:875 could you pls add smth like EdgeStore::has_source(&EdgeKey, &AtUri) -> bool which would intern the source / test membership without having to materialize out a Vec? that way we could avoid deadlock from list_filtered

oyster.cafe 5w ago

bobbin/crates/xrpc/src/lib.rs:891-894 i think u should swap these checks in the any iterator for efficiency heh

oyster.cafe 5w ago

bobbin/crates/xrpc/src/filter.rs:57 pls stop inferring the repo from the subject - could you give ListFilter::predicate a distinct rusty newtype for a repo-subject that only the repo-keyed routes can create (and therefore make the *By routes not pass anything)?

what I mean is that it shouldn't be the case that SubjectRef::Did that is sometimes a repo and sometimes an author :p

oyster.cafe 5w ago

bobbin/crates/xrpc/src/lib.rs:808 this seems to get repo from view.value.repo but the filter gets it from the request subject - could you please derive the repo ident only once per request and use the same value on both sides?

and/or hey could put the parsed subject into enrich_issue_view / enrich_pull_view which would make this contradiction un-representable