abuse-enforcement-service: fail closed when the allowlist store errors

README.md

@@ -109,8 +109,8 @@ Ranking sets the order. Whether a post can be shown at all is decided separately
 │  ┌────────────────────────────────────────────────────────────────────────────────────┐  │
 │  │ 5. SCORING                                                                         │  │
 │  │    <a href="home-mixer/scorers/phoenix_scorer.rs">PhoenixScorer</a>   a probability for each action the viewer might take             │  │
-│  │    <a href="home-mixer/scorers/ranking_scorer.rs">RankingScorer</a>   weighted sum, then repeated-author decay, an                    │  │
-│  │                    out-of-network discount, a new-author boost                     │  │
+│  │    <a href="home-mixer/scorers/ranking_scorer.rs">RankingScorer</a>   weighted sum, then author-size IPS, origin-author            │  │
+│  │                    diversity decay, size-aware OON discount, new-author boost      │  │
 │  │    <a href="home-mixer/scorers/vm_ranker.rs">VMRanker</a>        calls the reranking service in <a href="vm-ranker/">vm-ranker/</a>                       │  │
 │  └────────────────────────────────────────────────────────────────────────────────────┘  │
 │                                            ▼                                             │
@@ -329,10 +329,11 @@ Final Score = Σ (weight_i × P(action_i))
 
 Positive actions carry positive weights, negative actions negative ones. The weights are in [`home-mixer/params/param.rs`](home-mixer/params/param.rs); the arithmetic is in [`home-mixer/scorers/ranking_scorer.rs`](home-mixer/scorers/ranking_scorer.rs).
 
-Three adjustments follow:
+Four adjustments follow:
 
-- **Author Diversity**: each post after an author's first is multiplied by a decaying factor, down to a floor.
-- **Out-of-Network Discount**: posts from accounts the viewer does not follow are multiplied by a factor below 1, as are replies and reposts from accounts the viewer does follow.
+- **Author-size IPS**: a batch-mean-normalized inverse-propensity multiplier on `ln(1 + followers)`. Equal Phoenix scores are not ranked by audience size. Quality still wins: the default clamp is 2x, so a large-author post that is 3x more relevant still ranks higher. Math: [`docs/MERITOCRATIC_AUTHOR_SIZE_IPS.md`](docs/MERITOCRATIC_AUTHOR_SIZE_IPS.md).
+- **Author Diversity**: each post after an author's first is multiplied by a decaying factor, down to a floor. By default the decay key is the *content origin* (`retweeted_user_id` when present), so a viral original cannot flood the slate through many distinct retweeters.
+- **Out-of-Network Discount**: posts from accounts the viewer does not follow are multiplied by a factor below 1, as are replies and reposts from accounts the viewer does follow. Size-aware relief softens that tax for small authors (default: half relief below 1k followers, tapering to none by 100k) so discovery is not double-penalized by audience size and the follow graph. See [`docs/FEED_FAIRNESS.md`](docs/FEED_FAIRNESS.md).
 - **New-Author Boost**: posts from authors whose impressions are below a threshold are lifted toward a target position.
 
 `VMRanker` then calls [`vm-ranker/`](vm-ranker/), a separate service that reorders the result.
@@ -356,7 +357,7 @@ Three adjustments follow:
 | `PreviouslySeenPostsBackupFilter` | The same, from a second record of impressions                                                     |
 | `PreviouslyServedPostsFilter`     | Posts already served earlier in the session                                                       |
 | `MutedKeywordFilter`              | Posts matching the viewer's muted keywords                                                        |
-| `AuthorSocialgraphFilter`         | Posts from accounts the viewer blocks or mutes                                                    |
+| `AuthorSocialgraphFilter`         | Posts from accounts the viewer blocks or mutes, including quotes/reposts of muted or blocked authors |
 | `VideoFilter`                     | Video posts, when the request excludes video                                                      |
 | `TopicIdsFilter`                  | Posts outside the requested topics, and posts in excluded topics                                  |
 | `NewUserMinEngagementFilter`      | For new accounts, out-of-network posts below an engagement threshold                              |

abuse-enforcement-service/service-lib/src/allowlist.rs

@@ -29,15 +29,17 @@ impl ManhattanAllowlist {
         }
     }
 
-    pub async fn get(&self, user_id: i64) -> Option<AllowlistRecord> {
-        let (entry, ttl_secs) = self.get_entity(EntityType::User, user_id).await?;
-        Some(AllowlistRecord {
-            user_id,
-            added_by: entry.added_by,
-            reason: entry.reason,
-            added_at: entry.added_at,
-            ttl_secs,
-        })
+    pub async fn get(&self, user_id: i64) -> anyhow::Result<Option<AllowlistRecord>> {
+        Ok(self
+            .get_entity(EntityType::User, user_id)
+            .await?
+            .map(|(entry, ttl_secs)| AllowlistRecord {
+                user_id,
+                added_by: entry.added_by,
+                reason: entry.reason,
+                added_at: entry.added_at,
+                ttl_secs,
+            }))
     }
 
     pub async fn add(
@@ -54,30 +56,41 @@ impl ManhattanAllowlist {
         self.remove_entity(EntityType::User, user_id).await
     }
 
+    /// Looks up an allowlist entry. `Ok(None)` means the store confirmed the
+    /// entity is absent. A store or decode failure is returned as `Err` so
+    /// callers can distinguish "not allowlisted" from "could not check", and
+    /// never treat an unreadable entry as not allowlisted.
     pub async fn get_entity(
         &self,
         entity_type: EntityType,
         entity_id: i64,
-    ) -> Option<(AllowlistEntry, i64)> {
+    ) -> anyhow::Result<Option<(AllowlistEntry, i64)>> {
         let lkey = Self::entity_lkey(entity_type, entity_id);
         let item = match self
             .client
             .get(self.tenant.clone(), [MH_PKEY], [lkey.as_str()])
             .await
         {
-            Ok(v) => v?,
+            Ok(Some(item)) => item,
+            Ok(None) => return Ok(None),
             Err(e) => {
-                warn!("Manhattan allowlist GET failed: {e}");
+                warn!("Manhattan allowlist GET failed for {lkey}: {e}");
                 crate::metrics::MANHATTAN_ERRORS_TOTAL.inc();
-                return None;
+                return Err(anyhow::anyhow!(
+                    "Manhattan allowlist GET failed for {lkey}: {e}"
+                ));
             }
         };
-        let entry: AllowlistEntry = serde_json::from_slice(item.value().as_bytes()).ok()?;
+        let entry: AllowlistEntry =
+            serde_json::from_slice(item.value().as_bytes()).map_err(|e| {
+                warn!("Manhattan allowlist entry decode failed for {lkey}: {e}");
+                anyhow::anyhow!("Manhattan allowlist entry decode failed for {lkey}: {e}")
+            })?;
         let ttl_secs = crate::manhattan::remaining_ttl_secs(
             item.expires_at()
                 .map(|d| d.timestamp_nanos_opt().unwrap_or(0) as u64),
         );
-        Some((entry, ttl_secs))
+        Ok(Some((entry, ttl_secs)))
     }
 
     pub async fn add_entity(

abuse-enforcement-service/service-lib/src/lib.rs

@@ -218,7 +218,7 @@ async fn run_enforcement_inner(
     let mut facts = match entity_type {
         EntityType::User => {
             let user_id = entity_id;
-            let allowlist = fetch_user_allowlist(ctx.allowlist.as_ref(), user_id).await;
+            let allowlist = fetch_user_allowlist(ctx.allowlist.as_ref(), user_id).await?;
             if allowlist.is_allowlisted {
                 let partial = Facts {
                     entity_type,
@@ -265,6 +265,7 @@ async fn run_enforcement_inner(
                 fetch_entity_allowlist(ctx.allowlist.as_ref(), EntityType::Post, entity_id),
                 fetch_user_allowlist(ctx.allowlist.as_ref(), author_id),
             );
+            let (post_allowlist, author_allowlist) = (post_allowlist?, author_allowlist?);
             if post_allowlist.is_allowlisted || author_allowlist.is_allowlisted {
                 let reason = if post_allowlist.is_allowlisted {
                     "post_in_allowlist"

abuse-enforcement-service/service-lib/src/service.rs

@@ -1549,9 +1549,12 @@ pub async fn handle_allowlist_delete(
         );
     };
     info!("DELETE /api/allowlist/{user_id}");
+    // Audit snapshot only; a failed read must not block the delete.
     let before_json = al
         .get(user_id)
         .await
+        .ok()
+        .flatten()
         .map(|r| serde_json::to_string(&r).unwrap_or_default())
         .unwrap_or_default();
     let result = al.remove(user_id).await;
@@ -1619,7 +1622,21 @@ pub async fn bulk_allowlist_impl(
             continue;
         }
 
-        let existing_ttl = al.get(entry.user_id).await.map(|r| r.ttl_secs);
+        let existing_ttl = match al.get(entry.user_id).await {
+            Ok(existing) => existing.map(|r| r.ttl_secs),
+            Err(e) => {
+                errors += 1;
+                results.push(BulkRowResult {
+                    user_id: entry.user_id,
+                    ok: false,
+                    mode: String::new(),
+                    error: Some(format!("allowlist lookup failed: {e}")),
+                    existing_ttl_secs: None,
+                    ttl_secs: Some(entry.ttl_secs),
+                });
+                continue;
+            }
+        };
         let mode = if existing_ttl.is_some() {
             would_update += 1;
             "update"
@@ -1783,6 +1800,7 @@ pub async fn handle_allowlist_list(State(state): State<Arc<AppState>>) -> impl I
     responses(
         (status = 200, description = "Allowlist record for this user_id", body = AllowlistRecord),
         (status = 404, description = "No allowlist entry for this user_id"),
+        (status = 500, description = "Manhattan read failed"),
         (status = 503, description = "Allowlist not configured"),
     ),
 )]
@@ -1798,11 +1816,18 @@ pub async fn handle_allowlist_get(
         );
     };
     match al.get(user_id).await {
-        Some(record) => (
+        Ok(Some(record)) => (
             StatusCode::OK,
             Json(serde_json::to_value(&record).unwrap_or_default()),
         ),
-        None => (StatusCode::NOT_FOUND, Json(json!({"error": "not found"}))),
+        Ok(None) => (StatusCode::NOT_FOUND, Json(json!({"error": "not found"}))),
+        Err(e) => {
+            error!("GET /api/allowlist/{user_id} failed: {e}");
+            (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": e.to_string()})),
+            )
+        }
     }
 }
 
@@ -1820,6 +1845,7 @@ pub async fn handle_allowlist_get(
         (status = 200, description = "Allowlist record for this entity", body = Object),
         (status = 400, description = "Unknown entity_type"),
         (status = 404, description = "No allowlist entry for this entity"),
+        (status = 500, description = "Manhattan read failed"),
         (status = 503, description = "Allowlist not configured"),
     ),
 )]
@@ -1843,7 +1869,7 @@ pub async fn handle_allowlist_get_entity(
         );
     };
     match al.get_entity(et, entity_id).await {
-        Some((entry, ttl_secs)) => (
+        Ok(Some((entry, ttl_secs))) => (
             StatusCode::OK,
             Json(json!({
                 "entity_type": et.as_str(),
@@ -1854,7 +1880,14 @@ pub async fn handle_allowlist_get_entity(
                 "ttl_secs": ttl_secs,
             })),
         ),
-        None => (StatusCode::NOT_FOUND, Json(json!({"error": "not found"}))),
+        Ok(None) => (StatusCode::NOT_FOUND, Json(json!({"error": "not found"}))),
+        Err(e) => {
+            error!("GET /api/allowlist/{}/{entity_id} failed: {e}", et.as_str());
+            (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": e.to_string()})),
+            )
+        }
     }
 }
 
@@ -2009,9 +2042,12 @@ pub async fn handle_allowlist_delete_entity(
         );
     };
     info!("DELETE /api/allowlist/{}/{entity_id}", et.as_str());
+    // Audit snapshot only; a failed read must not block the delete.
     let before_json = al
         .get_entity(et, entity_id)
         .await
+        .ok()
+        .flatten()
         .map(|(entry, ttl_secs)| {
             json!({
                 "entity_type": et.as_str(),