Side A fixes a real, concrete bug: nested RwLock reads across match arms that would deadlock the RoomCreate/RoomGrant RPC paths (tokio::sync::RwLock is not reentrant), and backs it with a new integration test plus fixes to flaky/deadlock-prone test infra (log-file redirection, HTTP timeouts) and incorrect test assertions. Side B is a broad but mostly mechanical refactor (renaming from_storage_str->from_stored, threading CanonicalItemUrl/newtype Deref impls through many call sites) that improves type-safety but doesn't fix an active defect, making it closer to structural churn than a critical correctness fix.
constitution · epochs · watch · epoch 3
comparison
c_9608dc0d38ab (tommy-mor) vs c_c42f908efc44 (tommy-mor)
download prompt · raw event · cmp_4d9c191a517ffc
council reasoning
A fixes real tokio RwLock deadlocks by scoping verify/cap/user-exists guards so nested read/write cannot hold a non-reentrant guard across the match, and backs that with a room-create integration test plus test harness fixes (log-file piping, HTTP timeouts) that stop false hangs. B is a solid type-safety refactor (Deref on URL newtypes, CanonicalItemUrl through resolve_item/pools/validate) that removes string wrap/clone noise but does not correct broken runtime behavior.
Side A fixes a concrete concurrency bug by shortening the lifetime of `state.reduced.read().await` guards before later `read()`/`write()` operations, explicitly preventing Tokio `RwLock` self-deadlocks in `RoomCreate` and `RoomGrant`, and adds an integration test covering private room creation. Side B is largely a type-safety refactor that replaces many `String` usages with `CanonicalItemUrl` and adds `Deref` implementations, improving API cleanliness but with less direct impact on runtime correctness than the deadlock fix.
sides
A — c_9608dc0d38ab (tommy-mor)
message
[9ecc4e2e] nice
diff preview
diff --git a/server/src/api/rpc.rs b/server/src/api/rpc.rs
index 5675dcd2e28062acbe3c037b94b77c33adf32474..a18241f3ed7b4a248748fcefc799f9801ca62461 100644
--- a/server/src/api/rpc.rs
+++ b/server/src/api/rpc.rs
@@ -936,7 +936,15 @@ pub async fn handle_rpc_batch(
line_ok(RpcResult::ForumThreads(rpc_list_forum_threads(&reduced, &room)))
}
RpcCommand::RoomCreate { slug, visibility } => {
- match verify_bearer_principal(&headers, &*state.reduced.read().await) {
+ // Scope the first read so its guard drops before any nested `read().await` / `write().await`.
+ // A guard from `match verify(..., &*state.reduced.read().await)` would otherwise live for the
+ // whole `match` and deadlock here (tokio::sync::RwLock is not reentrant).
+ let principal = {
+ let reduced = state.reduced.read().await;
+ verify_bearer_principal(&headers, &*reduced)
+ };
+ match principal {
+ Err((_, m)) => line_err(m, None),
Ok(principal) => {
let slug = slug.trim().to_lowercase();
if slug.is_empty() || slug.len() > 64 {
@@ -994,7 +1002,6 @@ pub async fn handle_rpc_batch(
}
}
}
- Err((_, m)) => line_err(m, None),
}
}
RpcCommand::RoomGrant {
@@ -1002,17 +1009,28 @@ pub async fn handle_rpc_batch(
username,
capability,
} => {
- match verify_bearer_principal(&headers, &*state.reduced.read().await) {
+ let principal = {
+ let reduced = state.reduced.read().await;
+ verify_bearer_principal(&headers, &*reduced)
+ };
+ match principal {
Err((_, m)) => line_err(m, None),
Ok(principal) => {
- let reduced = state.reduced.read().await;
- if !reduced.user_has_cap(&room, &principal, ThreadCapability::Manage) {
+ let can_manage = {
+ let reduced = state.reduced.read().await;
+ reduced.user_has_cap(&room, &principal, ThreadCapability::Manage)
+ };
+ if !can_manage {
line_err("requires Manage capability", None)
} else {
match parse_username(&username) {
Err(msg) => line_err("invalid username", Some(msg)),
Ok(target) => {
- if !reduced.users_by_provider.values().any(|u| u == &target) {
+ let user_exists = {
+ let reduced = state.reduced.read().await;
+ reduced.users_by_provider.values().any(|u| u == &target)
+ };
+ if !user_exists {
line_err(format!("user @{target} not found"), None)
} else {
match parse_capability(&capability) {
diff --git a/server/tests/integration.rs b/server/tests/integration.rs
index 9162bd5bee84035e3908bfb9c8e201b7878ca339..b930120da09fe7d307f0411b84fb639fb8bd0b15 100644
--- a/server/tests/integration.rs
+++ b/server/tests/integration.rs
@@ -95,6 +95,23 @@ async fn test_healthz() {
assert_eq!(response.text().await.unwrap(), "ok");
}
+#[tokio::test]
+async fn test_room_create_private_rpc() {
+ let (addr, _tmp, _log, _handle) = create_test_server().await;
+ let client = reqwest::Client::new();
+ let batch = serde_json::json!([{
+ "RoomCreate": { "slug": "secret-project", "visibility": "private" }
+ }]);
+ let body = rpc_batch(&client, addr, Some(&test_bearer()), batch).await;
+ let line = &body["results"][0];
+ assert_eq!(line["ok"], true, "room create: {:?}", line);
+ let room_id = line["result"]["RoomCreated"]["room_id"].as_str().unwrap();
+ assert!(
+ room_id.contains("/secret-project"),
+ "expected room_id to contain slug, got {room_id}"
+ );
+}
+
#[tokio::test]
async fn test_index_page() {
// HTML routes are offline during the auth-v3 refactor.
diff --git a/test/auth.bb b/test/auth.bb
index 611e04f1806ef81d678a91f499597fe55f691dd7..a67cc1de763434176d2f68e419dd37a8cd328395 100644
--- a/test/auth.bb
+++ b/test/auth.bb
@@ -176,7 +176,7 @@
(assert! (= 200 (:status poll)) "pending-session poll returns 200")
(let [poll-json (json/parse-string (:body poll) true)]
(assert! (:complete poll-json) "pending session complete=true")
- (assert! (= "@bbuser" (:user poll-json)) "poll returns @bbuser")
+ (assert! (= "bbuser" (:user poll-json)) "poll returns stored username bbuser")
(assert! (clojure.string/starts-with? (:token poll-json) "slug_") "poll returns bearer token")
(println "\nwhoami…")
@@ -184,7 +184,7 @@
:headers {"Authorization" (str "Bearer " (:token poll-json))})]
(assert! (= 200 (:status who)) "whoami returns 200")
(let [who-json (json/parse-string (:body who) true)]
- (assert! (= "@bbuser" (:user who-json)) "whoami user is @bbuser"))))))
+ (assert! (= "bbuser" (:user who-json)) "whoami user is bbuser (stored form)"))))))
(println "\nCLI: identity start → OAuth → identity poll → whoami…")
(let [cli-home (str tmp-dir "/cli-home")
@@ -208,7 +208,7 @@
(str "identity poll exits 0 (stderr: " (:err poll-proc) ")"))
(let [poll-cli (json/parse-string (:out poll-proc) true)]
(assert! (= "complete" (:phase poll-cli)) "identity poll --json phase")
- (assert! (= "@cliuser" (:user poll-cli)) "CLI poll user")
+ (assert! (= "cliuser" (:user poll-cli)) "CLI poll user (stored form)")
(assert! (clojure.string/starts-with? (:token poll-cli) "slug_") "CLI poll token")
(let [token-path (str cli-home "/.config/slugsocial/token")]
(assert! (fs/exists? token-path) "token written under isolated HOME")
@@ -219,7 +219,7 @@
(assert! (zero? (:exit who-proc))
(str "whoami exits 0 (stderr: " (:err who-proc) ")"))
(let [who-cli (json/parse-string (:out who-proc) true)]
- (assert! (= "@cliuser" (:user who-cli)) "CLI whoami uses saved token"))))))))
+ (assert! (= "cliuser" (:user who-cli)) "CLI whoami uses saved token"))))))))
(finally
(when-some [s @!server] (common/kill-server s))
diff --git a/test/common.bb b/test/common.bb
index ebb16c615a8b40e8830b5d5765d1e935a45538d0..4ed450377cdbb020f5ff164a812dfbbfc23c0f47 100644
--- a/test/common.bb
+++ b/test/common.bb
@@ -78,9 +78,20 @@
(defn start-server
"Start the slugsocial-server binary with the given env map.
- Returns the babashka.process map."
- [server-bin env-map]
- (p/process [server-bin] {:out :inherit :err :inherit :env env-map}))
+ Returns the babashka.process map.
+
+ When `log-file` (string path) is provided, stdout and stderr are appended there
+ instead of inheriting the parent descriptors. Inheriting shared pipes while the
+ parent blocks on HTTP I/O can fill the pipe buffer and deadlock the server on log writes."
+ ([server-bin env-map]
+ (start-server server-bin env-map nil))
+ ([server-bin env-map log-file]
+ (p/process [server-bin]
+ (if log-file
+ ;; Two string paths (same file): babashka.process can deref the process cleanly.
+ ;; :err :out + ProcessBuilder$Redirect breaks stream copying in deref/kill-server.
+ {:env env-map :out log-file :err log-file}
+ {:out :inherit :err :inherit :env env-map}))))
(defn kill-server
"Forcibly kill a server process (babashka.process map) and wait for it to exit."
diff --git a/test/grants.bb b/test/grants.bb
index 7066c1f2a57f39a362052f8524206dccf37bb7b1..793ba216f2de76a77eb20a76d08403db07ff411b 100644
--- a/test/grants.bb
+++ b/test/grants.bb
@@ -35,13 +35,14 @@
(defn- http-client []
(-> (java.net.http.HttpClient/newBuilder)
(.followRedirects java.net.http.HttpClient$Redirect/ALWAYS)
+ (.connectTimeout (java.time.Duration/ofSeconds 15))
(.build)))
(defn- http-get [url & {:keys [headers]}]
(let [b (java.net.http.HttpRequest/newBuilder (java.net.URI/create url))]
(doseq [[k v] (or headers {})]
(.header b k v))
- (let [req (-> b (.GET) (.build))
+ (let [req (-> b (.timeout (java.time.Duration/ofSeconds 60)) (.GET) (.build))
resp (.send (http-client) req (java.net.http.HttpResponse$BodyHandlers/ofString))]
{:status (.statusCode resp) :body (.body resp)})))
@@ -52,6 +53,7 @@
(doseq [[k v] (or headers {})]
(.header b k v))
(let [req (-> b
+ (.timeout (java.time.Duration/ofSeconds 60))
(.POST (java.net.http.HttpRequest$BodyPublishers/ofString body))
(.build))
resp (.send (http-client) req (java.net.http.HttpResponse$BodyHandlers/ofString))]
@@ -67,6 +69,7 @@
b (java.net.http.HttpRequest/newBuilder (java.net.URI/create url))]
(.header b "Content-Type" "application/x-www-form-urlencoded")
(let [req (-> b
+ (.timeout (java.time.Duration/ofSeconds 60))
(.POST (java.net.http.HttpRequest$BodyPublishers/ofString pairs))
(.build))
resp (.send (http-client) req (java.net.http.HttpResponse$BodyHandlers/ofString))]
B — c_c42f908efc44 (tommy-mor)
message
[674964ef] refactor: Deref for href newtypes, CanonicalItemUrl through resolve_item - Implement Deref<Target=str> for GardenItemUrl, ForumThreadUrl, TildeOntologyPath - resolve_item returns CanonicalItemUrl; validate uses HashSet<CanonicalItemUrl> - compute_scope_rank_changes keys are CanonicalItemUrl; pair RPC uses Vec pool - pick_random_distinct_canonical; connectivity stats on &[CanonicalItemUrl] - Global rank unranked uses stored ids before GardenItemUrl mapping Made-with: Cursor
diff preview
diff --git a/server/src/api/helpers.rs b/server/src/api/helpers.rs
index 03b3e77911ccd662bec8635345dafe2593cf242e..1b291db83df7364a026f2e147e0a29a70a399371 100644
--- a/server/src/api/helpers.rs
+++ b/server/src/api/helpers.rs
@@ -30,13 +30,13 @@ pub fn now_ms() -> i64 {
t.as_millis() as i64
}
-/// Resolve an item path as a first-class canonical path.
-pub fn resolve_item(item: &str) -> Result<String, String> {
+/// Resolve DSL/user input to a stored canonical item id.
+pub fn resolve_item(item: &str) -> Result<CanonicalItemUrl, String> {
let canonical = canonicalize_item(item);
if canonical.is_empty() {
return Err(format!("empty item path: `{}`", item));
}
- Ok(canonical)
+ Ok(CanonicalItemUrl(canonical))
}
pub fn parse_parent_specs(parent: Option<&String>) -> Vec<String> {
@@ -94,7 +94,7 @@ pub fn paginate_rankings(
(out_components, out_unranked)
}
-pub fn pick_random_distinct(items: &[String]) -> Option<(String, String)> {
+pub fn pick_random_distinct_canonical(items: &[CanonicalItemUrl]) -> Option<(CanonicalItemUrl, CanonicalItemUrl)> {
use rand::seq::SliceRandom;
if items.len() < 2 {
return None;
@@ -123,15 +123,12 @@ pub fn is_pair_voted(group: &crate::reducer::GroupState, a: &str, b: &str) -> bo
group.voted_pairs.contains(&(i, j))
}
-pub fn compute_connectivity_stats(group: &crate::reducer::GroupState, pool: &[String]) -> ConnectivityStats {
+pub fn compute_connectivity_stats(group: &crate::reducer::GroupState, pool: &[CanonicalItemUrl]) -> ConnectivityStats {
let n = pool.len();
let global_idxs: Vec<Option<usize>> = pool
.iter()
- .map(|it| {
- let key = CanonicalItemUrl(it.clone());
- group.item_to_idx.get(&key).copied()
- })
+ .map(|it| group.item_to_idx.get(it).copied())
.collect();
let present: Vec<usize> = global_idxs.iter().filter_map(|x| *x).collect();
diff --git a/server/src/api/mod.rs b/server/src/api/mod.rs
index cf22cb0129366c3aed031bc86f3197a4321cb806..a10ce662105cff8fad949c6b83f7035ce79bed18 100644
--- a/server/src/api/mod.rs
+++ b/server/src/api/mod.rs
@@ -24,7 +24,7 @@ pub use auth::{
pub use helpers::{
api_error, compute_connectivity_stats, is_pair_voted, now_ms, paginate_rankings,
- parse_parent_specs, pick_random_distinct, resolve_item, sha256_hex, vote_touches_path,
+ parse_parent_specs, pick_random_distinct_canonical, resolve_item, sha256_hex, vote_touches_path,
};
pub use rpc::handle_rpc_batch;
diff --git a/server/src/api/rpc.rs b/server/src/api/rpc.rs
index 5f7d50188f1381267402f2e57e671234ef5db2fd..de0955d0887d32740e1fd18365205c5bdb53c247 100644
--- a/server/src/api/rpc.rs
+++ b/server/src/api/rpc.rs
@@ -29,7 +29,7 @@ use crate::{
use super::auth::verify_bearer_principal;
use super::helpers::{
compute_connectivity_stats, is_pair_voted, now_ms, paginate_rankings, parse_parent_specs,
- pick_random_distinct, resolve_item, vote_touches_path,
+ pick_random_distinct_canonical, resolve_item, vote_touches_path,
};
use super::validate::{normalize_room_and_thread, validate_ingest_document};
@@ -146,21 +146,21 @@ fn authorize_room_read(reduced: &ReducerState, headers: &HeaderMap, room: &str)
}
fn compute_scope_rank_changes(
- parent: &str,
+ parent: &CanonicalItemUrl,
before: &crate::scope_rank::ChildrenRankings,
after: &crate::scope_rank::ChildrenRankings,
room_wire: &str,
) -> Option<ScopeRankChanges> {
- fn build_positions(rankings: &crate::scope_rank::ChildrenRankings) -> HashMap<String, Option<RankPosition>> {
+ fn build_positions(rankings: &crate::scope_rank::ChildrenRankings) -> HashMap<CanonicalItemUrl, Option<RankPosition>> {
let mut map = HashMap::new();
for comp in &rankings.component_rankings {
let total = comp.ranked.len();
for (i, item) in comp.ranked.iter().enumerate() {
- map.insert(item.item.as_str().to_string(), Some(RankPosition { rank: i + 1, of: total }));
+ map.insert(item.item.clone(), Some(RankPosition { rank: i + 1, of: total }));
}
}
for item in &rankings.unranked_items {
- map.insert(item.as_str().to_string(), None);
+ map.insert(item.clone(), None);
}
map
}
@@ -168,7 +168,7 @@ fn compute_scope_rank_changes(
let before_pos = build_positions(before);
let after_pos = build_positions(after);
- let all_items: std::collections::BTreeSet<String> = before_pos.keys().cloned()
+ let all_items: std::collections::BTreeSet<CanonicalItemUrl> = before_pos.keys().cloned()
.chain(after_pos.keys().cloned())
.collect();
@@ -184,7 +184,7 @@ fn compute_scope_rank_changes(
};
if changed {
changes.push(RankChange {
- item: GardenItemUrl::from_storage_str(&item, room_wire),
+ item: GardenItemUrl::from_stored(&item, room_wire),
before: b,
after: a,
});
@@ -203,11 +203,7 @@ fn compute_scope_rank_changes(
});
Some(ScopeRankChanges {
- parent: if parent.is_empty() {
- "/".to_string()
- } else {
- GardenItemUrl::from_storage_str(parent, room_wire).into_inner()
- },
+ parent: GardenItemUrl::from_stored(parent, room_wire).into_inner(),
changes,
})
}
@@ -473,8 +469,8 @@ async fn rpc_post(
for s in &v.doc.statements {
if let dsl::Stmt::Vote { item1, item2, .. } = s {
if let (Ok(a), Ok(b)) = (resolve_item(item1), resolve_item(item2)) {
- if let Some(p) = CanonicalItemUrl::parse(&a).and_then(|c| c.parent()) { parents.insert(p); }
- if let Some(p) = CanonicalItemUrl::parse(&b).and_then(|c| c.parent()) { parents.insert(p); }
+ if let Some(p) = a.parent() { parents.insert(p); }
+ if let Some(p) = b.parent() { parents.insert(p); }
}
}
}
@@ -525,7 +521,7 @@ async fn rpc_post(
.filter_map(|p| {
let before = pre_rankings.get(p)?;
let after = crate::scope_rank::build_children_rankings(content, p);
- compute_scope_rank_changes(p.as_str(), before, &after, &room_key)
+ compute_scope_rank_changes(p, before, &after, &room_key)
})
.collect();
if v.is_empty() { None } else { Some(v) }
@@ -638,8 +634,8 @@ async fn rpc_check(
for s in &v.doc.statements {
if let dsl::Stmt::Vote { item1, item2, .. } = s {
if let (Ok(a), Ok(b)) = (resolve_item(item1), resolve_item(item2)) {
- if let Some(p) = CanonicalItemUrl::parse(&a).and_then(|c| c.parent()) { parents.insert(p); }
- if let Some(p) = CanonicalItemUrl::parse(&b).and_then(|c| c.parent()) { parents.insert(p); }
+ if let Some(p) = a.parent() { parents.insert(p); }
+ if let Some(p) = b.parent() { parents.insert(p); }
}
}
}
@@ -961,7 +957,7 @@ fn rpc_search(reduced: &ReducerState, q: &str, limit: usize, principal: Option<&
async fn rpc_get_pair(state: &AppState, room: String, parent_path: String) -> Result<RpcResult, RpcErr> {
let scope = scope_from_room_wire(&room);
let reduced_arc = state.reduced.clone();
- let pool: Vec<String> = {
+ let pool: Vec<CanonicalItemUrl> = {
let reduced = reduced_arc.read().await;
let content = content_for_room(&reduced, &room);
let tmp = if parent_path.trim().is_empty() {
@@ -970,12 +966,11 @@ async fn rpc_get_pair(state: &AppState, room: String, parent_path: String) -> Re
Some(parent_path.clone())
};
let specs = parse_parent_specs(tmp.as_ref());
- let raw_pool: Vec<CanonicalItemUrl> = if specs.is_empty() {
+ if specs.is_empty() {
content.ranking_group.idx_to_item.clone()
} else {
crate::scope_rank::resolve_scope(content, &specs)
- };
- raw_pool.into_iter().map(|it| it.0).collect()
+ }
};
if pool.len() < 2 {
return Err((
@@ -983,31 +978,30 @@ async fn rpc_get_pair(state: &AppState, room: String, parent_path: String) -> Re
Some("add items via ingest".into()),
));
}
- let selected: Option<(String, String)> = {
+ let selected: Option<(CanonicalItemUrl, CanonicalItemUrl)> = {
let mut reduced = reduced_arc.write().await;
let content = reduced.content.entry(scope.clone()).or_default();
let group = &mut content.ranking_group;
if group.idx_to_item.is_empty() {
- pick_random_distinct(&pool)
+ pick_random_distinct_canonical(&pool)
} else {
let mut rng = rand::thread_rng();
- let idxs: Vec<usize> = pool.iter()
- .filter_map(|it| {
- let key = CanonicalItemUrl(it.clone());
- group.item_to_idx.get(&key).copied()
- })
+ let idxs: Vec<usize> = pool
+ .iter()
+ .filter_map(|it| group.item_to_idx.get(it).copied())
.collect();
let ranked = ranked_items_subset(group, &idxs, 10000, 1e-8);
- let ranked_set: HashSet<String> = ranked.iter().map(|r| r.item.as_str().to_string()).collect();
- let unsorted: Vec<String> = pool.iter()
+ let ranked_set: HashSet<CanonicalItemUrl> = ranked.iter().map(|r| r.item.clone()).collect();
+ let unsorted: Vec<CanonicalItemUrl> = pool
+ .iter()
.filter(|it| !ranked_set.contains(*it))
.cloned()
.collect();
- let mut pick: Option<(String, String)> = None;
+ let mut pick: Option<(CanonicalItemUrl, CanonicalItemUrl)> = None;
if !unsorted.is_empty() {
if let Some(left) = unsorted.choose(&mut rng).cloned() {
- let mut candidates: Vec<String> = if !ranked.is_empty() {
- ranked.iter().map(|r| r.item.as_str().to_string()).collect()
+ let mut candidates: Vec<CanonicalItemUrl> = if !ranked.is_empty() {
+ ranked.iter().map(|r| r.item.clone()).collect()
} else {
pool.clone()
};
@@ -1021,21 +1015,21 @@ async fn rpc_get_pair(state: &AppState, room: String, parent_path: String) -> Re
let a = ranked[i].item.as_str();
let b = ranked[i + 1].item.as_str();
if a != b && !is_pair_voted(group, a, b) {
- pick = Some((a.to_string(), b.to_string()));
+ pick = Some((ranked[i].item.clone(), ranked[i + 1].item.clone()));
break;
}
}
if pick.is_none() {
for _ in 0..64 {
let (Some(a), Some(b)) = (pool.choose(&mut rng).cloned(), pool.choose(&mut rng).cloned()) else { break; };
- if a != b && !is_pair_voted(group, &a, &b) {
+ if a != b && !is_pair_voted(group, a.as_str(), b.as_str()) {
pick = Some((a, b));
break;
}
}
}
}
- pick.or_else(|| pick_random_distinct(&pool))
+ pick.or_else(|| pick_random_distinct_canonical(&pool))
}
};
let Some((left, right)) = selected else {
@@ -1043,8 +1037,8 @@ async fn rpc_get_pair(state: &AppState, room: String, parent_path: String) -> Re
};
let reduced = reduced_arc.read().await;
let content = content_for_ro
… preview truncated; 8,683 characters omittedHardlinks — judgments / attempts / prompt
judgments
attempts
Prompt text is loaded only by the download route.