feat: drop a page hit instead of storing it under the wrong contact when tying a browser to a ticket fails after a lost identification race (attach now fails closed and the edge retries), and make the edge dedupe an atomic request-owned claim so a failed concurrent forward can only release its own entry and never a successful request's

internal/app/websitetracking/service.go

@@ -336,7 +336,11 @@ func (s *service) Ingest(ctx context.Context, req *models.WebsiteHitRequest) (*m
 			}
 			// A ticket from another workspace's campaign proves nothing here.
 			if terr == nil && ok && ticketOrg == site.OrganizationID {
-				visitor, result.NewVisitorKey = s.attach(ctx, site.OrganizationID, visitor, contactID, now)
+				attached, key, aerr := s.attach(ctx, site.OrganizationID, visitor, contactID, now)
+				if aerr != nil {
+					return nil, aerr
+				}
+				visitor, result.NewVisitorKey = attached, key
 			}
 		}
 	}
@@ -358,37 +362,43 @@ func (s *service) Ingest(ctx context.Context, req *models.WebsiteHitRequest) (*m
 	return result, nil
 }
 
+// errAttach means the browser could not be tied to the ticket's contact
+// safely; the hit is dropped rather than stored under the wrong person, and
+// the edge retries it.
+var errAttach = errors.New("visitor attach failed")
+
 // attach ties the browser to the ticket's contact. An anonymous row is
 // claimed with a guarded update; if a concurrent hit won that race for a
 // different contact, or the row already belonged to someone else, the browser
-// is split onto a fresh row so neither contact inherits the other's history.
-func (s *service) attach(ctx context.Context, orgID uuid.UUID, visitor *models.WebsiteVisitor, contactID uuid.UUID, now time.Time) (*models.WebsiteVisitor, string) {
+// is split onto a fresh row. Any failure on the way returns errAttach so the
+// caller never inserts under a row it does not own.
+func (s *service) attach(ctx context.Context, orgID uuid.UUID, visitor *models.WebsiteVisitor, contactID uuid.UUID, now time.Time) (*models.WebsiteVisitor, string, error) {
 	if visitor.ContactID == nil {
 		updated, err := s.repo.IdentifyVisitor(ctx, visitor.ID, contactID, "email_link", now)
 		if err != nil {
 			db.CaptureError(err, "", nil, "websitetracking IdentifyVisitor")
-			return visitor, ""
+			return nil, "", errAttach
 		}
 		if updated {
 			visitor.ContactID = &contactID
-			return visitor, ""
+			return visitor, "", nil
 		}
 		current, err := s.repo.GetVisitorByID(ctx, visitor.ID)
 		if err != nil {
 			db.CaptureError(err, "", nil, "websitetracking GetVisitorByID")
-			return visitor, ""
+			return nil, "", errAttach
 		}
 		visitor = current
 	}
 	if visitor.ContactID != nil && *visitor.ContactID == contactID {
-		return visitor, ""
+		return visitor, "", nil
 	}
 	fresh, err := s.repo.CreateVisitor(ctx, orgID, newKey(), contactID, "email_link", now)
 	if err != nil {
 		db.CaptureError(err, "", nil, "websitetracking CreateVisitor")
-		return visitor, ""
+		return nil, "", errAttach
 	}
-	return fresh, fresh.VisitorKey
+	return fresh, fresh.VisitorKey, nil
 }
 
 // locate fills location from the request IP at the workspace's precision;

tracking/src/handlers.rs

@@ -346,9 +346,9 @@ pub async fn track_page_hit(
     }
 
     // Reloads and double-fires: acknowledged, not counted.
-    if state.hits.is_duplicate(&payload.v, &payload.u).await {
+    let Some(claim) = state.hits.claim(&payload.v, &payload.u).await else {
         return StatusCode::NO_CONTENT.into_response();
-    }
+    };
 
     let origin_host = headers
         .get(header::ORIGIN)
@@ -392,7 +392,7 @@ pub async fn track_page_hit(
         Outcome::Malformed => (StatusCode::BAD_REQUEST, "Invalid payload").into_response(),
         Outcome::Unavailable => {
             // Nothing was stored, so the next attempt must not be deduped away.
-            state.hits.forget(&dedupe_visitor, &dedupe_url).await;
+            state.hits.forget(&dedupe_visitor, &dedupe_url, claim).await;
             (StatusCode::SERVICE_UNAVAILABLE, "Try again shortly").into_response()
         }
     }

tracking/src/hits.rs

@@ -139,7 +139,9 @@ pub struct HitForwarder {
     internal_token: String,
     unknown_sites: Cache<String, ()>,
     miss_budget: Cache<String, Arc<AtomicU32>>,
-    dedupe: Cache<String, ()>,
+    /// (visitor, URL) -> the id of the request that owns the entry.
+    dedupe: Cache<String, u64>,
+    next_request: AtomicU64,
     breaker_failures: AtomicU32,
     breaker_open_until_ms: AtomicU64,
     started: Instant,
@@ -168,27 +170,39 @@ impl HitForwarder {
                 .max_capacity(200_000)
                 .time_to_live(DEDUPE_WINDOW)
                 .build(),
+            next_request: AtomicU64::new(1),
             breaker_failures: AtomicU32::new(0),
             breaker_open_until_ms: AtomicU64::new(0),
             started: Instant::now(),
         }
     }
 
-    /// True when this browser already reported this URL inside the window.
-    pub async fn is_duplicate(&self, visitor: &str, url: &str) -> bool {
-        let key = format!("{}:{}", visitor, url);
-        if self.dedupe.contains_key(&key) {
-            return true;
+    /// Claims the (visitor, URL) pair for this request. None when another
+    /// request inside the window already owns it; otherwise the claim token
+    /// to hand back to `forget` if the forward fails. The insert is atomic, so
+    /// two concurrent requests cannot both pass.
+    pub async fn claim(&self, visitor: &str, url: &str) -> Option<u64> {
+        let token = self.next_request.fetch_add(1, Ordering::Relaxed);
+        let entry = self
+            .dedupe
+            .entry(format!("{}:{}", visitor, url))
+            .or_insert(token)
+            .await;
+        if entry.is_fresh() {
+            Some(token)
+        } else {
+            None
         }
-        self.dedupe.insert(key, ()).await;
-        false
     }
 
-    /// Drops a dedupe entry after a failed forward so the retry is counted.
-    pub async fn forget(&self, visitor: &str, url: &str) {
-        self.dedupe
-            .invalidate(&format!("{}:{}", visitor, url))
-            .await;
+    /// Releases a claim after a failed forward so the retry is counted. Only
+    /// the owning request can release it: a concurrent request that succeeded
+    /// keeps its entry.
+    pub async fn forget(&self, visitor: &str, url: &str, token: u64) {
+        let key = format!("{}:{}", visitor, url);
+        if self.dedupe.get(&key).await == Some(token) {
+            self.dedupe.invalidate(&key).await;
+        }
     }
 
     pub async fn forward(&self, hit: ForwardedHit, source: &str) -> Outcome {
@@ -309,6 +323,22 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn claim_is_exclusive_and_forget_only_releases_the_owner() {
+        let f = HitForwarder::new("http://127.0.0.1:9".into(), "t".into());
+        let a = f.claim("v", "u").await.expect("first claim wins");
+        assert!(
+            f.claim("v", "u").await.is_none(),
+            "second claim inside the window is a duplicate"
+        );
+        // A stale token (a concurrent loser) cannot release the winner's entry.
+        f.forget("v", "u", a + 1).await;
+        assert!(f.claim("v", "u").await.is_none());
+        // The owner can.
+        f.forget("v", "u", a).await;
+        assert!(f.claim("v", "u").await.is_some());
+    }
+
     #[test]
     fn accepts_a_plain_page_view() {
         assert!(payload("https://example.com/pricing").validate());