+218 -5

Cargo.lock

··· 39 39 ] 40 40 41 41 [[package]] 42 42 + name = "android-tzdata" 43 43 + version = "0.1.1" 44 44 + source = "registry+https://github.com/rust-lang/crates.io-index" 45 45 + checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0" 46 46 + 47 47 + [[package]] 48 48 + name = "android_system_properties" 49 49 + version = "0.1.5" 50 50 + source = "registry+https://github.com/rust-lang/crates.io-index" 51 51 + checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" 52 52 + dependencies = [ 53 53 + "libc", 54 54 + ] 55 55 + 56 56 + [[package]] 42 57 name = "anstream" 43 58 version = "0.6.18" 44 59 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 326 341 checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" 327 342 328 343 [[package]] 344 344 + name = "chrono" 345 345 + version = "0.4.39" 346 346 + source = "registry+https://github.com/rust-lang/crates.io-index" 347 347 + checksum = "7e36cc9d416881d2e24f9a963be5fb1cd90966419ac844274161d10488b3e825" 348 348 + dependencies = [ 349 349 + "android-tzdata", 350 350 + "iana-time-zone", 351 351 + "num-traits", 352 352 + "serde", 353 353 + "windows-targets", 354 354 + ] 355 355 + 356 356 + [[package]] 329 357 name = "clang-sys" 330 358 version = "1.8.1" 331 359 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 443 471 ] 444 472 445 473 [[package]] 474 474 + name = "darling" 475 475 + version = "0.20.10" 476 476 + source = "registry+https://github.com/rust-lang/crates.io-index" 477 477 + checksum = "6f63b86c8a8826a49b8c21f08a2d07338eec8d900540f8630dc76284be802989" 478 478 + dependencies = [ 479 479 + "darling_core", 480 480 + "darling_macro", 481 481 + ] 482 482 + 483 483 + [[package]] 484 484 + name = "darling_core" 485 485 + version = "0.20.10" 486 486 + source = "registry+https://github.com/rust-lang/crates.io-index" 487 487 + checksum = "95133861a8032aaea082871032f5815eb9e98cef03fa916ab4500513994df9e5" 488 488 + dependencies = [ 489 489 + "fnv", 490 490 + "ident_case", 491 491 + "proc-macro2", 492 492 + "quote", 493 493 + "strsim", 494 494 + "syn", 495 495 + ] 496 496 + 497 497 + [[package]] 498 498 + name = "darling_macro" 499 499 + version = "0.20.10" 500 500 + source = "registry+https://github.com/rust-lang/crates.io-index" 501 501 + checksum = "d336a2a514f6ccccaa3e09b02d41d35330c07ddf03a62165fcec10bb561c7806" 502 502 + dependencies = [ 503 503 + "darling_core", 504 504 + "quote", 505 505 + "syn", 506 506 + ] 507 507 + 508 508 + [[package]] 446 509 name = "data-encoding" 447 510 version = "2.7.0" 448 511 source = "registry+https://github.com/rust-lang/crates.io-index" 449 512 checksum = "0e60eed09d8c01d3cee5b7d30acb059b76614c918fa0f992e0dd6eeb10daad6f" 513 513 + 514 514 + [[package]] 515 515 + name = "deranged" 516 516 + version = "0.3.11" 517 517 + source = "registry+https://github.com/rust-lang/crates.io-index" 518 518 + checksum = "b42b6fa04a440b495c8b04d0e71b707c585f83cb9cb28cf8cd0d976c315e31b4" 519 519 + dependencies = [ 520 520 + "powerfmt", 521 521 + "serde", 522 522 + ] 450 523 451 524 [[package]] 452 525 name = "digest" ··· 635 708 "futures-core", 636 709 "futures-sink", 637 710 "http", 638 638 - "indexmap", 711 711 + "indexmap 2.7.1", 639 712 "slab", 640 713 "tokio", 641 714 "tokio-util", 642 715 "tracing", 643 716 ] 717 717 + 718 718 + [[package]] 719 719 + name = "hashbrown" 720 720 + version = "0.12.3" 721 721 + source = "registry+https://github.com/rust-lang/crates.io-index" 722 722 + checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" 644 723 645 724 [[package]] 646 725 name = "hashbrown" ··· 750 829 ] 751 830 752 831 [[package]] 832 832 + name = "iana-time-zone" 833 833 + version = "0.1.61" 834 834 + source = "registry+https://github.com/rust-lang/crates.io-index" 835 835 + checksum = "235e081f3925a06703c2d0117ea8b91f042756fd6e7a6e5d901e8ca1a996b220" 836 836 + dependencies = [ 837 837 + "android_system_properties", 838 838 + "core-foundation-sys", 839 839 + "iana-time-zone-haiku", 840 840 + "js-sys", 841 841 + "wasm-bindgen", 842 842 + "windows-core 0.52.0", 843 843 + ] 844 844 + 845 845 + [[package]] 846 846 + name = "iana-time-zone-haiku" 847 847 + version = "0.1.2" 848 848 + source = "registry+https://github.com/rust-lang/crates.io-index" 849 849 + checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" 850 850 + dependencies = [ 851 851 + "cc", 852 852 + ] 853 853 + 854 854 + [[package]] 855 855 + name = "ident_case" 856 856 + version = "1.0.1" 857 857 + source = "registry+https://github.com/rust-lang/crates.io-index" 858 858 + checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" 859 859 + 860 860 + [[package]] 861 861 + name = "indexmap" 862 862 + version = "1.9.3" 863 863 + source = "registry+https://github.com/rust-lang/crates.io-index" 864 864 + checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99" 865 865 + dependencies = [ 866 866 + "autocfg", 867 867 + "hashbrown 0.12.3", 868 868 + "serde", 869 869 + ] 870 870 + 871 871 + [[package]] 753 872 name = "indexmap" 754 873 version = "2.7.1" 755 874 source = "registry+https://github.com/rust-lang/crates.io-index" 756 875 checksum = "8c9c992b02b5b4c94ea26e32fe5bccb7aa7d9f390ab5c1221ff895bc7ea8b652" 757 876 dependencies = [ 758 877 "equivalent", 759 759 - "hashbrown", 878 878 + "hashbrown 0.15.2", 879 879 + "serde", 760 880 ] 761 881 762 882 [[package]] ··· 897 1017 "metrics-process", 898 1018 "rocksdb", 899 1019 "serde", 1020 1020 + "serde_with", 900 1021 "tempfile", 901 1022 "tinyjson", 902 1023 "tokio", ··· 989 1110 "http-body-util", 990 1111 "hyper", 991 1112 "hyper-util", 992 992 - "indexmap", 1113 1113 + "indexmap 2.7.1", 993 1114 "ipnet", 994 1115 "metrics", 995 1116 "metrics-util", ··· 1023 1144 dependencies = [ 1024 1145 "crossbeam-epoch", 1025 1146 "crossbeam-utils", 1026 1026 - "hashbrown", 1147 1147 + "hashbrown 0.15.2", 1027 1148 "metrics", 1028 1149 "quanta", 1029 1150 "rand", ··· 1103 1224 ] 1104 1225 1105 1226 [[package]] 1227 1227 + name = "num-conv" 1228 1228 + version = "0.1.0" 1229 1229 + source = "registry+https://github.com/rust-lang/crates.io-index" 1230 1230 + checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9" 1231 1231 + 1232 1232 + [[package]] 1233 1233 + name = "num-traits" 1234 1234 + version = "0.2.19" 1235 1235 + source = "registry+https://github.com/rust-lang/crates.io-index" 1236 1236 + checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" 1237 1237 + dependencies = [ 1238 1238 + "autocfg", 1239 1239 + ] 1240 1240 + 1241 1241 + [[package]] 1106 1242 name = "object" 1107 1243 version = "0.36.7" 1108 1244 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1213 1349 version = "1.10.0" 1214 1350 source = "registry+https://github.com/rust-lang/crates.io-index" 1215 1351 checksum = "280dc24453071f1b63954171985a0b0d30058d287960968b9b2aca264c8d4ee6" 1352 1352 + 1353 1353 + [[package]] 1354 1354 + name = "powerfmt" 1355 1355 + version = "0.2.0" 1356 1356 + source = "registry+https://github.com/rust-lang/crates.io-index" 1357 1357 + checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" 1216 1358 1217 1359 [[package]] 1218 1360 name = "ppv-lite86" ··· 1533 1675 ] 1534 1676 1535 1677 [[package]] 1678 1678 + name = "serde_with" 1679 1679 + version = "3.12.0" 1680 1680 + source = "registry+https://github.com/rust-lang/crates.io-index" 1681 1681 + checksum = "d6b6f7f2fcb69f747921f79f3926bd1e203fce4fef62c268dd3abfb6d86029aa" 1682 1682 + dependencies = [ 1683 1683 + "base64", 1684 1684 + "chrono", 1685 1685 + "hex", 1686 1686 + "indexmap 1.9.3", 1687 1687 + "indexmap 2.7.1", 1688 1688 + "serde", 1689 1689 + "serde_derive", 1690 1690 + "serde_json", 1691 1691 + "serde_with_macros", 1692 1692 + "time", 1693 1693 + ] 1694 1694 + 1695 1695 + [[package]] 1696 1696 + name = "serde_with_macros" 1697 1697 + version = "3.12.0" 1698 1698 + source = "registry+https://github.com/rust-lang/crates.io-index" 1699 1699 + checksum = "8d00caa5193a3c8362ac2b73be6b9e768aa5a4b2f721d8f4b339600c3cb51f8e" 1700 1700 + dependencies = [ 1701 1701 + "darling", 1702 1702 + "proc-macro2", 1703 1703 + "quote", 1704 1704 + "syn", 1705 1705 + ] 1706 1706 + 1707 1707 + [[package]] 1536 1708 name = "sha1" 1537 1709 version = "0.10.6" 1538 1710 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1676 1848 ] 1677 1849 1678 1850 [[package]] 1851 1851 + name = "time" 1852 1852 + version = "0.3.37" 1853 1853 + source = "registry+https://github.com/rust-lang/crates.io-index" 1854 1854 + checksum = "35e7868883861bd0e56d9ac6efcaaca0d6d5d82a2a7ec8209ff492c07cf37b21" 1855 1855 + dependencies = [ 1856 1856 + "deranged", 1857 1857 + "itoa", 1858 1858 + "num-conv", 1859 1859 + "powerfmt", 1860 1860 + "serde", 1861 1861 + "time-core", 1862 1862 + "time-macros", 1863 1863 + ] 1864 1864 + 1865 1865 + [[package]] 1866 1866 + name = "time-core" 1867 1867 + version = "0.1.2" 1868 1868 + source = "registry+https://github.com/rust-lang/crates.io-index" 1869 1869 + checksum = "ef927ca75afb808a4d64dd374f00a2adf8d0fcff8e7b184af886c3c87ec4a3f3" 1870 1870 + 1871 1871 + [[package]] 1872 1872 + name = "time-macros" 1873 1873 + version = "0.2.19" 1874 1874 + source = "registry+https://github.com/rust-lang/crates.io-index" 1875 1875 + checksum = "2834e6017e3e5e4b9834939793b282bc03b37a3336245fa820e35e233e2a85de" 1876 1876 + dependencies = [ 1877 1877 + "num-conv", 1878 1878 + "time-core", 1879 1879 + ] 1880 1880 + 1881 1881 + [[package]] 1679 1882 name = "tinyjson" 1680 1883 version = "2.5.1" 1681 1884 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1855 2058 dependencies = [ 1856 2059 "cfg-if", 1857 2060 "once_cell", 2061 2061 + "rustversion", 1858 2062 "wasm-bindgen-macro", 1859 2063 ] 1860 2064 ··· 1942 2146 source = "registry+https://github.com/rust-lang/crates.io-index" 1943 2147 checksum = "dd04d41d93c4992d421894c18c8b43496aa748dd4c081bac0dc93eb0489272b6" 1944 2148 dependencies = [ 1945 1945 - "windows-core", 2149 2149 + "windows-core 0.58.0", 2150 2150 + "windows-targets", 2151 2151 + ] 2152 2152 + 2153 2153 + [[package]] 2154 2154 + name = "windows-core" 2155 2155 + version = "0.52.0" 2156 2156 + source = "registry+https://github.com/rust-lang/crates.io-index" 2157 2157 + checksum = "33ab640c8d7e35bf8ba19b884ba838ceb4fba93a4e8c65a9059d08afcfc683d9" 2158 2158 + dependencies = [ 1946 2159 "windows-targets", 1947 2160 ] 1948 2161

+19 -2

cozy-setup (move to another repo).md

··· 53 53 match { 54 54 expression {header.User-Agent}.startsWith("Mozilla/5.0") 55 55 } 56 56 - events 2 57 57 - window 15s 56 56 + events 1000 57 57 + window 30s 58 58 log_key true 59 59 } 60 60 } ··· 74 74 cache 75 75 } 76 76 ``` 77 77 + well... the gateway fell over IMMEDIATELY with like 2 req/sec from deletions, with that ^^ config. for now i removed everything except the reverse proxy config + normal caddy metrics and it's running fine on vanilla caddy. i did try reducing the rate-limiting configs to a single, fixed-key global limit but it still ate all the ram and died. maybe badger w/ the cache config was still a problem. maybe it would have been ok on a machine with more than 1GB mem. 78 78 + 79 79 + 80 80 + alternative proxies: 81 81 + 82 82 + - nginx. i should probably just use this. acme-client is a piece of cake to set up, and i know how to configure it. 83 83 + - haproxy. also kind of familiar, it's old and stable. no idea how it handle low-mem (our 1gb) vs nginx. 84 84 + - sozu. popular rust thing, fast. doesn't have rate-limiting or cache feature? 85 85 + - rpxy. like caddy (auto-tls) but in rust and actually fast? has an "experimental" cache feature. but the cache feature looks good. 86 86 + - rama. build-your-own proxy. not sure that it has both cache and limiter in their standard features? 87 87 + - pingora. build-your-own cloudflare, so like, probably stable. has tools for cache and limiting. low-mem...? 88 88 + - cache stuff in pingora seems a little... hit and miss (byeeeee). only a test impl for Storage for the main cache feature? 89 89 + - but the rate-limiter has a guide: https://github.com/cloudflare/pingora/blob/main/docs/user_guide/rate_limiter.md 90 90 + 91 91 + what i want is low-resource reverse proxy with built-in rate-limiting and caching. but maybe cache (and/or ratelimiting) could be external to the reverse proxy 92 92 + - varnish is a dedicated cache. has https://github.com/varnish/varnish-modules/blob/master/src/vmod_vsthrottle.vcc 93 93 + - apache traffic control has experimental rate-limiting plugins 77 94 78 95 79 96 - victoriametrics

+1

link_aggregator/Cargo.toml

··· 18 18 metrics-process = "2.4.0" 19 19 rocksdb = { version = "0.23.0", optional = true } 20 20 serde = { version = "1.0.215", features = ["derive"] } 21 21 + serde_with = { version = "3.12.0", features = ["hex"] } 21 22 tinyjson = "2.5.1" 22 23 tokio-util = "0.7.13" 23 24 zstd = "0.13.2"

+9 -6

link_aggregator/readme.md

··· 74 74 - [x] move ownership of canonical seq to an owned non-atomic 75 75 - [x] custom path for db storage 76 76 - [x] api server to look up backlink count 77 77 - - [ ] other useful endpoints for the api server 77 77 + - [~] other useful endpoints for the api server 78 78 - [x] show all nisd/path links to target 79 79 - - [ ] get backlinking dids 80 80 - - [ ] paging for all backlinking dids 81 81 - - [ ] get count + most recent dids 79 79 + - [x] get backlinking dids 80 80 + - [x] paging for all backlinking dids 81 81 + - [x] get count + most recent dids 82 82 - [ ] get count with any dids from provided set 83 83 - [~] write this readme 84 84 - [?] fix it sometimes getting stuck ··· 100 100 - [ ] deadletter queue of some kind for failed db writes 101 101 - [ ] also for valid json that was rejected? 102 102 - [x] get it running on raspi 103 103 - - [~] get an estimate of disk usage per day after a few days of running 103 103 + - [x] get an estimate of disk usage per day after a few days of running 104 104 + - very close to 1GB with data model before adding rkeys to linkers + fixing paths 104 105 - [ ] make the did_init check only happen on test config (or remove it) 105 106 - [ ] actual error types (thiserror?) for lib-ish code 106 107 - [ ] clean up the main readme ··· 116 117 - [ ] put delete-account tasks into a separate (persisted?) task queue for the writer so it can work on them incrementally. 117 118 118 119 data fixes 119 119 - - [ ] add rkey to linkers 🤦‍♀️ 120 120 + - [x] add rkey to linkers 🤦‍♀️ 121 121 + - [x] don't remove deleted links from the reverse records -- null them out. this will keep things stable for paging. 122 122 + - [x] don't show deactivated accounts in link responses 120 123 - [ ] links: 121 124 - [ ] pull `$type`/`type` from object children of arrays (distinguish replies, quotes, etc) 122 125 - [ ] actually define the format (deal with in-band dots etc)

+1 -1

link_aggregator/src/lib.rs

··· 27 27 } 28 28 } 29 29 30 30 - #[derive(Debug, PartialEq)] 30 30 + #[derive(Debug, PartialEq, Serialize, Deserialize)] 31 31 pub struct RecordId { 32 32 pub did: Did, 33 33 pub collection: String,

+93 -4

link_aggregator/src/server.rs

··· 1 1 use axum::{extract::Query, http, routing::get, Json, Router}; 2 2 - use serde::Deserialize; 2 2 + use bincode::Options; 3 3 + use serde::{Deserialize, Serialize}; 4 4 + use serde_with::serde_as; 3 5 use std::collections::HashMap; 4 6 use tokio::net::{TcpListener, ToSocketAddrs}; 5 7 use tokio::task::block_in_place; 6 8 use tokio_util::sync::CancellationToken; 7 9 8 10 use crate::storage::LinkReader; 11 11 + use link_aggregator::RecordId; 12 12 + 13 13 + const DEFAULT_CURSOR_LIMIT: u64 = 16; 14 14 + const DEFAULT_CURSOR_LIMIT_MAX: u64 = 100; 9 15 10 16 pub async fn serve<S, A>(store: S, addr: A, stay_alive: CancellationToken) -> anyhow::Result<()> 11 17 where ··· 22 28 }), 23 29 ) 24 30 .route( 31 31 + "/links", 32 32 + get({ 33 33 + let store = store.clone(); 34 34 + move |query| async { block_in_place(|| get_links(query, store)) } 35 35 + }), 36 36 + ) 37 37 + .route( 25 38 "/links/all/count", 26 39 get({ 27 40 let store = store.clone(); ··· 44 57 } 45 58 46 59 #[derive(Deserialize)] 47 47 - struct GetLinksQuery { 60 60 + struct GetLinksCountQuery { 48 61 target: String, 49 62 collection: String, 50 63 path: String, 51 64 } 52 52 - 53 65 fn count_links( 54 54 - query: Query<GetLinksQuery>, 66 66 + query: Query<GetLinksCountQuery>, 55 67 store: impl LinkReader, 56 68 ) -> Result<String, http::StatusCode> { 57 69 store ··· 61 73 } 62 74 63 75 #[derive(Deserialize)] 76 76 + struct GetLinkItemsQuery { 77 77 + target: String, 78 78 + collection: String, 79 79 + path: String, 80 80 + cursor: Option<OpaqueApiCursor>, 81 81 + limit: Option<u64>, 82 82 + // TODO: allow reverse (er, forward) order as well 83 83 + } 84 84 + #[derive(Serialize)] 85 85 + struct GetLinkItemsResponse { 86 86 + // what does staleness mean? 87 87 + // - new links have appeared. would be nice to offer a `since` cursor to fetch these. and/or, 88 88 + // - links have been deleted. hmm. 89 89 + total: u64, 90 90 + linking_records: Vec<RecordId>, 91 91 + cursor: Option<OpaqueApiCursor>, 92 92 + } 93 93 + fn get_links( 94 94 + query: Query<GetLinkItemsQuery>, 95 95 + store: impl LinkReader, 96 96 + ) -> Result<Json<GetLinkItemsResponse>, http::StatusCode> { 97 97 + let until = query 98 98 + .cursor 99 99 + .clone() 100 100 + .map(|oc| ApiCursor::try_from(oc).map_err(|_| http::StatusCode::BAD_REQUEST)) 101 101 + .transpose()? 102 102 + .map(|c| c.next); 103 103 + 104 104 + let limit = query.limit.unwrap_or(DEFAULT_CURSOR_LIMIT); 105 105 + if limit > DEFAULT_CURSOR_LIMIT_MAX { 106 106 + return Err(http::StatusCode::BAD_REQUEST); 107 107 + } 108 108 + 109 109 + let paged = store 110 110 + .get_links(&query.target, &query.collection, &query.path, limit, until) 111 111 + .map_err(|_| http::StatusCode::INTERNAL_SERVER_ERROR)?; 112 112 + 113 113 + let cursor = paged.next.map(|next| { 114 114 + ApiCursor { 115 115 + version: paged.version, 116 116 + next, 117 117 + } 118 118 + .into() 119 119 + }); 120 120 + 121 121 + Ok(Json(GetLinkItemsResponse { 122 122 + total: paged.version.0, 123 123 + linking_records: paged.items, 124 124 + cursor, 125 125 + })) 126 126 + } 127 127 + 128 128 + #[derive(Deserialize)] 64 129 struct GetAllLinksQuery { 65 130 target: String, 66 131 } ··· 74 139 .map(Json) 75 140 .map_err(|_| http::StatusCode::INTERNAL_SERVER_ERROR) 76 141 } 142 142 + 143 143 + #[serde_as] 144 144 + #[derive(Clone, Serialize, Deserialize)] // for json 145 145 + struct OpaqueApiCursor(#[serde_as(as = "serde_with::hex::Hex")] Vec<u8>); 146 146 + 147 147 + #[derive(Serialize, Deserialize)] // for bincode 148 148 + struct ApiCursor { 149 149 + version: (u64, u64), // (collection length, deleted item count) 150 150 + next: u64, 151 151 + } 152 152 + 153 153 + impl TryFrom<OpaqueApiCursor> for ApiCursor { 154 154 + type Error = bincode::Error; 155 155 + 156 156 + fn try_from(item: OpaqueApiCursor) -> Result<Self, Self::Error> { 157 157 + bincode::DefaultOptions::new().deserialize(&item.0) 158 158 + } 159 159 + } 160 160 + 161 161 + impl From<ApiCursor> for OpaqueApiCursor { 162 162 + fn from(item: ApiCursor) -> Self { 163 163 + OpaqueApiCursor(bincode::DefaultOptions::new().serialize(&item).unwrap()) 164 164 + } 165 165 + }

+112 -23

link_aggregator/src/storage/mem_store.rs

··· 1 1 - use super::{LinkReader, LinkStorage}; 1 1 + use super::{LinkReader, LinkStorage, PagedAppendingCollection}; 2 2 use anyhow::Result; 3 3 - use link_aggregator::{Did, RecordId}; 3 3 + use link_aggregator::{ActionableEvent, Did, RecordId}; 4 4 use links::CollectedLink; 5 5 use std::collections::HashMap; 6 6 use std::sync::{Arc, Mutex}; ··· 9 9 #[derive(Debug, Clone)] 10 10 pub struct MemStorage(Arc<Mutex<MemStorageData>>); 11 11 12 12 + type Linkers = Vec<Option<(Did, RKey)>>; // optional because we replace with None for deleted links to keep cursors stable 13 13 + 12 14 #[derive(Debug, Default)] 13 15 struct MemStorageData { 14 14 - dids: HashMap<Did, bool>, // bool: active or nah 15 15 - targets: HashMap<Target, HashMap<Source, Vec<Did>>>, // target -> (collection, path) -> did[] 16 16 + dids: HashMap<Did, bool>, // bool: active or nah 17 17 + targets: HashMap<Target, HashMap<Source, Linkers>>, // target -> (collection, path) -> (did, rkey)?[] 16 18 links: HashMap<Did, HashMap<RepoId, Vec<(RecordPath, Target)>>>, // did -> collection:rkey -> (path, target)[] 17 19 } 18 20 ··· 20 22 pub fn new() -> Self { 21 23 Self(Arc::new(Mutex::new(MemStorageData::default()))) 22 24 } 23 23 - } 24 25 25 25 - impl LinkStorage for MemStorage { 26 26 fn add_links(&mut self, record_id: &RecordId, links: &[CollectedLink]) { 27 27 let mut data = self.0.lock().unwrap(); 28 28 for link in links { ··· 32 32 .or_default() 33 33 .entry(Source::new(&record_id.collection, &link.path)) 34 34 .or_default() 35 35 - .push(record_id.did()); 35 35 + .push(Some((record_id.did(), RKey(record_id.rkey())))); 36 36 data.links 37 37 .entry(record_id.did()) 38 38 .or_default() ··· 52 52 { 53 53 let link_targets = link_targets.clone(); // satisfy borrowck 54 54 for (record_path, target) in link_targets { 55 55 - let dids = data 56 56 - .targets 55 55 + data.targets 57 56 .get_mut(&target) 58 57 .expect("must have the target if we have a link saved") 59 58 .get_mut(&Source::new(&record_id.collection, &record_path.0)) 60 60 - .expect("must have the target at this path if we have a link to it saved"); 61 61 - // search from the end: more likely to be visible and deletes are usually soon after creates 62 62 - // only delete one instance: a user can create multiple links to something, we're only deleting one 63 63 - // (we don't know which one in the list we should be deleting, and it hopefully mostly doesn't matter) 64 64 - let pos = dids 65 65 - .iter() 66 66 - .rposition(|d| *d == record_id.did) 67 67 - .expect("must be in dids list if we have a link to it"); 68 68 - dids.remove(pos); 59 59 + .expect("must have the target at this path if we have a link to it saved") 60 60 + .iter_mut() 61 61 + .rfind(|d| **d == Some((record_id.did(), RKey(record_id.rkey())))) 62 62 + .expect("must be in dids list if we have a link to it") 63 63 + .take(); 69 64 } 70 65 } 71 66 data.links ··· 73 68 .map(|cr| cr.remove(&repo_id)); 74 69 } 75 70 71 71 + fn update_links(&mut self, record_id: &RecordId, new_links: &[CollectedLink]) { 72 72 + self.remove_links(record_id); 73 73 + self.add_links(record_id, new_links); 74 74 + } 75 75 + 76 76 fn set_account(&mut self, did: &Did, active: bool) { 77 77 let mut data = self.0.lock().unwrap(); 78 78 if let Some(account) = data.dids.get_mut(did) { ··· 92 92 .expect("must have the target if we have a link saved") 93 93 .get_mut(&Source::new(&repo_id.collection, &record_path.0)) 94 94 .expect("must have the target at this path if we have a link to it saved") 95 95 - .retain(|d| d != did); 95 95 + .iter_mut() 96 96 + .find(|d| **d == Some((did.clone(), repo_id.rkey.clone()))) 97 97 + .expect("lkasjdlfkj") 98 98 + .take(); 96 99 } 97 100 } 98 101 } 99 102 data.links.remove(did); // nb: this is removing by a whole prefix in kv context 100 103 data.dids.remove(did); 101 104 } 105 105 + } 106 106 + 107 107 + impl LinkStorage for MemStorage { 108 108 + fn push(&mut self, event: &ActionableEvent, _cursor: u64) -> Result<()> { 109 109 + match event { 110 110 + ActionableEvent::CreateLinks { record_id, links } => self.add_links(record_id, links), 111 111 + ActionableEvent::UpdateLinks { 112 112 + record_id, 113 113 + new_links, 114 114 + } => self.update_links(record_id, new_links), 115 115 + ActionableEvent::DeleteRecord(record_id) => self.remove_links(record_id), 116 116 + ActionableEvent::ActivateAccount(did) => self.set_account(did, true), 117 117 + ActionableEvent::DeactivateAccount(did) => self.set_account(did, false), 118 118 + ActionableEvent::DeleteAccount(did) => self.delete_account(did), 119 119 + } 120 120 + Ok(()) 121 121 + } 102 122 103 123 fn to_readable(&mut self) -> impl LinkReader { 104 124 self.clone() ··· 114 134 let Some(dids) = paths.get(&Source::new(collection, path)) else { 115 135 return Ok(0); 116 136 }; 117 117 - let count = dids.len().try_into()?; 118 118 - Ok(count) 137 137 + Ok(dids.iter().flatten().count().try_into()?) 138 138 + } 139 139 + 140 140 + fn get_links( 141 141 + &self, 142 142 + target: &str, 143 143 + collection: &str, 144 144 + path: &str, 145 145 + limit: u64, 146 146 + until: Option<u64>, 147 147 + ) -> Result<PagedAppendingCollection<RecordId>> { 148 148 + let data = self.0.lock().unwrap(); 149 149 + let Some(paths) = data.targets.get(&Target::new(target)) else { 150 150 + return Ok(PagedAppendingCollection { 151 151 + version: (0, 0), 152 152 + items: Vec::new(), 153 153 + next: None, 154 154 + }); 155 155 + }; 156 156 + let Some(did_rkeys) = paths.get(&Source::new(collection, path)) else { 157 157 + return Ok(PagedAppendingCollection { 158 158 + version: (0, 0), 159 159 + items: Vec::new(), 160 160 + next: None, 161 161 + }); 162 162 + }; 163 163 + 164 164 + let total = did_rkeys.len(); 165 165 + let end = until 166 166 + .map(|u| std::cmp::min(u as usize, total)) 167 167 + .unwrap_or(total); 168 168 + let begin = end.saturating_sub(limit as usize); 169 169 + let next = if begin == 0 { None } else { Some(begin as u64) }; 170 170 + 171 171 + let alive = did_rkeys.iter().flatten().count(); 172 172 + let gone = total - alive; 173 173 + 174 174 + let items: Vec<_> = did_rkeys[begin..end] 175 175 + .iter() 176 176 + .rev() 177 177 + .flatten() 178 178 + .filter(|(did, _)| *data.dids.get(did).expect("did must be in dids")) 179 179 + .map(|(did, rkey)| RecordId { 180 180 + did: did.clone(), 181 181 + rkey: rkey.0.clone(), 182 182 + collection: collection.to_string(), 183 183 + }) 184 184 + .collect(); 185 185 + 186 186 + Ok(PagedAppendingCollection { 187 187 + version: (total as u64, gone as u64), 188 188 + items, 189 189 + next, 190 190 + }) 191 191 + } 192 192 + 193 193 + fn get_all_counts(&self, target: &str) -> Result<HashMap<String, HashMap<String, u64>>> { 194 194 + let data = self.0.lock().unwrap(); 195 195 + let mut out: HashMap<String, HashMap<String, u64>> = HashMap::new(); 196 196 + if let Some(asdf) = data.targets.get(&Target::new(target)) { 197 197 + for (Source { collection, path }, linkers) in asdf { 198 198 + let count = linkers.iter().flatten().count().try_into()?; 199 199 + out.entry(collection.to_string()) 200 200 + .or_default() 201 201 + .insert(path.to_string(), count); 202 202 + } 203 203 + } 204 204 + Ok(out) 119 205 } 120 206 121 207 fn summarize(&self, qsize: u32) { ··· 156 242 } 157 243 158 244 #[derive(Debug, PartialEq, Hash, Eq, Clone)] 245 245 + struct RKey(String); 246 246 + 247 247 + #[derive(Debug, PartialEq, Hash, Eq, Clone)] 159 248 struct RepoId { 160 249 collection: String, 161 161 - rkey: String, 250 250 + rkey: RKey, 162 251 } 163 252 164 253 impl RepoId { 165 254 fn from_record_id(record_id: &RecordId) -> Self { 166 255 Self { 167 256 collection: record_id.collection.clone(), 168 168 - rkey: record_id.rkey.clone(), 257 257 + rkey: RKey(record_id.rkey.clone()), 169 258 } 170 259 } 171 260 }

+416 -29

link_aggregator/src/storage/mod.rs

··· 1 1 use anyhow::Result; 2 2 - use link_aggregator::{ActionableEvent, Did, RecordId}; 3 3 - use links::CollectedLink; 2 2 + use link_aggregator::{ActionableEvent, RecordId}; 4 3 use std::collections::HashMap; 5 4 6 5 pub mod mem_store; ··· 11 10 #[cfg(feature = "rocks")] 12 11 pub use rocks_store::RocksStorage; 13 12 13 13 + #[derive(Debug, PartialEq)] 14 14 + pub struct PagedAppendingCollection<T> { 15 15 + pub version: (u64, u64), // (collection length, deleted item count) 16 16 + pub items: Vec<T>, 17 17 + pub next: Option<u64>, 18 18 + } 19 19 + 14 20 pub trait LinkStorage: Send + Sync { 21 21 + /// jetstream cursor from last saved actions, if available 15 22 fn get_cursor(&mut self) -> Result<Option<u64>> { 16 23 Ok(None) 17 17 - } 18 18 - fn push(&mut self, event: &ActionableEvent, _cursor: u64) -> Result<()> { 19 19 - match event { 20 20 - ActionableEvent::CreateLinks { record_id, links } => self.add_links(record_id, links), 21 21 - ActionableEvent::UpdateLinks { 22 22 - record_id, 23 23 - new_links, 24 24 - } => self.update_links(record_id, new_links), 25 25 - ActionableEvent::DeleteRecord(record_id) => self.remove_links(record_id), 26 26 - ActionableEvent::ActivateAccount(did) => self.set_account(did, true), 27 27 - ActionableEvent::DeactivateAccount(did) => self.set_account(did, false), 28 28 - ActionableEvent::DeleteAccount(did) => self.delete_account(did), 29 29 - } 30 30 - Ok(()) 31 24 } 32 25 33 33 - fn add_links(&mut self, record_id: &RecordId, links: &[CollectedLink]); 34 34 - fn remove_links(&mut self, record_id: &RecordId); 35 35 - fn update_links(&mut self, record_id: &RecordId, new_links: &[CollectedLink]) { 36 36 - self.remove_links(record_id); 37 37 - self.add_links(record_id, new_links); 38 38 - } 39 39 - fn set_account(&mut self, did: &Did, active: bool); 40 40 - fn delete_account(&mut self, did: &Did); 26 26 + fn push(&mut self, event: &ActionableEvent, cursor: u64) -> Result<()>; 41 27 42 28 // readers are off from the writer instance 43 29 fn to_readable(&mut self) -> impl LinkReader; ··· 45 31 46 32 pub trait LinkReader: Clone + Send + Sync + 'static { 47 33 fn get_count(&self, target: &str, collection: &str, path: &str) -> Result<u64>; 48 48 - fn get_all_counts(&self, _target: &str) -> Result<HashMap<String, HashMap<String, u64>>> { 49 49 - Ok(HashMap::new()) 50 50 - } 34 34 + 35 35 + fn get_links( 36 36 + &self, 37 37 + target: &str, 38 38 + collection: &str, 39 39 + path: &str, 40 40 + limit: u64, 41 41 + until: Option<u64>, 42 42 + ) -> Result<PagedAppendingCollection<RecordId>>; 43 43 + 44 44 + fn get_all_counts(&self, _target: &str) -> Result<HashMap<String, HashMap<String, u64>>>; 51 45 52 46 // todo: remove it 53 47 fn summarize(&self, qsize: u32) { ··· 58 52 #[cfg(test)] 59 53 mod tests { 60 54 use super::*; 61 61 - use links::Link; 55 55 + use links::{CollectedLink, Link}; 62 56 63 57 macro_rules! test_each_storage { 64 58 ($test_name:ident, |$storage_label:ident| $test_code:block) => { ··· 493 487 assert_eq!(storage.get_count("a.com", "app.t.c", ".def.uri")?, 0); 494 488 }); 495 489 496 496 - // todo: test update where the new record has no links -> this test might need to be in main? 490 490 + test_each_storage!(get_links_zero, |storage| { 491 491 + assert_eq!( 492 492 + storage.get_links("a.com", "app.t.c", ".abc.uri", 100, None)?, 493 493 + PagedAppendingCollection { 494 494 + version: (0, 0), 495 495 + items: vec![], 496 496 + next: None, 497 497 + } 498 498 + ); 499 499 + }); 500 500 + 501 501 + test_each_storage!(get_links_basic, |storage| { 502 502 + storage.push( 503 503 + &ActionableEvent::CreateLinks { 504 504 + record_id: RecordId { 505 505 + did: "did:plc:asdf".into(), 506 506 + collection: "app.t.c".into(), 507 507 + rkey: "asdf".into(), 508 508 + }, 509 509 + links: vec![CollectedLink { 510 510 + target: Link::Uri("a.com".into()), 511 511 + path: ".abc.uri".into(), 512 512 + }], 513 513 + }, 514 514 + 0, 515 515 + )?; 516 516 + assert_eq!( 517 517 + storage.get_links("a.com", "app.t.c", ".abc.uri", 100, None)?, 518 518 + PagedAppendingCollection { 519 519 + version: (1, 0), 520 520 + items: vec![RecordId { 521 521 + did: "did:plc:asdf".into(), 522 522 + collection: "app.t.c".into(), 523 523 + rkey: "asdf".into(), 524 524 + }], 525 525 + next: None, 526 526 + } 527 527 + ); 528 528 + }); 529 529 + 530 530 + test_each_storage!(get_links_paged, |storage| { 531 531 + for i in 1..=5 { 532 532 + storage.push( 533 533 + &ActionableEvent::CreateLinks { 534 534 + record_id: RecordId { 535 535 + did: format!("did:plc:asdf-{i}").into(), 536 536 + collection: "app.t.c".into(), 537 537 + rkey: "asdf".into(), 538 538 + }, 539 539 + links: vec![CollectedLink { 540 540 + target: Link::Uri("a.com".into()), 541 541 + path: ".abc.uri".into(), 542 542 + }], 543 543 + }, 544 544 + 0, 545 545 + )?; 546 546 + } 547 547 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, None)?; 548 548 + assert_eq!( 549 549 + links, 550 550 + PagedAppendingCollection { 551 551 + version: (5, 0), 552 552 + items: vec![ 553 553 + RecordId { 554 554 + did: "did:plc:asdf-5".into(), 555 555 + collection: "app.t.c".into(), 556 556 + rkey: "asdf".into(), 557 557 + }, 558 558 + RecordId { 559 559 + did: "did:plc:asdf-4".into(), 560 560 + collection: "app.t.c".into(), 561 561 + rkey: "asdf".into(), 562 562 + }, 563 563 + ], 564 564 + next: Some(3), 565 565 + } 566 566 + ); 567 567 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, links.next)?; 568 568 + assert_eq!( 569 569 + links, 570 570 + PagedAppendingCollection { 571 571 + version: (5, 0), 572 572 + items: vec![ 573 573 + RecordId { 574 574 + did: "did:plc:asdf-3".into(), 575 575 + collection: "app.t.c".into(), 576 576 + rkey: "asdf".into(), 577 577 + }, 578 578 + RecordId { 579 579 + did: "did:plc:asdf-2".into(), 580 580 + collection: "app.t.c".into(), 581 581 + rkey: "asdf".into(), 582 582 + }, 583 583 + ], 584 584 + next: Some(1), 585 585 + } 586 586 + ); 587 587 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, links.next)?; 588 588 + assert_eq!( 589 589 + links, 590 590 + PagedAppendingCollection { 591 591 + version: (5, 0), 592 592 + items: vec![RecordId { 593 593 + did: "did:plc:asdf-1".into(), 594 594 + collection: "app.t.c".into(), 595 595 + rkey: "asdf".into(), 596 596 + },], 597 597 + next: None, 598 598 + } 599 599 + ); 600 600 + }); 601 601 + 602 602 + test_each_storage!(get_links_exact_multiple, |storage| { 603 603 + for i in 1..=4 { 604 604 + storage.push( 605 605 + &ActionableEvent::CreateLinks { 606 606 + record_id: RecordId { 607 607 + did: format!("did:plc:asdf-{i}").into(), 608 608 + collection: "app.t.c".into(), 609 609 + rkey: "asdf".into(), 610 610 + }, 611 611 + links: vec![CollectedLink { 612 612 + target: Link::Uri("a.com".into()), 613 613 + path: ".abc.uri".into(), 614 614 + }], 615 615 + }, 616 616 + 0, 617 617 + )?; 618 618 + } 619 619 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, None)?; 620 620 + assert_eq!( 621 621 + links, 622 622 + PagedAppendingCollection { 623 623 + version: (4, 0), 624 624 + items: vec![ 625 625 + RecordId { 626 626 + did: "did:plc:asdf-4".into(), 627 627 + collection: "app.t.c".into(), 628 628 + rkey: "asdf".into(), 629 629 + }, 630 630 + RecordId { 631 631 + did: "did:plc:asdf-3".into(), 632 632 + collection: "app.t.c".into(), 633 633 + rkey: "asdf".into(), 634 634 + }, 635 635 + ], 636 636 + next: Some(2), 637 637 + } 638 638 + ); 639 639 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, links.next)?; 640 640 + assert_eq!( 641 641 + links, 642 642 + PagedAppendingCollection { 643 643 + version: (4, 0), 644 644 + items: vec![ 645 645 + RecordId { 646 646 + did: "did:plc:asdf-2".into(), 647 647 + collection: "app.t.c".into(), 648 648 + rkey: "asdf".into(), 649 649 + }, 650 650 + RecordId { 651 651 + did: "did:plc:asdf-1".into(), 652 652 + collection: "app.t.c".into(), 653 653 + rkey: "asdf".into(), 654 654 + }, 655 655 + ], 656 656 + next: None, 657 657 + } 658 658 + ); 659 659 + }); 660 660 + 661 661 + test_each_storage!(page_links_while_new_links_arrive, |storage| { 662 662 + for i in 1..=4 { 663 663 + storage.push( 664 664 + &ActionableEvent::CreateLinks { 665 665 + record_id: RecordId { 666 666 + did: format!("did:plc:asdf-{i}").into(), 667 667 + collection: "app.t.c".into(), 668 668 + rkey: "asdf".into(), 669 669 + }, 670 670 + links: vec![CollectedLink { 671 671 + target: Link::Uri("a.com".into()), 672 672 + path: ".abc.uri".into(), 673 673 + }], 674 674 + }, 675 675 + 0, 676 676 + )?; 677 677 + } 678 678 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, None)?; 679 679 + assert_eq!( 680 680 + links, 681 681 + PagedAppendingCollection { 682 682 + version: (4, 0), 683 683 + items: vec![ 684 684 + RecordId { 685 685 + did: "did:plc:asdf-4".into(), 686 686 + collection: "app.t.c".into(), 687 687 + rkey: "asdf".into(), 688 688 + }, 689 689 + RecordId { 690 690 + did: "did:plc:asdf-3".into(), 691 691 + collection: "app.t.c".into(), 692 692 + rkey: "asdf".into(), 693 693 + }, 694 694 + ], 695 695 + next: Some(2), 696 696 + } 697 697 + ); 698 698 + storage.push( 699 699 + &ActionableEvent::CreateLinks { 700 700 + record_id: RecordId { 701 701 + did: "did:plc:asdf-5".into(), 702 702 + collection: "app.t.c".into(), 703 703 + rkey: "asdf".into(), 704 704 + }, 705 705 + links: vec![CollectedLink { 706 706 + target: Link::Uri("a.com".into()), 707 707 + path: ".abc.uri".into(), 708 708 + }], 709 709 + }, 710 710 + 0, 711 711 + )?; 712 712 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, links.next)?; 713 713 + assert_eq!( 714 714 + links, 715 715 + PagedAppendingCollection { 716 716 + version: (5, 0), 717 717 + items: vec![ 718 718 + RecordId { 719 719 + did: "did:plc:asdf-2".into(), 720 720 + collection: "app.t.c".into(), 721 721 + rkey: "asdf".into(), 722 722 + }, 723 723 + RecordId { 724 724 + did: "did:plc:asdf-1".into(), 725 725 + collection: "app.t.c".into(), 726 726 + rkey: "asdf".into(), 727 727 + }, 728 728 + ], 729 729 + next: None, 730 730 + } 731 731 + ); 732 732 + }); 733 733 + 734 734 + test_each_storage!(page_links_while_some_are_deleted, |storage| { 735 735 + for i in 1..=4 { 736 736 + storage.push( 737 737 + &ActionableEvent::CreateLinks { 738 738 + record_id: RecordId { 739 739 + did: format!("did:plc:asdf-{i}").into(), 740 740 + collection: "app.t.c".into(), 741 741 + rkey: "asdf".into(), 742 742 + }, 743 743 + links: vec![CollectedLink { 744 744 + target: Link::Uri("a.com".into()), 745 745 + path: ".abc.uri".into(), 746 746 + }], 747 747 + }, 748 748 + 0, 749 749 + )?; 750 750 + } 751 751 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, None)?; 752 752 + assert_eq!( 753 753 + links, 754 754 + PagedAppendingCollection { 755 755 + version: (4, 0), 756 756 + items: vec![ 757 757 + RecordId { 758 758 + did: "did:plc:asdf-4".into(), 759 759 + collection: "app.t.c".into(), 760 760 + rkey: "asdf".into(), 761 761 + }, 762 762 + RecordId { 763 763 + did: "did:plc:asdf-3".into(), 764 764 + collection: "app.t.c".into(), 765 765 + rkey: "asdf".into(), 766 766 + }, 767 767 + ], 768 768 + next: Some(2), 769 769 + } 770 770 + ); 771 771 + storage.push( 772 772 + &ActionableEvent::DeleteRecord(RecordId { 773 773 + did: "did:plc:asdf-2".into(), 774 774 + collection: "app.t.c".into(), 775 775 + rkey: "asdf".into(), 776 776 + }), 777 777 + 0, 778 778 + )?; 779 779 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, links.next)?; 780 780 + assert_eq!( 781 781 + links, 782 782 + PagedAppendingCollection { 783 783 + version: (4, 1), 784 784 + items: vec![RecordId { 785 785 + did: "did:plc:asdf-1".into(), 786 786 + collection: "app.t.c".into(), 787 787 + rkey: "asdf".into(), 788 788 + },], 789 789 + next: None, 790 790 + } 791 791 + ); 792 792 + }); 793 793 + 794 794 + test_each_storage!(page_links_accounts_inactive, |storage| { 795 795 + for i in 1..=4 { 796 796 + storage.push( 797 797 + &ActionableEvent::CreateLinks { 798 798 + record_id: RecordId { 799 799 + did: format!("did:plc:asdf-{i}").into(), 800 800 + collection: "app.t.c".into(), 801 801 + rkey: "asdf".into(), 802 802 + }, 803 803 + links: vec![CollectedLink { 804 804 + target: Link::Uri("a.com".into()), 805 805 + path: ".abc.uri".into(), 806 806 + }], 807 807 + }, 808 808 + 0, 809 809 + )?; 810 810 + } 811 811 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, None)?; 812 812 + assert_eq!( 813 813 + links, 814 814 + PagedAppendingCollection { 815 815 + version: (4, 0), 816 816 + items: vec![ 817 817 + RecordId { 818 818 + did: "did:plc:asdf-4".into(), 819 819 + collection: "app.t.c".into(), 820 820 + rkey: "asdf".into(), 821 821 + }, 822 822 + RecordId { 823 823 + did: "did:plc:asdf-3".into(), 824 824 + collection: "app.t.c".into(), 825 825 + rkey: "asdf".into(), 826 826 + }, 827 827 + ], 828 828 + next: Some(2), 829 829 + } 830 830 + ); 831 831 + storage.push( 832 832 + &ActionableEvent::DeactivateAccount("did:plc:asdf-1".into()), 833 833 + 0, 834 834 + )?; 835 835 + let links = storage.get_links("a.com", "app.t.c", ".abc.uri", 2, links.next)?; 836 836 + assert_eq!( 837 837 + links, 838 838 + PagedAppendingCollection { 839 839 + version: (4, 0), 840 840 + items: vec![RecordId { 841 841 + did: "did:plc:asdf-2".into(), 842 842 + collection: "app.t.c".into(), 843 843 + rkey: "asdf".into(), 844 844 + },], 845 845 + next: None, 846 846 + } 847 847 + ); 848 848 + }); 849 849 + 850 850 + test_each_storage!(get_all_counts, |storage| { 851 851 + storage.push( 852 852 + &ActionableEvent::CreateLinks { 853 853 + record_id: RecordId { 854 854 + did: "did:plc:asdf".into(), 855 855 + collection: "app.t.c".into(), 856 856 + rkey: "asdf".into(), 857 857 + }, 858 858 + links: vec![ 859 859 + CollectedLink { 860 860 + target: Link::Uri("a.com".into()), 861 861 + path: ".abc.uri".into(), 862 862 + }, 863 863 + CollectedLink { 864 864 + target: Link::Uri("a.com".into()), 865 865 + path: ".def.uri".into(), 866 866 + }, 867 867 + ], 868 868 + }, 869 869 + 0, 870 870 + )?; 871 871 + assert_eq!(storage.get_all_counts("a.com")?, { 872 872 + let mut counts = HashMap::new(); 873 873 + let mut t_c_counts = HashMap::new(); 874 874 + t_c_counts.insert(".abc.uri".into(), 1); 875 875 + t_c_counts.insert(".def.uri".into(), 1); 876 876 + counts.insert("app.t.c".into(), t_c_counts); 877 877 + counts 878 878 + }); 879 879 + }); 880 880 + 881 881 + test_each_storage!(get_all_counts_no_links, |storage| { 882 882 + assert_eq!(storage.get_all_counts("bad-example.com")?, HashMap::new()); 883 883 + }); 497 884 }

+179 -42

link_aggregator/src/storage/rocks_store.rs

··· 1 1 - use super::{ActionableEvent, LinkReader, LinkStorage}; 1 1 + use super::{ActionableEvent, LinkReader, LinkStorage, PagedAppendingCollection}; 2 2 use anyhow::{bail, Result}; 3 3 use bincode::Options as BincodeOptions; 4 4 use link_aggregator::{Did, RecordId}; ··· 42 42 #[derive(Debug, Clone)] 43 43 pub struct RocksStorage { 44 44 db: Arc<DBWithThreadMode<MultiThreaded>>, // TODO: mov seqs here (concat merge op will be fun) 45 45 - did_id_table: IdTable<Did, DidIdValue>, 46 46 - target_id_table: IdTable<TargetKey, TargetId>, 45 45 + did_id_table: IdTable<Did, DidIdValue, true>, 46 46 + target_id_table: IdTable<TargetKey, TargetId, false>, 47 47 is_writer: bool, 48 48 } 49 49 50 50 trait IdTableValue: ValueFromRocks + Clone { 51 51 fn new(v: u64) -> Self; 52 52 + fn id(&self) -> u64; 52 53 } 53 54 #[derive(Debug, Clone)] 54 55 struct IdTableBase<Orig, IdVal: IdTableValue> 55 56 where 57 57 + Orig: KeyFromRocks, 56 58 for<'a> &'a Orig: AsRocksKey, 57 59 { 58 60 _key_marker: PhantomData<Orig>, ··· 63 65 } 64 66 impl<Orig, IdVal: IdTableValue> IdTableBase<Orig, IdVal> 65 67 where 68 68 + Orig: KeyFromRocks, 66 69 for<'a> &'a Orig: AsRocksKey, 67 70 { 68 71 fn cf_descriptor(&self) -> ColumnFamilyDescriptor { 69 72 ColumnFamilyDescriptor::new(&self.name, rocks_opts_base()) 70 73 } 71 71 - fn init(mut self, db: &DBWithThreadMode<MultiThreaded>) -> Result<IdTable<Orig, IdVal>> { 74 74 + fn init<const WITH_REVERSE: bool>( 75 75 + mut self, 76 76 + db: &DBWithThreadMode<MultiThreaded>, 77 77 + ) -> Result<IdTable<Orig, IdVal, WITH_REVERSE>> { 72 78 if db.cf_handle(&self.name).is_none() { 73 79 bail!("failed to get cf handle from db -- was the db open with our .cf_descriptor()?"); 74 80 } ··· 101 107 } 102 108 impl<O, I: IdTableValue> Drop for IdTableBase<O, I> 103 109 where 110 110 + O: KeyFromRocks, 104 111 for<'a> &'a O: AsRocksKey, 105 112 { 106 113 fn drop(&mut self) { ··· 113 120 } 114 121 } 115 122 #[derive(Debug, Clone)] 116 116 - struct IdTable<Orig, IdVal: IdTableValue> 123 123 + struct IdTable<Orig, IdVal: IdTableValue, const WITH_REVERSE: bool> 117 124 where 125 125 + Orig: KeyFromRocks, 118 126 for<'a> &'a Orig: AsRocksKey, 119 127 { 120 128 base: IdTableBase<Orig, IdVal>, 121 129 priv_id_seq: u64, 122 130 } 123 123 - impl<Orig: Clone, IdVal: IdTableValue> IdTable<Orig, IdVal> 131 131 + impl<Orig: Clone, IdVal: IdTableValue, const WITH_REVERSE: bool> IdTable<Orig, IdVal, WITH_REVERSE> 124 132 where 133 133 + Orig: KeyFromRocks, 125 134 for<'v> &'v IdVal: AsRocksValue, 126 135 for<'k> &'k Orig: AsRocksKey, 127 136 { ··· 147 156 Ok(None) 148 157 } 149 158 } 150 150 - fn get_or_create_id_val( 159 159 + fn __get_or_create_id_val<CF>( 151 160 &mut self, 161 161 + cf: &CF, 152 162 db: &DBWithThreadMode<MultiThreaded>, 153 163 batch: &mut WriteBatch, 154 164 orig: &Orig, 155 155 - ) -> Result<IdVal> { 156 156 - let cf = db.cf_handle(&self.base.name).unwrap(); 165 165 + ) -> Result<IdVal> 166 166 + where 167 167 + CF: AsColumnFamilyRef, 168 168 + { 157 169 Ok(self.get_id_val(db, orig)?.unwrap_or_else(|| { 158 170 let prev_priv_seq = self.priv_id_seq; 159 171 self.priv_id_seq += 1; ··· 164 176 ); 165 177 let id_value = IdVal::new(self.priv_id_seq); 166 178 batch.put(self.base.seq_key(), self.priv_id_seq.to_le_bytes()); 167 167 - batch.put_cf(&cf, _rk(orig), _rv(&id_value)); 179 179 + batch.put_cf(cf, _rk(orig), _rv(&id_value)); 168 180 id_value 169 181 })) 170 182 } 171 183 } 184 184 + impl<Orig: Clone, IdVal: IdTableValue> IdTable<Orig, IdVal, true> 185 185 + where 186 186 + Orig: KeyFromRocks, 187 187 + for<'v> &'v IdVal: AsRocksValue, 188 188 + for<'k> &'k Orig: AsRocksKey, 189 189 + { 190 190 + fn get_or_create_id_val( 191 191 + &mut self, 192 192 + db: &DBWithThreadMode<MultiThreaded>, 193 193 + batch: &mut WriteBatch, 194 194 + orig: &Orig, 195 195 + ) -> Result<IdVal> { 196 196 + let cf = db.cf_handle(&self.base.name).unwrap(); 197 197 + let id_val = self.__get_or_create_id_val(&cf, db, batch, orig)?; 198 198 + // TODO: assert that the original is never a u64 that could collide 199 199 + batch.put_cf(&cf, id_val.id().to_be_bytes(), _rk(orig)); // reversed rk/rv on purpose here :/ 200 200 + Ok(id_val) 201 201 + } 202 202 + 203 203 + fn get_val_from_id( 204 204 + &self, 205 205 + db: &DBWithThreadMode<MultiThreaded>, 206 206 + id: u64, 207 207 + ) -> Result<Option<Orig>> { 208 208 + let cf = db.cf_handle(&self.base.name).unwrap(); 209 209 + if let Some(orig_bytes) = db.get_cf(&cf, id.to_be_bytes())? { 210 210 + // HACK ish 211 211 + Ok(Some(_kr(&orig_bytes)?)) 212 212 + } else { 213 213 + Ok(None) 214 214 + } 215 215 + } 216 216 + } 217 217 + impl<Orig: Clone, IdVal: IdTableValue> IdTable<Orig, IdVal, false> 218 218 + where 219 219 + Orig: KeyFromRocks, 220 220 + for<'v> &'v IdVal: AsRocksValue, 221 221 + for<'k> &'k Orig: AsRocksKey, 222 222 + { 223 223 + fn get_or_create_id_val( 224 224 + &mut self, 225 225 + db: &DBWithThreadMode<MultiThreaded>, 226 226 + batch: &mut WriteBatch, 227 227 + orig: &Orig, 228 228 + ) -> Result<IdVal> { 229 229 + let cf = db.cf_handle(&self.base.name).unwrap(); 230 230 + self.__get_or_create_id_val(&cf, db, batch, orig) 231 231 + } 232 232 + } 172 233 173 234 impl IdTableValue for DidIdValue { 174 235 fn new(v: u64) -> Self { 175 236 DidIdValue(DidId(v), true) 237 237 + } 238 238 + fn id(&self) -> u64 { 239 239 + self.0 .0 176 240 } 177 241 } 178 242 impl IdTableValue for TargetId { 179 243 fn new(v: u64) -> Self { 180 244 TargetId(v) 181 245 } 246 246 + fn id(&self) -> u64 { 247 247 + self.0 248 248 + } 182 249 } 183 250 184 251 impl RocksStorage { 185 252 pub fn new(path: impl AsRef<Path>) -> Result<Self> { 186 186 - let did_id_table = IdTable::setup(DID_IDS_CF); 187 187 - let target_id_table = IdTable::setup(TARGET_IDS_CF); 253 253 + let did_id_table = IdTable::<_, _, true>::setup(DID_IDS_CF); 254 254 + let target_id_table = IdTable::<_, _, false>::setup(TARGET_IDS_CF); 188 255 189 256 let db = DBWithThreadMode::open_cf_descriptors( 190 257 &get_db_opts(), ··· 230 297 231 298 let current_seq = current_did_id_seq.load(Ordering::SeqCst); 232 299 233 233 - for did_id in &tls.0 { 234 234 - let DidId(ref n) = did_id; 300 300 + for did_id_rkey in &tls.0 { 301 301 + let DidId(ref n) = did_id_rkey.0; 235 302 if current_seq > 0 && *n > current_seq { 236 303 eprintln!("problem with merge_op_extend_did_ids. existing: {tls:?}"); 237 304 eprintln!( ··· 243 310 244 311 for op in operands { 245 312 let new_linkers: TargetLinkers = _vr(op).unwrap(); 246 246 - for DidId(ref n) in &new_linkers.0 { 313 313 + for (DidId(ref n), _) in &new_linkers.0 { 247 314 if current_seq > 0 && *n > current_seq { 248 315 let orig: Option<TargetLinkers> = 249 316 existing.map(|existing_bytes| _vr(existing_bytes).unwrap()); ··· 254 321 panic!("did_id a did to a number higher than the current sequence"); 255 322 } 256 323 } 257 257 - tls.0.extend(&new_linkers.0); 324 324 + tls.0.extend(new_linkers.0); 258 325 } 259 326 Some(_rv(&tls)) 260 327 } ··· 309 376 batch: &mut WriteBatch, 310 377 target_id: &TargetId, 311 378 linker_did_id: &DidId, 379 379 + linker_rkey: &RKey, 312 380 ) { 313 381 let cf = self.db.cf_handle(TARGET_LINKERS_CF).unwrap(); 314 382 batch.merge_cf( 315 383 &cf, 316 384 _rk(target_id), 317 317 - _rv(&TargetLinkers(vec![*linker_did_id])), 385 385 + _rv(&TargetLinkers(vec![(*linker_did_id, linker_rkey.clone())])), 318 386 ); 319 387 } 320 388 fn update_target_linkers<F>( ··· 434 502 .target_id_table 435 503 .get_or_create_id_val(&self.db, batch, &target_key) 436 504 .unwrap(); 437 437 - self.merge_target_linker(batch, &target_id, &did_id); 505 505 + self.merge_target_linker(batch, &target_id, &did_id, &RKey(record_id.rkey())); 438 506 439 507 record_link_targets.add(RecordLinkTarget(RPath(path.clone()), target_id)) 440 508 } ··· 478 546 eprintln!("working on #{i}: {rpath:?} / {target_id:?}"); 479 547 panic!("it was empty"); 480 548 } 481 481 - if !linkers.remove_last_linker(&linking_did_id) { 549 549 + if !linkers.remove_linker(&linking_did_id, &RKey(record_id.rkey.clone())) { 482 550 eprintln!("about to blow up because a linked target apparently does not have us in its dids."); 483 551 eprintln!("removing links for: {record_id:?}"); 484 552 eprintln!("found links: {record_link_targets:?}"); ··· 520 588 return; // ignore updates for dids we don't know about 521 589 }; 522 590 self.delete_did_id_value(batch, did); 591 591 + // TODO: also delete the reverse!! 523 592 524 593 // use a separate batch for all their links, since it can be a lot and make us crash at around 1GiB batch size. 525 594 // this should still hopefully be crash-safe: as long as we don't actually delete the DidId entry until after all links are cleared. ··· 535 604 536 605 for (j, RecordLinkTarget(path, target_link_id)) in links.0.iter().enumerate() { 537 606 self.update_target_linkers(&mut mini_batch, target_link_id, |mut linkers| { 538 538 - if !linkers.remove_last_linker(&did_id) { 607 607 + if !linkers.remove_linker(&did_id, &record_link_key.2) { 539 608 eprintln!( 540 609 "DELETING ACCOUNT: blowing up: missing linker entry in linked target." 541 610 ); ··· 621 690 Ok(()) 622 691 } 623 692 624 624 - fn add_links(&mut self, _record_id: &RecordId, _links: &[CollectedLink]) { 625 625 - unimplemented!("customized push"); 626 626 - } 627 627 - fn remove_links(&mut self, _record_id: &RecordId) { 628 628 - unimplemented!("customized push"); 629 629 - } 630 630 - fn set_account(&mut self, _did: &Did, _active: bool) { 631 631 - unimplemented!("customized push"); 632 632 - } 633 633 - fn delete_account(&mut self, _did: &Did) { 634 634 - unimplemented!("customized push"); 635 635 - } 636 636 - 637 693 fn to_readable(&mut self) -> impl LinkReader { 638 694 let mut readable = self.clone(); 639 695 readable.is_writer = false; ··· 647 703 let target_seq = self.target_id_table.base.id_seq.load(Ordering::SeqCst); 648 704 println!("queue: {qsize}. did seq: {did_seq}, target seq: {target_seq}."); 649 705 } 706 706 + 650 707 fn get_count(&self, target: &str, collection: &str, path: &str) -> Result<u64> { 651 708 let target_key = TargetKey( 652 709 Target(target.to_string()), ··· 654 711 RPath(path.to_string()), 655 712 ); 656 713 if let Some(target_id) = self.target_id_table.get_id_val(&self.db, &target_key)? { 657 657 - Ok(self.get_target_linkers(&target_id)?.count()) 714 714 + let (alive, _) = self.get_target_linkers(&target_id)?.count(); 715 715 + Ok(alive) 658 716 } else { 659 717 Ok(0) 660 718 } 661 719 } 720 720 + 721 721 + fn get_links( 722 722 + &self, 723 723 + target: &str, 724 724 + collection: &str, 725 725 + path: &str, 726 726 + limit: u64, 727 727 + until: Option<u64>, 728 728 + ) -> Result<PagedAppendingCollection<RecordId>> { 729 729 + let target_key = TargetKey( 730 730 + Target(target.to_string()), 731 731 + Collection(collection.to_string()), 732 732 + RPath(path.to_string()), 733 733 + ); 734 734 + 735 735 + let Some(target_id) = self.target_id_table.get_id_val(&self.db, &target_key)? else { 736 736 + return Ok(PagedAppendingCollection { 737 737 + version: (0, 0), 738 738 + items: Vec::new(), 739 739 + next: None, 740 740 + }); 741 741 + }; 742 742 + 743 743 + let linkers = self.get_target_linkers(&target_id)?; 744 744 + 745 745 + let (alive, gone) = linkers.count(); 746 746 + let total = alive + gone; 747 747 + let end = until.map(|u| std::cmp::min(u, total)).unwrap_or(total) as usize; 748 748 + let begin = end.saturating_sub(limit as usize); 749 749 + let next = if begin == 0 { None } else { Some(begin as u64) }; 750 750 + 751 751 + let did_id_rkeys = linkers.0[begin..end].iter().rev().collect::<Vec<_>>(); 752 752 + 753 753 + let mut items = Vec::with_capacity(did_id_rkeys.len()); 754 754 + // TODO: use get-many (or multi-get or whatever it's called) 755 755 + for (did_id, rkey) in did_id_rkeys { 756 756 + if let Some(did) = self.did_id_table.get_val_from_id(&self.db, did_id.0)? { 757 757 + let Some(DidIdValue(_, active)) = self.did_id_table.get_id_val(&self.db, &did)? 758 758 + else { 759 759 + eprintln!("failed to look up did_value from did_id {did_id:?}: {did:?}: data consistency bug?"); 760 760 + continue; 761 761 + }; 762 762 + if !active { 763 763 + continue; 764 764 + } 765 765 + items.push(RecordId { 766 766 + did, 767 767 + collection: collection.to_string(), 768 768 + rkey: rkey.0.clone(), 769 769 + }); 770 770 + } else { 771 771 + eprintln!("failed to look up did from did_id {did_id:?}"); 772 772 + } 773 773 + } 774 774 + 775 775 + Ok(PagedAppendingCollection { 776 776 + version: (total, gone), 777 777 + items, 778 778 + next, 779 779 + }) 780 780 + } 781 781 + 662 782 fn get_all_counts(&self, target: &str) -> Result<HashMap<String, HashMap<String, u64>>> { 663 783 let mut out: HashMap<String, HashMap<String, u64>> = HashMap::new(); 664 784 for (target_key, target_id) in self.iter_targets_for_target(&Target(target.into())) { 665 785 let TargetKey(_, Collection(ref collection), RPath(ref path)) = target_key; 666 666 - let count = self.get_target_linkers(&target_id)?.count(); 786 786 + let (count, _) = self.get_target_linkers(&target_id)?.count(); 667 787 out.entry(collection.into()) 668 788 .or_default() 669 789 .insert(path.clone(), count); ··· 685 805 686 806 // temp 687 807 impl KeyFromRocks for Did {} 808 808 + impl AsRocksKey for &DidId {} 688 809 689 810 // target_ids table 690 811 impl AsRocksKey for &TargetKey {} ··· 730 851 #[derive(Debug, Clone, Serialize, Deserialize)] 731 852 struct RPath(String); 732 853 733 733 - #[derive(Debug, Clone, Serialize, Deserialize)] 854 854 + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] 734 855 struct RKey(String); 735 856 857 857 + impl RKey { 858 858 + fn empty() -> Self { 859 859 + RKey("".to_string()) 860 860 + } 861 861 + } 862 862 + 736 863 // did ids 737 864 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] 738 865 struct DidId(u64); 866 866 + 867 867 + impl DidId { 868 868 + fn empty() -> Self { 869 869 + DidId(0) 870 870 + } 871 871 + } 739 872 740 873 #[derive(Debug, Clone, Serialize, Deserialize)] 741 874 struct DidIdValue(DidId, bool); // active or not ··· 759 892 struct TargetKey(Target, Collection, RPath); 760 893 761 894 #[derive(Debug, Default, Serialize, Deserialize)] 762 762 - struct TargetLinkers(Vec<DidId>); 895 895 + struct TargetLinkers(Vec<(DidId, RKey)>); 763 896 764 897 impl TargetLinkers { 765 765 - fn remove_last_linker(&mut self, did: &DidId) -> bool { 766 766 - if let Some(last_position) = self.0.iter().rposition(|d| d == did) { 767 767 - self.0.remove(last_position); 898 898 + fn remove_linker(&mut self, did: &DidId, rkey: &RKey) -> bool { 899 899 + if let Some(entry) = self.0.iter_mut().rfind(|d| **d == (*did, rkey.clone())) { 900 900 + *entry = (DidId::empty(), RKey::empty()); 768 901 true 769 902 } else { 770 903 false 771 904 } 772 905 } 773 773 - fn count(&self) -> u64 { 774 774 - self.0.len() as u64 906 906 + fn count(&self) -> (u64, u64) { 907 907 + // (linkers, deleted links) 908 908 + let total = self.0.len() as u64; 909 909 + let alive = self.0.iter().filter(|(DidId(id), _)| *id != 0).count() as u64; 910 910 + let gone = total - alive; 911 911 + (alive, gone) 775 912 } 776 913 } 777 914