From 2b695d6892dceeb601d615ef08f7922607d8ba88 Mon Sep 17 00:00:00 2001 From: tomdebone Date: Thu, 10 Sep 2026 07:08:46 +0200 Subject: [PATCH] =?UTF-8?q?fix(appview):=20Unfollows=20=C3=BCber=20den=20F?= =?UTF-8?q?irehose=20anwendbar=20machen?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ein Delete-Event trägt nur did + rkey, keinen Record-Body. `follows` hatte aber nur (follower_did, subject_did) und speicherte den rkey nicht — es gab also keinen Weg vom rkey zum subject_did, und der Indexer hat solche Ops geloggt und übersprungen. Unfollows hingen damit allein am Best-Effort-Push, genau der Abhängigkeit, die der Firehose beseitigen soll. Migration 0011 ergänzt die rkey-Spalte plus einen partiellen Index für den Lookup. Der Primärschlüssel bleibt (follower_did, subject_did), damit die Upserts über Push, Firehose und Replay hinweg idempotent bleiben; ein rkey im Schlüssel würde aus einem Re-Follow eine zweite Zeile machen und die Follower-Zahl verdoppeln. Der Index ist bewusst nicht unique: sonst würde ausgerechnet der Fall, für den das hier existiert — verlorener Delete, dann ein neuer Create — zu einem abgebrochenen Write. delete_follow_by_rkey löst und löscht in einem Statement (RETURNING), also ohne Rennen zwischen Auflösen und Löschen. Findet es nichts — alte Zeile ohne rkey, schon gelöscht, veralteter rkey — ist das kein Fehler. Der Push-Pfad über subject_did bleibt unverändert. Likes und Reposts haben die Lücke nicht: dort ist der rkey Teil der Zeilenidentität. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX --- crates/appview/src/indexer.rs | 441 +++++++++++++++++++++-- crates/appview/src/ingest.rs | 5 + migrations/appview/0011_follows_rkey.sql | 79 ++++ 3 files changed, 498 insertions(+), 27 deletions(-) create mode 100644 migrations/appview/0011_follows_rkey.sql diff --git a/crates/appview/src/indexer.rs b/crates/appview/src/indexer.rs index d5b75a0..832842a 100644 --- a/crates/appview/src/indexer.rs +++ b/crates/appview/src/indexer.rs @@ -693,24 +693,46 @@ where // -- follows --------------------------------------------------------------- +/// Insert or update the follow edge `follower_did -> subject_did`. +/// +/// `rkey` is the record key of the `app.bsky.graph.follow` record this +/// edge came from. It is stored as a *second access path* to the row — +/// the primary key stays `(follower_did, subject_did)`, which is what +/// keeps this upsert idempotent across the push path, the firehose and +/// any replay of either. See migration 0011 for the full reasoning. +/// +/// Pass `None` only when the caller genuinely has no rkey. On conflict +/// the column is `COALESCE(EXCLUDED.rkey, follows.rkey)`: a newer record +/// overwrites it (youngest record wins, so a re-follow's rkey replaces +/// the old one and a stale delete for the old rkey can no longer match), +/// but a caller that omits the rkey must not blank out one another +/// transport already recorded — that would re-open the very gap this +/// column closes. pub async fn upsert_follow( db: &PgPool, follower_did: &str, subject_did: &str, + rkey: Option<&str>, record: Option<&Value>, ) -> Result<()> { let created_at = parse_created_at( record.and_then(|r| r.get("createdAt")).and_then(|v| v.as_str()), ); + // An empty rkey is not an rkey — treat it like the absent case so a + // caller forwarding a blank field can't write a row that a + // `WHERE rkey = ''` delete would later match by accident. + let rkey = rkey.filter(|r| !r.is_empty()); sqlx::query( - r#"INSERT INTO follows (follower_did, subject_did, created_at) - VALUES ($1, $2, $3) + r#"INSERT INTO follows (follower_did, subject_did, rkey, created_at) + VALUES ($1, $2, $3, $4) ON CONFLICT (follower_did, subject_did) DO UPDATE SET + rkey = COALESCE(EXCLUDED.rkey, follows.rkey), created_at = EXCLUDED.created_at, indexed_at = now()"#, ) .bind(follower_did) .bind(subject_did) + .bind(rkey) .bind(created_at) .execute(db) .await?; @@ -733,6 +755,12 @@ pub async fn upsert_follow( Ok(()) } +/// Delete the follow edge by its relationship identity. +/// +/// This is the PDS-push path: `/internal/ingest-commit` carries the +/// `subject_did` from the PDS's own snapshot, so the row can be +/// addressed directly. Idempotent — deleting an edge that is already +/// gone is a no-op, not an error. pub async fn delete_follow( db: &PgPool, follower_did: &str, @@ -748,6 +776,78 @@ pub async fn delete_follow( Ok(()) } +/// Delete the follow edge that came from record `rkey` in +/// `follower_did`'s repo. +/// +/// This is the firehose / Jetstream path. A delete op carries only +/// `did` + `rkey` and no record body, so the subject DID has to be +/// recovered from the row itself — which is exactly what the `rkey` +/// column added in migration 0011 is for. The `DELETE ... RETURNING` +/// resolves and removes in one statement (the same shape +/// [`delete_like`] uses to recover its `post_uri`), so there is no +/// window in which another writer could move the row between the +/// lookup and the delete. +/// +/// Returns the `subject_did` that was unfollowed, or `None` when +/// nothing matched. `None` is a normal outcome, never an error: +/// +/// * the row predates migration 0011 and has no rkey (the push path +/// with its `subject_did` still handles those), or +/// * the delete already landed over the other transport, or +/// * the follow was re-created under a newer rkey, in which case this +/// delete is a stale replay and the live edge must be left alone. +/// +/// The caller logs and moves on — an unfollow we cannot place must not +/// stall the frames queued behind it. +pub async fn delete_follow_by_rkey( + db: &PgPool, + follower_did: &str, + rkey: &str, +) -> Result> { + if rkey.is_empty() { + // Guard the degenerate case explicitly: `rkey = ''` can never + // identify a record, and letting it through would mean an empty + // value written by some future caller could be matched here. + tracing::warn!( + follower_did, + "follow delete with an empty rkey; nothing to do" + ); + return Ok(None); + } + // Deleted with `fetch_all` rather than `fetch_optional` because the + // index on `(follower_did, rkey)` is deliberately not unique (see + // migration 0011): in the pathological case of a duplicated rkey, + // every matching row is a follow whose record is gone, so all of + // them should go. + let rows: Vec<(String,)> = sqlx::query_as( + "DELETE FROM follows WHERE follower_did = $1 AND rkey = $2 \ + RETURNING subject_did", + ) + .bind(follower_did) + .bind(rkey) + .fetch_all(db) + .await?; + + match rows.into_iter().next() { + Some((subject_did,)) => { + tracing::debug!( + follower_did, rkey, subject_did, + "applied an unfollow by rkey" + ); + Ok(Some(subject_did)) + } + None => { + tracing::debug!( + follower_did, rkey, + "follow delete by rkey matched no row (already gone, \ + re-created under a newer rkey, or indexed before the \ + rkey column existed); skipping" + ); + Ok(None) + } + } +} + /// Extract the subject DID from a follow record (`{ "subject": "did:..."}`). pub fn follow_subject_did(record: Option<&Value>) -> Option { record? @@ -906,8 +1006,21 @@ pub async fn apply_commit( applied = true; } "app.bsky.graph.follow" => { - let subject_did = match op.action.as_str() { - "create" => match follow_subject_did(op.record.as_ref()) { + // Both actions need the rkey. On a create it is stored + // alongside the edge; on a delete it is the *only* + // thing identifying the edge, because a delete op + // carries no record body and therefore no subject DID. + let rkey = op + .rkey + .clone() + .or_else(|| { + op.path + .as_deref() + .and_then(|p| p.rsplit('/').next().map(str::to_string)) + }) + .filter(|r| !r.is_empty()); + if op.action == "create" { + let subject_did = match follow_subject_did(op.record.as_ref()) { Some(s) => s, None => { tracing::warn!( @@ -915,33 +1028,35 @@ pub async fn apply_commit( ); continue; } - }, - "delete" => { - // Jetstream delete on follows carries no record - // value, so we can't know which subject was - // unfollowed. The PDS-driven internal ingest path - // handles this — it knows the subject from its - // own snapshot. - tracing::warn!( - "follow delete via Jetstream lacks subject; \ - route through /internal/ingest-commit instead" - ); - continue; - } - _ => continue, - }; - if op.action == "create" { + }; upsert_follow( db, &ev.did, &subject_did, + rkey.as_deref(), op.record.as_ref(), ) .await?; + applied = true; } else if op.action == "delete" { - delete_follow(db, &ev.did, &subject_did).await?; + // The rkey → subject_did lookup added in migration + // 0011. Before it, this arm could only log and skip, + // which left unfollows depending entirely on the + // PDS's best-effort push: one lost request and the + // follow stayed indexed forever. + let Some(rkey) = rkey else { + tracing::warn!( + did = %ev.did, + "follow delete op has no rkey; skipping" + ); + continue; + }; + delete_follow_by_rkey(db, &ev.did, &rkey).await?; + // Applied even when no row matched: the event was + // understood and acted on, which is what this flag + // reports (same as the post / like / repost deletes). + applied = true; } - applied = true; } "app.bsky.actor.profile" => { // Jetstream carries profile records as plain @@ -1356,8 +1471,10 @@ mod tests { .unwrap(); assert_eq!(count, 1); - // Delete via the internal API (not via Jetstream — Jetstream - // delete on follows doesn't carry the subject). + // Delete through the PDS-push path, which addresses the edge by + // `(follower, subject)` because the PDS knows the subject from + // its own snapshot. (The firehose path deletes by rkey instead + // — see `firehose_unfollow_deletes_by_rkey` below.) delete_follow(&db, "did:plc:test", "did:plc:b").await.unwrap(); let (count,): (i64,) = sqlx::query_as( "SELECT COUNT(*) FROM follows WHERE follower_did = $1 AND subject_did = $2", @@ -1369,6 +1486,274 @@ mod tests { .unwrap(); assert_eq!(count, 0); } + + // -- unfollow over the firehose --------------------------------------- + // + // The tests below cover the gap migration 0011 closes: a delete op + // carries only `did` + `rkey`, so the edge has to be recoverable + // from the rkey alone. They use per-run unique DIDs because the + // suite shares one database with every other test module. + + fn follow_did(tag: &str) -> String { + format!("did:plc:follow_{}_{}", tag, uuid::Uuid::new_v4().simple()) + } + + /// A commit event shaped like the ones `pds_firehose::events_from_frame` + /// hands to `apply_commit`: single-op, `record` present on create and + /// absent on delete. + fn follow_event( + did: &str, + rkey: &str, + action: &str, + subject: Option<&str>, + ) -> JetstreamEvent { + let mut commit = json!({ + "operation": action, + "collection": "app.bsky.graph.follow", + "rkey": rkey, + "path": format!("app.bsky.graph.follow/{rkey}"), + }); + if let Some(subject) = subject { + commit["cid"] = json!("bafyfollow"); + commit["record"] = json!({ + "subject": subject, + "createdAt": "2026-01-01T00:00:00Z", + }); + } + JetstreamEvent { + did: did.to_string(), + time_us: 1_700_000_000_000_000, + kind: "commit".into(), + commit: Some(commit), + identity: None, + account: None, + } + } + + async fn follow_rows(db: &PgPool, follower: &str) -> Vec<(String, Option)> { + sqlx::query_as( + "SELECT subject_did, rkey FROM follows WHERE follower_did = $1 \ + ORDER BY subject_did", + ) + .bind(follower) + .fetch_all(db) + .await + .unwrap() + } + + /// The core case: a follow that arrived over the firehose is removed + /// by a delete op that names nothing but the rkey. + #[tokio::test] + async fn firehose_unfollow_deletes_by_rkey() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let follower = follow_did("er"); + let subject = follow_did("ee"); + + apply_commit(&db, &follow_event(&follower, "frk1", "create", Some(&subject))) + .await + .unwrap(); + assert_eq!( + follow_rows(&db, &follower).await, + vec![(subject.clone(), Some("frk1".to_string()))], + "the create must store the rkey next to the edge" + ); + + // The delete op carries no record and no subject — only the rkey. + let applied = apply_commit(&db, &follow_event(&follower, "frk1", "delete", None)) + .await + .unwrap(); + assert!(applied, "a follow delete is now actionable, not skipped"); + assert!( + follow_rows(&db, &follower).await.is_empty(), + "the unfollow must remove the edge" + ); + + // Replaying the same delete (reconnect, or the push path racing + // the firehose) must stay a silent no-op. + apply_commit(&db, &follow_event(&follower, "frk1", "delete", None)) + .await + .unwrap(); + assert!(follow_rows(&db, &follower).await.is_empty()); + + let _ = sqlx::query("DELETE FROM notifications WHERE recipient_did = $1") + .bind(&subject) + .execute(&db) + .await; + } + + /// A delete for an rkey we never indexed resolves to nothing. That + /// is a normal outcome (the follow was never seen, or is already + /// gone), so it must not error and must not touch other rows. + #[tokio::test] + async fn delete_follow_by_unknown_rkey_is_a_noop() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let follower = follow_did("er"); + let subject = follow_did("ee"); + + upsert_follow(&db, &follower, &subject, Some("frk1"), None) + .await + .unwrap(); + + let removed = delete_follow_by_rkey(&db, &follower, "no-such-rkey") + .await + .unwrap(); + assert!(removed.is_none(), "an unknown rkey resolves to no subject"); + assert_eq!( + follow_rows(&db, &follower).await, + vec![(subject.clone(), Some("frk1".to_string()))], + "an unmatched delete must leave every other edge alone" + ); + + // An empty rkey is guarded separately — it can never identify a + // record, and must not be allowed to match a blank column. + assert!(delete_follow_by_rkey(&db, &follower, "") + .await + .unwrap() + .is_none()); + assert_eq!(follow_rows(&db, &follower).await.len(), 1); + + delete_follow(&db, &follower, &subject).await.unwrap(); + let _ = sqlx::query("DELETE FROM notifications WHERE recipient_did = $1") + .bind(&subject) + .execute(&db) + .await; + } + + /// Rows written before migration 0011 have `rkey IS NULL`: there was + /// nothing to backfill them from. A delete-by-rkey must not find + /// them (and certainly must not match NULL against anything), while + /// the push path that names the subject keeps working. + #[tokio::test] + async fn legacy_row_without_rkey_still_deletes_via_subject() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let follower = follow_did("legacy"); + let subject = follow_did("ee"); + + // Insert the way migration 0001 through 0010 did — no rkey. + sqlx::query( + "INSERT INTO follows (follower_did, subject_did, created_at) \ + VALUES ($1, $2, now())", + ) + .bind(&follower) + .bind(&subject) + .execute(&db) + .await + .unwrap(); + + let removed = delete_follow_by_rkey(&db, &follower, "frk1").await.unwrap(); + assert!( + removed.is_none(), + "a row with no rkey is unreachable by rkey — by design" + ); + assert_eq!(follow_rows(&db, &follower).await.len(), 1); + + // The PDS push, which carries the subject, still removes it. + delete_follow(&db, &follower, &subject).await.unwrap(); + assert!(follow_rows(&db, &follower).await.is_empty()); + } + + /// Follow → unfollow → follow again produces a fresh rkey. The edge + /// must stay a single row (the primary key is the relationship, not + /// the record), the newest rkey must win, and a stale delete for the + /// old rkey must not tear down the live follow. + #[tokio::test] + async fn refollow_keeps_one_row_and_the_newest_rkey_wins() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let follower = follow_did("er"); + let subject = follow_did("ee"); + + apply_commit(&db, &follow_event(&follower, "frk1", "create", Some(&subject))) + .await + .unwrap(); + apply_commit(&db, &follow_event(&follower, "frk2", "create", Some(&subject))) + .await + .unwrap(); + assert_eq!( + follow_rows(&db, &follower).await, + vec![(subject.clone(), Some("frk2".to_string()))], + "one edge, carrying the youngest record's rkey" + ); + + // The old rkey is stale: its delete must find nothing. + assert!(delete_follow_by_rkey(&db, &follower, "frk1") + .await + .unwrap() + .is_none()); + assert_eq!( + follow_rows(&db, &follower).await.len(), + 1, + "a replayed delete for a superseded record must not unfollow" + ); + + // The current rkey does delete it. + assert_eq!( + delete_follow_by_rkey(&db, &follower, "frk2").await.unwrap(), + Some(subject.clone()), + "the delete resolves the subject it removed" + ); + assert!(follow_rows(&db, &follower).await.is_empty()); + + let _ = sqlx::query("DELETE FROM notifications WHERE recipient_did = $1") + .bind(&subject) + .execute(&db) + .await; + } + + /// A follow that first arrives over the PDS push (no rkey stored by + /// an older AppView, or a caller that has none) and is then seen + /// again over the firehose must end up with the rkey — otherwise + /// the firehose could never delete it. And a later push that omits + /// the rkey must not blank it out again. + #[tokio::test] + async fn rkey_is_filled_in_but_never_blanked_out() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let follower = follow_did("er"); + let subject = follow_did("ee"); + + upsert_follow(&db, &follower, &subject, None, None).await.unwrap(); + assert_eq!( + follow_rows(&db, &follower).await, + vec![(subject.clone(), None)] + ); + + // The firehose replay of the same follow supplies the rkey. + upsert_follow(&db, &follower, &subject, Some("frk1"), None) + .await + .unwrap(); + assert_eq!( + follow_rows(&db, &follower).await, + vec![(subject.clone(), Some("frk1".to_string()))] + ); + + // A subsequent write without one must leave it in place. + upsert_follow(&db, &follower, &subject, None, None).await.unwrap(); + assert_eq!( + follow_rows(&db, &follower).await, + vec![(subject.clone(), Some("frk1".to_string()))], + "COALESCE keeps the rkey the other transport already gave us" + ); + + delete_follow(&db, &follower, &subject).await.unwrap(); + let _ = sqlx::query("DELETE FROM notifications WHERE recipient_did = $1") + .bind(&subject) + .execute(&db) + .await; + } } #[cfg(test)] @@ -1555,7 +1940,9 @@ mod notification_tests { .await .unwrap(); // Self-follow is legal in the protocol; it must stay silent too. - upsert_follow(&db, &author, &author, None).await.unwrap(); + upsert_follow(&db, &author, &author, Some("frk1"), None) + .await + .unwrap(); let total: i64 = sqlx::query_scalar( "SELECT COUNT(*)::BIGINT FROM notifications WHERE recipient_did = $1", @@ -1657,10 +2044,10 @@ mod notification_tests { seed_post(&db, &subject, "p1").await; let record = json!({ "subject": subject, "createdAt": "2026-01-01T00:00:00Z" }); - upsert_follow(&db, &follower, &subject, Some(&record)) + upsert_follow(&db, &follower, &subject, Some("frk1"), Some(&record)) .await .unwrap(); - upsert_follow(&db, &follower, &subject, Some(&record)) + upsert_follow(&db, &follower, &subject, Some("frk1"), Some(&record)) .await .unwrap(); assert_eq!( diff --git a/crates/appview/src/ingest.rs b/crates/appview/src/ingest.rs index 465f394..f464579 100644 --- a/crates/appview/src/ingest.rs +++ b/crates/appview/src/ingest.rs @@ -220,10 +220,15 @@ async fn apply( .map(str::to_string) }) .ok_or_else(|| bad_request("follow create requires subject_did or record.subject"))?; + // Forward the rkey too. The push path doesn't need it to + // apply *this* write — it has the subject — but storing it + // is what lets a later firehose delete (which carries only + // did + rkey) find this row. See migration 0011. indexer::upsert_follow( &state.db, &req.did, &subject, + Some(req.rkey.as_str()), req.record.as_ref(), ) .await diff --git a/migrations/appview/0011_follows_rkey.sql b/migrations/appview/0011_follows_rkey.sql new file mode 100644 index 0000000..fdb8e55 --- /dev/null +++ b/migrations/appview/0011_follows_rkey.sql @@ -0,0 +1,79 @@ +-- AppView database schema 0011: remember which record a follow came from. +-- +-- Why +-- +-- A `follows` row was addressable only as `(follower_did, subject_did)`. +-- That is the right identity for the *relationship*, but it is not the +-- identity a delete event carries. A firehose / Jetstream delete op is +-- just `did` + `rkey`: +-- +-- {"action": "delete", "path": "app.bsky.graph.follow/3lmnop"} +-- +-- There is no record body on a delete — the record is gone, that is the +-- whole point of the event — so the subject DID is nowhere in it. With +-- no rkey stored, the indexer had no way from `3lmnop` back to +-- "did:plc:bob" and logged-and-skipped the op +-- (`indexer::apply_commit`, `app.bsky.graph.follow` arm). +-- +-- The practical consequence: unfollows only ever landed through the +-- PDS's best-effort `POST /internal/ingest-commit` push, which knows the +-- subject from its own snapshot. That push has no retry and no +-- acknowledgement (see `pds_firehose`'s module docs). If it was lost — +-- AppView restarting, request timing out — the follow stayed in the +-- index forever, and the firehose, the stream that exists precisely to +-- repair such gaps, could not repair this one. Storing the rkey closes +-- that hole: the firehose replay can now apply the unfollow on its own. +-- +-- The primary key deliberately stays `(follower_did, subject_did)` +-- ------------------------------------------------------------------ +-- It is what makes `upsert_follow` idempotent. The same follow reaches +-- us over both transports (push *and* firehose) and again after any +-- replay, and every one of those must converge on one row. Keying on +-- the rkey instead — or adding it to the key — would make a re-follow +-- under a fresh rkey a *second* row for the same relationship, and then +-- `follower_count` would count the same follower twice. +-- +-- So `rkey` is not identity here; it is a second *access path* to a row +-- the primary key already identifies. +-- +-- Nullable, because history has no rkey +-- ------------------------------------- +-- Every row written before this migration was inserted without one, and +-- there is nothing to backfill it from: the AppView never stored the +-- follow record itself. A NOT NULL column would need a fabricated +-- placeholder that a later delete could accidentally match. NULL says +-- exactly what is true — "we do not know which record this came from" — +-- and a delete-by-rkey simply finds nothing for those rows, which is the +-- documented no-op path in `indexer::delete_follow_by_rkey`. Those rows +-- keep working through the push path (which sends `subject_did`) and +-- heal on their own the next time the follow is re-created. +-- +-- Re-follow under a new rkey +-- -------------------------- +-- Follow → unfollow → follow again produces a *different* rkey each +-- time (rkeys are TIDs; the client never reuses one). The upsert +-- therefore hits the primary key and overwrites `rkey` with the newer +-- record's: the youngest record wins. That ordering is what makes a +-- late or replayed delete for the *old* rkey harmless — it matches no +-- row and is skipped, instead of tearing down a follow that is +-- currently live. +-- +-- The index is NOT unique +-- ----------------------- +-- `(follower_did, rkey)` is unique in practice — an rkey identifies one +-- record inside one repo's collection — but a unique index would turn +-- the one situation this migration exists for into a *write failure*: +-- if a delete was lost and a create later reused that rkey, the insert +-- would abort instead of the stale row being cleaned up. An index whose +-- only job is to serve a lookup should not be able to reject a write. +-- Partial (`WHERE rkey IS NOT NULL`) because a lookup key is never +-- NULL, so the pre-migration rows have no business bloating it. + +ALTER TABLE follows ADD COLUMN IF NOT EXISTS rkey TEXT; + +-- Serves `DELETE FROM follows WHERE follower_did = $1 AND rkey = $2 +-- RETURNING subject_did` — the delete path for a firehose +-- unfollow, which is the only lookup this column exists for. +CREATE INDEX IF NOT EXISTS follows_follower_rkey_idx + ON follows (follower_did, rkey) + WHERE rkey IS NOT NULL;