Tweepcred: Clamp + Entropy Guard to prevent coordinated lists & brigades

src/scala/com/twitter/graph/batch/job/tweepcred/Reputation.scala

@@ -29,20 +29,50 @@ object Reputation {
   private val constantDivisionFactorGt_threshFriendsToFollowersRatioReps = 3.0
   private val threshFriendsToFollowersRatioUMass = 0.6
   private val maxDivFactorReps = 50
+  private val maxDailyNegDelta = -100.0 // NEW: legacy cap retained for compatibility
+  private val normalCap = -100.0 // NEW: dynamic cap (normal conditions)
+  private val attackCap = -20.0 // NEW: dynamic cap when a brigade is detected
+  private val brigadeDiversityThreshold = 0.5 // NEW: <50% unique issuers => coordinated attack
+
+  private def incrCounter(name: String): Unit = () // NEW: telemetry stub; no-op to preserve structure
 
   /**
    * reduce pagerank of users with low followers but high followings
    */
   def adjustReputationsPostCalculation(mass: Double, numFollowers: Int, numFollowings: Int) = {
+    adjustReputationsPostCalculation(mass, numFollowers, numFollowings, Seq.empty) // NEW: delegate to extended method with empty issuers
+  }
+
+  /**
+   * NEW overload to support entropy/diversity guard and dynamic caps
+   */
+  def adjustReputationsPostCalculation(mass: Double, numFollowers: Int, numFollowings: Int, blockIssuers: Seq[Long]) = { // NEW: extended signature with issuers
     if (numFollowings > threshAbsNumFriendsReps) {
-      val friendsToFollowersRatio = (1.0 + numFollowings) / (1.0 + numFollowers)
-      val divFactor =
-        scala.math.exp(
-          constantDivisionFactorGt_threshFriendsToFollowersRatioReps *
-            (friendsToFollowersRatio - threshFriendsToFollowersRatioUMass) *
-            scala.math.log(scala.math.log(numFollowings))
-        )
-      mass / ((divFactor min maxDivFactorReps) max 1.0)
+
+      if (mass.isNaN || mass.isInfinity) return mass
+      if (numFollowers < 0 || numFollowings < 0) return mass
+
+      val baseMass = math.max(0.0, mass)
+
+      val friendsToFollowersRatio = (1.0 + numFollowings.toDouble) / (1.0 + numFollowers.toDouble)
+      val divFactorRaw = math.exp(
+        constantDivisionFactorGt_threshFriendsToFollowersRatioReps *
+          (friendsToFollowersRatio - threshFriendsToFollowersRatioUMass) *
+          math.log(scala.math.log1p(numFollowings.toDouble))
+      )
+      val divFactor = ((divFactorRaw min maxDivFactorReps) max 1.0)
+      val adjusted = baseMass / divFactor
+
+      val diversity = if (blockIssuers.nonEmpty) blockIssuers.distinct.size.toDouble / blockIssuers.size else 1.0 // NEW: issuer diversity (unique/total)
+      val isBrigade = diversity < brigadeDiversityThreshold // NEW: boolean flag for coordinated attack
+      val cap = if (isBrigade) attackCap else normalCap // NEW: select dynamic cap based on brigade flag
+      if (isBrigade) incrCounter("tweepcred.brigade.detected") // NEW: telemetry when brigade detected
+
+      val delta = adjusted - baseMass // FIX: correct penalty sign
+      val deltaCapped = math.min(0.0, math.max(delta, cap)) // FIX: clamp penalty to cap, never positive
+      if (delta < cap) incrCounter(if (isBrigade) "tweepcred.delta.capped.attack" else "tweepcred.delta.capped.normal") // NEW: telemetry when clamp engages
+
+      baseMass + deltaCapped // apply capped delta
     } else {
       mass
     }