diff --git a/internal/app/websitetracking/service.go b/internal/app/websitetracking/service.go index 42b50fec..4ddca32d 100644 --- a/internal/app/websitetracking/service.go +++ b/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; diff --git a/tracking/src/handlers.rs b/tracking/src/handlers.rs index c2d2a583..68ea44b8 100644 --- a/tracking/src/handlers.rs +++ b/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() } } diff --git a/tracking/src/hits.rs b/tracking/src/hits.rs index fcbf7e48..26082199 100644 --- a/tracking/src/hits.rs +++ b/tracking/src/hits.rs @@ -139,7 +139,9 @@ pub struct HitForwarder { internal_token: String, unknown_sites: Cache, miss_budget: Cache>, - dedupe: Cache, + /// (visitor, URL) -> the id of the request that owns the entry. + dedupe: Cache, + 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 { + 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());