Side A fixes a real correctness bug (implicit feed cutoff using timestamps that can collide or move backwards) with a principled ingest-order anchor, adds a room field to prevent private-content leakage, and backs this with substantial multi-user permission integration tests. Side B is a pure refactor (splitting forum.rs into submodules) plus an unrelated dev tooling script, providing organizational value but no functional fix or new guarantees, and its diff even reintroduces some inline duplication it had previously factored out.
constitution · epochs · watch · epoch 3
comparison
c_0a9a8eab32ba (tommy-mor) vs c_59974b9da42a (tommy-mor)
download prompt · raw event · cmp_996ff13538ab12
council reasoning
A fixes real feed correctness and privacy: implicit catch-up anchors on durable ingest index (not ms timestamps), keeps visibility checks so multi-user private rooms neither drop concurrent posts nor leak revoked content, and adds room on FeedPost plus targeted integration tests. B mostly splits forum.rs into modules, inlines a few RPC helpers, and adds a macOS sample-fixture script—organizational churn without comparable behavioral value.
Side A fixes core feed correctness by introducing a shared `rpc_feed` implementation that anchors implicit catch-up to durable ingest order instead of timestamps, enforces per-viewer visibility checks, includes room metadata, and adds extensive integration tests covering timestamp collisions, private-room permissions, and revoked access. Side B is primarily a code organization/refactoring change that splits forum code into new files and inlines some helper calls, with only a small auxiliary profiling script added; it does not materially change project behavior.
sides
A — c_0a9a8eab32ba (tommy-mor)
message
[c94456ff] Make feed catch-up stable and permission-aware Anchor implicit feeds to durable ingest order and cover multi-user private-room visibility so concurrent posts are not missed or leaked. Co-authored-by: Cursor <cursoragent@cursor.com>
diff preview
diff --git a/cli/src/main.rs b/cli/src/main.rs
index abb5a55b49f60fe28fbfd4ec02715cb94ea0b4ec..c4f1494df3aedbd8b883aea6249579aa8336abfe 100644
--- a/cli/src/main.rs
+++ b/cli/src/main.rs
@@ -878,6 +878,30 @@ mod tests {
"graph: 4 items, 3/6 pairs (50.0% density), 1 component, connected"
);
}
+
+ #[test]
+ fn feed_without_since_uses_logged_in_delegate_from_env() {
+ let key = "SLUG_DELEGATE";
+ let previous = std::env::var_os(key);
+ let expected = "00000000-0000-0000-0000-0000000000ee:test:local/model";
+ std::env::set_var(key, expected);
+
+ let cli = Cli::try_parse_from(["slugsocial", "feed"]).expect("parse feed");
+
+ match previous {
+ Some(value) => std::env::set_var(key, value),
+ None => std::env::remove_var(key),
+ }
+ match cli.cmd {
+ Some(Command::Feed {
+ delegate, since, ..
+ }) => {
+ assert_eq!(delegate.as_deref(), Some(expected));
+ assert!(since.is_none());
+ }
+ _ => panic!("expected feed command"),
+ }
+ }
}
async fn run_scoped(base: &str, room: &str, sub: ScopedCmd) -> Result<()> {
@@ -1626,7 +1650,15 @@ async fn run() -> Result<()> {
} else {
for p in &resp.posts {
let ago = slug_types::timeago::timeago(now_ms, p.ts);
- println!("<post id=\"{}\" ts=\"{}\">", p.id, ago);
+ let thread_attr = p
+ .thread
+ .as_deref()
+ .map(|thread| format!(" thread=\"{thread}\""))
+ .unwrap_or_default();
+ println!(
+ "<post id=\"{}\" ts=\"{}\" room=\"{}\"{}>",
+ p.id, ago, p.room, thread_attr
+ );
print!("{}", p.body);
if !p.body.ends_with('\n') { println!(); }
println!("</post>");
diff --git a/server/src/api/rpc.rs b/server/src/api/rpc.rs
index afc4f95bef160c1e38ecff2c096d6440cd94e2b3..46d748f918d9b225acc4ecedfe5a1793089407b5 100644
--- a/server/src/api/rpc.rs
+++ b/server/src/api/rpc.rs
@@ -72,6 +72,80 @@ fn can_view_scope(reduced: &ReducerState, scope: &ScopeId, principal: Option<&st
}
}
+/// Build a feed in durable ingest order.
+///
+/// An implicit feed boundary is an ingest position, not only its millisecond timestamp. Two users
+/// can post in the same millisecond, and wall-clock timestamps can move backwards during replay.
+/// Explicit `since` remains a timestamp query for API compatibility, but scans the whole ordered
+/// ledger rather than assuming timestamps are monotonic.
+fn rpc_feed(
+ reduced: &ReducerState,
+ viewer: &str,
+ delegate: Option<String>,
+ requested_since: Option<i64>,
+ implicit_anchor: Option<(usize, i64)>,
+ limit: usize,
+) -> FeedResponse {
+ let since = requested_since.or_else(|| implicit_anchor.map(|(_, ts)| ts));
+ let implicit_anchor_index = requested_since
+ .is_none()
+ .then(|| implicit_anchor.map(|(index, _)| index))
+ .flatten();
+
+ let matching: Vec<&str> = reduced
+ .ingests_ordered
+ .iter()
+ .enumerate()
+ .rev()
+ .filter(|(index, id)| {
+ reduced.ingests_by_id.get(id.as_str()).is_some_and(|ing| {
+ match requested_since {
+ Some(cutoff) => ing.ts > cutoff,
+ None => implicit_anchor_index.is_none_or(|anchor| *index > anchor),
+ }
+ })
+ })
+ .map(|(_, id)| id.as_str())
+ .filter(|id| {
+ reduced.ingests_by_id.get(*id).is_some_and(|ing| {
+ let scope = scope_from_room_wire(&ing.room_id);
+ can_view_scope(reduced, &scope, Some(viewer))
+ })
+ })
+ .filter(|id| !reduced.redacted_posts.contains(*id))
+ .collect();
+
+ let total = matching.len();
+ let posts = matching
+ .into_iter()
+ .take(limit)
+ .filter_map(|id| reduced.ingests_by_id.get(id))
+ .map(|ing| {
+ let scope = scope_from_room_wire(&ing.room_id);
+ let thread_post_index = reduced.try_thread_post_index_chronological(
+ &scope,
+ &ing.thread_tag,
+ &ing.id,
+ );
+ FeedPost {
+ ts: ing.ts,
+ id: ing.id.clone(),
+ room: ing.room_id.clone(),
+ thread: Some(ing.thread_tag.clone()),
+ thread_post_index,
+ body: ing.raw.clone(),
+ }
+ })
+ .collect();
+
+ FeedResponse {
+ delegate,
+ since,
+ posts,
+ total,
+ }
+}
+
fn principal_from_optional_bearer(headers: &HeaderMap, reduced: &ReducerState) -> Result<Option<String>, RpcErr> {
if headers.contains_key(axum::http::header::AUTHORIZATION) {
verify_bearer_principal(headers, reduced)
@@ -1506,60 +1580,27 @@ pub async fn handle_rpc_batch(
Some("this delegate is not bound to your signed-in account".into()),
)
} else {
- let since_default = reduced
+ let implicit_anchor = reduced
.ingests_ordered
.iter()
+ .enumerate()
.rev()
- .filter_map(|id| reduced.ingests_by_id.get(id))
- .find(|ing| {
- if ing.delegate.as_deref() != Some(delegate_stored.as_str()) {
- return false;
- }
- let scope = scope_from_room_wire(&ing.room_id);
- can_view_scope(&reduced, &scope, Some(viewer.as_str()))
- })
- .map(|ing| ing.ts);
- let since = since.or(since_default);
- let cutoff = since.unwrap_or(0);
- let limit = limit.unwrap_or(DEFAULT_LIMIT).min(MAX_LIMIT);
- let matching: Vec<&str> = reduced.ingests_ordered.iter().rev()
- .map(|id| id.as_str())
- .take_while(|id| reduced.ingests_by_id.get(*id).is_some_and(|ing| ing.ts > cutoff))
- .filter(|id| {
- reduced.ingests_by_id.get(*id).is_some_and(|ing| {
- let scope = scope_from_room_wire(&ing.room_id);
- can_view_scope(&reduced, &scope, Some(viewer.as_str()))
+ .find_map(|(index, id)| {
+ reduced.ingests_by_id.get(id).and_then(|ing| {
+ (ing.delegate.as_deref()
+ == Some(delegate_stored.as_str()))
+ .then_some((index, ing.ts))
})
- })
- .filter(|id| !reduced.redacted_posts.contains(*id))
- .collect();
- let total = matching.len();
- let posts: Vec<FeedPost> = matching.into_iter()
- .take(limit)
- .filter_map(|id| reduced.ingests_by_id.get(id))
- .map(|ing| {
- let scope = scope_from_room_wire(&ing.room_id);
- let thread_post_index = reduced
- .try_thread_post_index_chronological(
- &scope,
- &ing.thread_tag,
- &ing.id,
- );
- FeedPost {
- ts: ing.ts,
- id: ing.id.clone(),
- thread: Some(ing.thread_tag.clone()),
- thread_post_index,
- body: ing.raw.clone(),
- }
- })
- .collect();
- line_ok(RpcResult::Feed(FeedResponse {
- delegate: Some(delegate_stored),
+ });
+ let limit = limit.unwrap_or(DEFAULT_LIMIT).min(MAX_LIMIT);
+ line_ok(RpcResult::Feed(rpc_feed(
+ &reduced,
+ &viewer,
+ Some(delegate_stored),
since,
- posts,
- total,
- }))
+ implicit_anchor,
+ limit,
+ )))
};
drop(reduced);
line
@@ -1567,60 +1608,25 @@ pub async fn handle_rpc_batch(
None => {
// Session catch-up: last time *you* posted anything (delegate or not), so revisiting
// an old chat with only a token still gets a sane cutoff.
- let since_default = reduced
+ let implicit_anchor = reduced
.ingests_ordered
.iter()
+ .enumerate()
.rev()
- .filter_map(|id| reduced.ingests_by_id.get(id))
- .find(|ing| {
- if ing.principal != viewer {
- return false;
- }
- let scope = scope_from_room_wire(&ing.room_id);
- can_view_scope(&reduced, &scope, Some(viewer.as_str()))
- })
- .map(|ing| ing.ts);
- let since = since.or(since_default);
- let cutoff = since.unwrap_or(0);
- let limit = limit.unwrap_or(DEFAULT_LIMIT).min(MAX_LIMIT);
- let matching: Vec<&str> = reduced.ingests_ordered.iter().rev()
- .map(|id| id.as_str())
- .take_while(|id| reduced.inges
… preview truncated; 14,758 characters omittedB — c_59974b9da42a (tommy-mor)
message
[c99adc26] refactor forum into files
diff preview
diff --git a/bb.edn b/bb.edn
index 4680e82d4227057d2eea6041dc5647beb759131d..ce0f6fcd45a67ead3295b90db2a301e9b304f3da 100644
--- a/bb.edn
+++ b/bb.edn
@@ -51,6 +51,11 @@
:requires ([test.walkthrough-fixture :as walkthrough-fixture])
:task (walkthrough-fixture/run-fixture)}
+ sample-fixture
+ {:doc "macOS: run `sample` on process(es) listening on the fixture TCP port (default 8080). Usage: bb sample-fixture [PORT] [DURATION_SEC] [OUT_DIR]"
+ :requires ([scripts.sample-fixture :as sample-fixture])
+ :task (apply sample-fixture/-main *command-line-args*)}
+
perf
{:doc "Performance test: concurrent HTTP requests to detect blocking I/O"
:requires ([scripts.perf :as perf])
diff --git a/scripts/sample_fixture.bb b/scripts/sample_fixture.bb
new file mode 100644
index 0000000000000000000000000000000000000000..ac2368623c5043ba6e027afef6b5b37d392767b0
--- /dev/null
+++ b/scripts/sample_fixture.bb
@@ -0,0 +1,70 @@
+(ns scripts.sample-fixture
+ "Find process(es) listening on the fixture port (default 8080) and run macOS `sample`."
+ (:require [babashka.fs :as fs]
+ [babashka.process :as p]
+ [clojure.string :as str]))
+
+(defn- usage []
+ (println "Usage: bb sample-fixture [PORT] [DURATION_SEC] [OUT_DIR]")
+ (println "")
+ (println " Finds PIDs bound to TCP LISTEN on PORT (default 8080), then runs")
+ (println " `sample` for each PID. OUT_DIR defaults to the current directory.")
+ (println "")
+ (println " Example: bb sample-fixture")
+ (println " bb sample-fixture 8080 10")
+ (println " bb sample-fixture 8080 5 /tmp")
+ (println "")
+ (println " Requires macOS (the `sample` tool)."))
+
+(defn- parse-long* [s]
+ (try (Long/parseLong s)
+ (catch NumberFormatException _ nil)))
+
+(defn- listen-pids [port]
+ (let [spec (str "TCP:" port)
+ {:keys [out exit]}
+ @(p/process ["lsof" "-nP" (str "-i" spec) "-sTCP:LISTEN" "-t"]
+ {:out :string :err :string})]
+ (when (zero? exit)
+ (->> (str/split-lines out)
+ (map str/trim)
+ (remove str/blank?)
+ (distinct)
+ vec))))
+
+(defn- sample-bin []
+ (or (fs/which "sample")
+ (throw (ex-info "macOS `sample` not found on PATH" {}))))
+
+(defn- run-sample! [sample duration-sec pid out-file]
+ (println (str "sampling PID " pid " for " duration-sec "s → " out-file))
+ (let [{:keys [exit err]} @(p/process [sample (str pid) (str duration-sec) "-file" out-file]
+ {:out :inherit :err :inherit})]
+ (when-not (zero? exit)
+ (binding [*out* *err*]
+ (println "sample failed:" err))
+ (System/exit exit))))
+
+(defn -main [& args]
+ (when (some #{"-h" "--help" "help"} args)
+ (usage)
+ (System/exit 0))
+ (let [port (or (some-> (first args) parse-long*) 8080)
+ duration-sec (or (some-> (second args) parse-long*) 5)
+ out-dir (or (nth args 2 nil) ".")
+ pids (listen-pids port)]
+ (when (or (nil? pids) (empty? pids))
+ (binding [*out* *err*]
+ (println (str "No process listening on TCP " port " (LISTEN). Is `bb fixture` running?")))
+ (System/exit 1))
+ (when-not (fs/exists? out-dir)
+ (binding [*out* *err*]
+ (println "Output directory does not exist:" out-dir))
+ (System/exit 1))
+ (let [sample (sample-bin)
+ ts (str (System/currentTimeMillis))]
+ (println (str "port " port " → PIDs " (str/join ", " pids)))
+ (doseq [pid pids]
+ (let [out-file (str (fs/path out-dir) "/slug-sample-" port "-" pid "-" ts ".txt")]
+ (run-sample! sample duration-sec pid (str out-file))))
+ (println "done."))))
diff --git a/server/src/html/forum.rs b/server/src/html/forum.rs
index a55fc1f79d03719eb74b216865b2c15199c48ea7..5ad8dfc84dd735d589432e2c613ff687a75f2e61 100644
--- a/server/src/html/forum.rs
+++ b/server/src/html/forum.rs
@@ -380,22 +380,6 @@ fn room_members_inner(members: &[RoomMemberRow]) -> Markup {
}
}
-pub(crate) fn set_room_members_expanded_rpc(room_wire: &str, expanded: bool) -> String {
- template_json_compact(&HtmlUiAction::SetRoomMembersExpanded {
- room_wire: room_wire.to_string(),
- expanded,
- })
- .expect("static json")
-}
-
-pub(crate) fn set_room_new_thread_compose_expanded_rpc(nav: &ThreadNav, expanded: bool) -> String {
- template_json_compact(&HtmlUiAction::SetRoomNewThreadComposeExpanded {
- room_wire: nav.room_wire.clone(),
- expanded,
- })
- .expect("static json")
-}
-
/// Fragment for `#room-members-section` — expand/collapse is server-driven via `POST /ui`.
pub(crate) fn room_members_section_markup(
reduced: &ReducerState,
@@ -406,13 +390,14 @@ pub(crate) fn room_members_section_markup(
if members.is_empty() {
return html! {};
}
- let rpc_open = set_room_members_expanded_rpc(room_id, true);
- let rpc_close = set_room_members_expanded_rpc(room_id, false);
html! {
div id="room-members-section" {
@if members_expanded {
form method="POST" action="/ui" {
- input type="hidden" name=(UI_RPC_FIELD) value=(rpc_close);
+ input type="hidden" name=(UI_RPC_FIELD) value=(template_json_compact(&HtmlUiAction::SetRoomMembersExpanded {
+ room_wire: room_id.to_string(),
+ expanded: false,
+ }).expect("static json"));
button type="submit" class="form-toggle" aria-expanded="true" {
"hide members & permissions"
}
@@ -422,7 +407,10 @@ pub(crate) fn room_members_section_markup(
}
} @else {
form method="POST" action="/ui" {
- input type="hidden" name=(UI_RPC_FIELD) value=(rpc_open);
+ input type="hidden" name=(UI_RPC_FIELD) value=(template_json_compact(&HtmlUiAction::SetRoomMembersExpanded {
+ room_wire: room_id.to_string(),
+ expanded: true,
+ }).expect("static json"));
button type="submit" class="form-toggle" aria-expanded="false" {
"members & permissions"
}
@@ -514,28 +502,24 @@ fn compose_form(nav: &ThreadNav, thread_tag: &str, show: bool) -> Markup {
if !show {
return html! {};
}
- let rpc_post = template_json_compact(&json!({
- "action": "post_ingest",
- "room": nav.room_wire,
- "thread_tag": thread_tag,
- "text": {"$form": "text"},
- "error_target": "thread-compose-errors",
- "form_id": "thread-compose-form",
- }))
- .unwrap();
- let rpc_check = template_json_compact(&json!({
- "action": "check_ingest",
- "room": nav.room_wire,
- "thread_tag": thread_tag,
- "text": {"$form": "text"},
- "error_target": "thread-compose-errors",
- "form_id": "thread-compose-form",
- }))
- .unwrap();
html! {
section class="compose" id="thread-compose" {
- form id="thread-compose-form" method="POST" action="/ui" data-check-action="/ui" data-check-rpc=(rpc_check) {
- input type="hidden" name=(UI_RPC_FIELD) value=(rpc_post);
+ form id="thread-compose-form" method="POST" action="/ui" data-check-action="/ui" data-check-rpc=(template_json_compact(&json!({
+ "action": "check_ingest",
+ "room": nav.room_wire,
+ "thread_tag": thread_tag,
+ "text": {"$form": "text"},
+ "error_target": "thread-compose-errors",
+ "form_id": "thread-compose-form",
+ })).unwrap()) {
+ input type="hidden" name=(UI_RPC_FIELD) value=(template_json_compact(&json!({
+ "action": "post_ingest",
+ "room": nav.room_wire,
+ "thread_tag": thread_tag,
+ "text": {"$form": "text"},
+ "error_target": "thread-compose-errors",
+ "form_id": "thread-compose-form",
+ })).unwrap());
textarea name="text" rows="5" cols="80" placeholder="prose or ~/items and votes…" {}
p {
button type="submit" { "post" }
@@ -712,7 +696,7 @@ pub async fn home(
p class="muted" { "dark = time-ordered · light = vote-ranked" }
div class="thread-feed-toolbar" {
form method="POST" action="/ui" {
- input type="hidden" name=(UI_RPC_FIELD) value=(expand_public_new_thread_rpc_value());
+ input type="hidden" name=(UI_RPC_FIELD) value=(template_json_compact(&HtmlUiAction::ExpandPublicNewThreadForm).expect("static json"));
button type="submit" class="section-add-btn" { "+" }
}
}
@@ -1002,14 +986,15 @@ fn new_thread_form_for_room(nav: &ThreadNav, show: bool, compose_expanded: bool)
if !show {
return html! {};
}
- let rpc_open = set_room_new_thread_compose_expanded_rpc(nav, true);
- let rpc_close = set_room_new_thread_compose_expanded_rpc(nav, false);
// Single root for Idiomorph when morphing `#room-new-thread-ui-slot` (expanded has form + section).
html! {
div class="room-new-thread-slot-inner" {
@if compose_expanded {
form method="POST" action="/ui" {
- input type="hidden" name=(UI_RPC_FIELD) value=(rpc_close);
+ input type="hidden" name=(UI_RPC_FIELD) value=(template_json_compact(&HtmlUiAction::SetRoomNewThreadComposeExpanded {
+ room_wire: nav.room_wire.clone(),
+ expanded: false,
+ }).expect("static json"));
button type="submit" class="form-toggle" aria-expanded="true" {
"-"
}
@@ -1039,7 +1024,10 @@ fn new_thread_form_for_room(nav: &ThreadNav, show: bool, compose_expanded: bool)
}
} @else {
form method="POST" action="/ui" {
- input type="hidden" name=(UI_RPC_FIELD) value=(rpc_open);
+ input type="hidden" name=(UI_RPC_FIELD) value=(template_json_compact(&HtmlUiAction::SetRoomNewThreadComposeExpanded {
+ room_wire: nav.room_wire.clone(),
+ expanded: true,
+ }).expect("static json"));
button type="submit" class="form-toggle" aria-expanded="false" {
"+"
}
@@ -1049,10 +1037,6 @@ fn new_thread_form_for_room(nav: &ThreadNav, show: bool, compose_expanded: bool)
}
}
-pub(crate) fn expand_public_new_thread_rpc_value() -> String {
- template_json_compact(&HtmlUiAction::ExpandPublicNewThreadForm).expect("static json")
-}
-
pub(crate) fn login_to_post_hint_markup() -> Markup {
html! {
p class="muted" { "log in to post" }
diff --git a/server/src/html/forum/access.rs b/server/src/html/forum/access.rs
new file mode 100644
index 0000000000000000000000000000000000000000..f9f7139b63442a6ef3a927b77a72805f15feec3f
--- /dev/null
+++ b/server/src/html/forum/access.rs
@@ -0,0 +1,16 @@
+use crate::events::ThreadCapability;
+use crate::reducer::ReducerState;
+
+pub(crate) fn user_can_view_room(reduced: &ReducerState, room_id: &str, username: Option<&str>) -> bool {
+ if !reduced.rooms.contains(room_id) {
+ return false;
+ }
+ let Some(u) = username else {
+ return false;
+ };
+ reduced.user_has_cap(room_id, u, ThreadCapability::View)
+}
+
+pub(crate) fn user_can_post_room(reduced: &ReducerState, room_id: &str, username: &str) -> bool {
+ reduced.user_has_cap(room_id, username, ThreadCapability::Post)
+}
diff --git a/server/src/html/forum/ingest
… preview truncated; 19,729 characters omittedHardlinks — judgments / attempts / prompt
judgments
attempts
Prompt text is loaded only by the download route.