Re-enqueue unmapped siblings after mapping

The tracker-mapping worker processes one pattern at a time and matchBySiblingOrigin only reads already-resolved siblings, so vendor propagation across a banner was forward-only. A sibling processed before its peer resolved a vendor (for example, one that failed the agent and fell back to an unmatched catalog row) was never revisited, even once a later sibling clearly identified the same third party. When a Process run newly establishes a common third party, re-arm mapping_requested_at on same-banner siblings that share an initiator domain and are still unpromoted and non-extension-sourced. The worker re-claims them and matchBySiblingOrigin now finds the freshly mapped pattern. Guarding on third_party_id IS NULL, mapping_requested_at IS NULL, and a not-pre-existing common third party keeps cascades finite. Signed-off-by: Émile Ré <emile@probo.com>
2026-05-29 01:10:59 +02:00
parent c11bc57c36
commit 8c10997681
3 changed files with 471 additions and 1 deletions
--- a/pkg/cookiebanner/tracker_mapping_worker.go
+++ b/pkg/cookiebanner/tracker_mapping_worker.go
@@ -131,6 +131,7 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
 				return fmt.Errorf("cannot load cookie banner for domain filtering: %w", err)
 			}

+			// FIXME: remove
 			banner.Origin = "https://t.probo.com"

 			var (
@@ -160,12 +161,23 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
 				}
 			}

+			// Whether a catalog third party was already known before the
+			// signal pipeline ran this round. Re-enqueuing siblings is
+			// only useful when this run is the one that resolves the
+			// vendor; a pre-existing link adds no new signal and gating
+			// on it keeps cascades finite.
+			commonThirdPartyPreexisted := commonThirdPartyID != nil
+
+			var domains []string
+
 			if commonThirdPartyID == nil {
-				domains, err := h.loadInitiatorDomains(ctx, tx, tp)
+				loaded, err := h.loadInitiatorDomains(ctx, tx, tp)
 				if err != nil {
 					return err
 				}

+				domains = loaded
+
 				// Sibling matching is an org-local co-occurrence signal:
 				// two patterns served from the same origin on the same
 				// banner are likely the same vendor, even when that origin
@@ -268,6 +280,17 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
 					log.String("pattern", tp.Pattern),
 					log.String("tracker_pattern_id", tp.ID.String()),
 				)
+
+				// This run newly resolved a catalog third party, so
+				// same-banner siblings that share an initiator domain but
+				// were processed earlier and left unmatched can now match
+				// against it. Re-arm their mapping so the worker revisits
+				// them; the guards keep already-mapped siblings untouched.
+				if commonThirdPartyID != nil && !commonThirdPartyPreexisted {
+					if err := h.reenqueueUnmappedSiblings(ctx, tx, tp, domains); err != nil {
+						return err
+					}
+				}
 			}

 			return nil
@@ -275,6 +298,43 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
 	)
 }

+// reenqueueUnmappedSiblings re-arms mapping_requested_at on same-banner
+// siblings sharing an initiator domain with tp that are still unpromoted,
+// so the worker re-evaluates them now that tp resolved a vendor.
+func (h *trackerMappingHandler) reenqueueUnmappedSiblings(
+	ctx context.Context,
+	tx pg.Tx,
+	tp coredata.TrackerPattern,
+	domains []string,
+) error {
+	scope := coredata.NewScopeFromObjectID(tp.ID)
+
+	var patterns coredata.TrackerPatterns
+
+	count, err := patterns.RequestMappingForUnmappedSiblings(
+		ctx,
+		tx,
+		scope,
+		tp.CookieBannerID,
+		tp.ID,
+		domains,
+	)
+	if err != nil {
+		return fmt.Errorf("cannot re-enqueue unmapped siblings: %w", err)
+	}
+
+	if count > 0 {
+		h.logger.InfoCtx(
+			ctx,
+			"re-enqueued unmapped sibling tracker patterns",
+			log.String("tracker_pattern_id", tp.ID.String()),
+			log.Int64("count", count),
+		)
+	}
+
+	return nil
+}
+
 // firstNonNil returns a when it is set, otherwise b. It keeps the first
 // catalog row id resolved by the pipeline stable: later signals upsert
 // the same row (same key) and return the same id, but the explicit guard
--- a/pkg/cookiebanner/tracker_mapping_worker_test.go
+++ b/pkg/cookiebanner/tracker_mapping_worker_test.go
@@ -1425,3 +1425,348 @@ func TestProcess_SiblingPromotionOnFirstPartyOrigin(t *testing.T) {
 	require.NotNil(t, reloadedTarget.ThirdPartyID, "target sharing a first-party origin must be promoted via its sibling")
 	assert.Equal(t, orgThirdParty.ID, *reloadedTarget.ThirdPartyID)
 }
+
+// TestProcess_ReenqueuesUnmappedSiblingOnResolve asserts that when a
+// pattern newly resolves a catalog third party, same-banner siblings
+// that share an initiator domain but are still unpromoted get their
+// mapping re-armed (backward propagation), while the already-promoted
+// sibling that supplied the resolution is left untouched.
+func TestProcess_ReenqueuesUnmappedSiblingOnResolve(t *testing.T) {
+	t.Parallel()
+
+	client := newTestPgClient(t)
+	ctx := context.Background()
+	fx := seedPromotionFixture(t, ctx, client)
+
+	now := time.Now().UTC().Truncate(time.Microsecond)
+
+	orgThirdParty := coredata.ThirdParty{
+		ID:                 gid.New(fx.scope.GetTenantID(), coredata.ThirdPartyEntityType),
+		OrganizationID:     fx.organizationID,
+		CommonThirdPartyID: &fx.commonThirdPartyID,
+		Name:               "Google LLC",
+		Category:           coredata.ThirdPartyCategoryAnalytics,
+		Certifications:     []string{},
+		Countries:          coredata.CountryCodes{},
+		CreatedAt:          now,
+		UpdatedAt:          now,
+	}
+
+	mappedSibling := coredata.TrackerPattern{
+		ID:                     gid.New(fx.scope.GetTenantID(), coredata.TrackerPatternEntityType),
+		OrganizationID:         fx.organizationID,
+		CookieBannerID:         fx.banner.ID,
+		CookieCategoryID:       fx.normalCategoryID,
+		CommonTrackerPatternID: &fx.commonPatternID,
+		ThirdPartyID:           &orgThirdParty.ID,
+		TrackerType:            coredata.TrackerTypeCookie,
+		Pattern:                "_gid_reenq",
+		MatchType:              coredata.TrackerPatternMatchTypeExact,
+		DisplayName:            "_gid_reenq",
+		CreatedAt:              now,
+		UpdatedAt:              now,
+	}
+
+	target := coredata.TrackerPattern{
+		ID:               gid.New(fx.scope.GetTenantID(), coredata.TrackerPatternEntityType),
+		OrganizationID:   fx.organizationID,
+		CookieBannerID:   fx.banner.ID,
+		CookieCategoryID: fx.normalCategoryID,
+		TrackerType:      coredata.TrackerTypeCookie,
+		Pattern:          "_ga_reenq_target",
+		MatchType:        coredata.TrackerPatternMatchTypeExact,
+		DisplayName:      "_ga_reenq_target",
+		CreatedAt:        now,
+		UpdatedAt:        now,
+	}
+
+	unmappedSibling := coredata.TrackerPattern{
+		ID:               gid.New(fx.scope.GetTenantID(), coredata.TrackerPatternEntityType),
+		OrganizationID:   fx.organizationID,
+		CookieBannerID:   fx.banner.ID,
+		CookieCategoryID: fx.normalCategoryID,
+		TrackerType:      coredata.TrackerTypeCookie,
+		Pattern:          "_unmapped_reenq",
+		MatchType:        coredata.TrackerPatternMatchTypeExact,
+		DisplayName:      "_unmapped_reenq",
+		CreatedAt:        now,
+		UpdatedAt:        now,
+	}
+
+	sharedDomain := "reenq-tracker.com"
+
+	require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
+		if err := orgThirdParty.Insert(ctx, tx, fx.scope); err != nil {
+			return err
+		}
+
+		for _, p := range []coredata.TrackerPattern{mappedSibling, target, unmappedSibling} {
+			if err := p.Insert(ctx, tx, fx.scope); err != nil {
+				return err
+			}
+		}
+
+		for id, identifier := range map[gid.GID]string{
+			mappedSibling.ID:   "_gid_reenq",
+			target.ID:          "_ga_reenq_target",
+			unmappedSibling.ID: "_unmapped_reenq",
+		} {
+			patternID := id
+			det := coredata.DetectedTracker{
+				ID:               gid.New(fx.scope.GetTenantID(), coredata.DetectedTrackerEntityType),
+				CookieBannerID:   fx.banner.ID,
+				TrackerPatternID: &patternID,
+				TrackerType:      coredata.TrackerTypeCookie,
+				Identifier:       identifier,
+				InitiatorDomain:  &sharedDomain,
+				LastDetectedAt:   now,
+				CreatedAt:        now,
+				UpdatedAt:        now,
+			}
+
+			if _, err := det.Upsert(ctx, tx, fx.scope); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	}))
+
+	h := newMappingHandler(client)
+	require.NoError(t, h.Process(ctx, target))
+
+	var reloadedTarget coredata.TrackerPattern
+
+	require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
+		return reloadedTarget.LoadByID(ctx, conn, fx.scope, target.ID)
+	}))
+
+	require.NotNil(t, reloadedTarget.ThirdPartyID, "target must resolve via its promoted sibling")
+
+	var reloadedUnmapped coredata.TrackerPattern
+
+	require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
+		return reloadedUnmapped.LoadByID(ctx, conn, fx.scope, unmappedSibling.ID)
+	}))
+
+	require.NotNil(t, reloadedUnmapped.MappingRequestedAt, "unmapped sibling sharing the origin must be re-enqueued")
+
+	var reloadedMapped coredata.TrackerPattern
+
+	require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
+		return reloadedMapped.LoadByID(ctx, conn, fx.scope, mappedSibling.ID)
+	}))
+
+	assert.Nil(t, reloadedMapped.MappingRequestedAt, "already-promoted sibling must not be re-enqueued")
+}
+
+// TestProcess_DoesNotReenqueuePromotedOrExtensionSiblings asserts that
+// the re-enqueue skips siblings that are already promoted or
+// EXTENSION-sourced, while still re-arming a plain unmapped sibling.
+func TestProcess_DoesNotReenqueuePromotedOrExtensionSiblings(t *testing.T) {
+	t.Parallel()
+
+	client := newTestPgClient(t)
+	ctx := context.Background()
+	fx := seedPromotionFixture(t, ctx, client)
+
+	now := time.Now().UTC().Truncate(time.Microsecond)
+
+	orgThirdParty := coredata.ThirdParty{
+		ID:                 gid.New(fx.scope.GetTenantID(), coredata.ThirdPartyEntityType),
+		OrganizationID:     fx.organizationID,
+		CommonThirdPartyID: &fx.commonThirdPartyID,
+		Name:               "Google LLC",
+		Category:           coredata.ThirdPartyCategoryAnalytics,
+		Certifications:     []string{},
+		Countries:          coredata.CountryCodes{},
+		CreatedAt:          now,
+		UpdatedAt:          now,
+	}
+
+	mappedSibling := coredata.TrackerPattern{
+		ID:                     gid.New(fx.scope.GetTenantID(), coredata.TrackerPatternEntityType),
+		OrganizationID:         fx.organizationID,
+		CookieBannerID:         fx.banner.ID,
+		CookieCategoryID:       fx.normalCategoryID,
+		CommonTrackerPatternID: &fx.commonPatternID,
+		ThirdPartyID:           &orgThirdParty.ID,
+		TrackerType:            coredata.TrackerTypeCookie,
+		Pattern:                "_gid_guard",
+		MatchType:              coredata.TrackerPatternMatchTypeExact,
+		DisplayName:            "_gid_guard",
+		CreatedAt:              now,
+		UpdatedAt:              now,
+	}
+
+	target := coredata.TrackerPattern{
+		ID:               gid.New(fx.scope.GetTenantID(), coredata.TrackerPatternEntityType),
+		OrganizationID:   fx.organizationID,
+		CookieBannerID:   fx.banner.ID,
+		CookieCategoryID: fx.normalCategoryID,
+		TrackerType:      coredata.TrackerTypeCookie,
+		Pattern:          "_ga_guard_target",
+		MatchType:        coredata.TrackerPatternMatchTypeExact,
+		DisplayName:      "_ga_guard_target",
+		CreatedAt:        now,
+		UpdatedAt:        now,
+	}
+
+	plainSibling := coredata.TrackerPattern{
+		ID:               gid.New(fx.scope.GetTenantID(), coredata.TrackerPatternEntityType),
+		OrganizationID:   fx.organizationID,
+		CookieBannerID:   fx.banner.ID,
+		CookieCategoryID: fx.normalCategoryID,
+		TrackerType:      coredata.TrackerTypeCookie,
+		Pattern:          "_plain_guard",
+		MatchType:        coredata.TrackerPatternMatchTypeExact,
+		DisplayName:      "_plain_guard",
+		CreatedAt:        now,
+		UpdatedAt:        now,
+	}
+
+	extensionSource := coredata.CookieSourceExtension
+	extensionSibling := coredata.TrackerPattern{
+		ID:               gid.New(fx.scope.GetTenantID(), coredata.TrackerPatternEntityType),
+		OrganizationID:   fx.organizationID,
+		CookieBannerID:   fx.banner.ID,
+		CookieCategoryID: fx.normalCategoryID,
+		TrackerType:      coredata.TrackerTypeCookie,
+		Pattern:          "_ext_guard",
+		MatchType:        coredata.TrackerPatternMatchTypeExact,
+		DisplayName:      "_ext_guard",
+		Source:           &extensionSource,
+		CreatedAt:        now,
+		UpdatedAt:        now,
+	}
+
+	sharedDomain := "guard-tracker.com"
+
+	require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
+		if err := orgThirdParty.Insert(ctx, tx, fx.scope); err != nil {
+			return err
+		}
+
+		patterns := map[gid.GID]string{
+			mappedSibling.ID:    "_gid_guard",
+			target.ID:           "_ga_guard_target",
+			plainSibling.ID:     "_plain_guard",
+			extensionSibling.ID: "_ext_guard",
+		}
+
+		for _, p := range []coredata.TrackerPattern{mappedSibling, target, plainSibling, extensionSibling} {
+			if err := p.Insert(ctx, tx, fx.scope); err != nil {
+				return err
+			}
+		}
+
+		for id, identifier := range patterns {
+			patternID := id
+			det := coredata.DetectedTracker{
+				ID:               gid.New(fx.scope.GetTenantID(), coredata.DetectedTrackerEntityType),
+				CookieBannerID:   fx.banner.ID,
+				TrackerPatternID: &patternID,
+				TrackerType:      coredata.TrackerTypeCookie,
+				Identifier:       identifier,
+				InitiatorDomain:  &sharedDomain,
+				LastDetectedAt:   now,
+				CreatedAt:        now,
+				UpdatedAt:        now,
+			}
+
+			if _, err := det.Upsert(ctx, tx, fx.scope); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	}))
+
+	h := newMappingHandler(client)
+	require.NoError(t, h.Process(ctx, target))
+
+	reload := func(id gid.GID) coredata.TrackerPattern {
+		var p coredata.TrackerPattern
+
+		require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
+			return p.LoadByID(ctx, conn, fx.scope, id)
+		}))
+
+		return p
+	}
+
+	require.NotNil(t, reload(target.ID).ThirdPartyID, "target must resolve via its promoted sibling")
+	require.NotNil(t, reload(plainSibling.ID).MappingRequestedAt, "plain unmapped sibling must be re-enqueued")
+	assert.Nil(t, reload(mappedSibling.ID).MappingRequestedAt, "promoted sibling must not be re-enqueued")
+	assert.Nil(t, reload(extensionSibling.ID).MappingRequestedAt, "EXTENSION-sourced sibling must not be re-enqueued")
+}
+
+// TestProcess_NoReenqueueWhenCommonThirdPartyPreexisted asserts that the
+// re-trigger path, where the pattern's linked catalog row already carries
+// a common third party, adds no new signal and therefore leaves unmapped
+// siblings untouched (the cascade terminator).
+func TestProcess_NoReenqueueWhenCommonThirdPartyPreexisted(t *testing.T) {
+	t.Parallel()
+
+	client := newTestPgClient(t)
+	ctx := context.Background()
+	fx := seedPromotionFixture(t, ctx, client)
+
+	now := time.Now().UTC().Truncate(time.Microsecond)
+
+	unmappedSibling := coredata.TrackerPattern{
+		ID:               gid.New(fx.scope.GetTenantID(), coredata.TrackerPatternEntityType),
+		OrganizationID:   fx.organizationID,
+		CookieBannerID:   fx.banner.ID,
+		CookieCategoryID: fx.normalCategoryID,
+		TrackerType:      coredata.TrackerTypeCookie,
+		Pattern:          "_unmapped_preexist",
+		MatchType:        coredata.TrackerPatternMatchTypeExact,
+		DisplayName:      "_unmapped_preexist",
+		CreatedAt:        now,
+		UpdatedAt:        now,
+	}
+
+	sharedDomain := "preexist-tracker.com"
+
+	require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
+		if err := unmappedSibling.Insert(ctx, tx, fx.scope); err != nil {
+			return err
+		}
+
+		for id, identifier := range map[gid.GID]string{
+			fx.trackerPattern.ID: fx.trackerPattern.Pattern,
+			unmappedSibling.ID:   "_unmapped_preexist",
+		} {
+			patternID := id
+			det := coredata.DetectedTracker{
+				ID:               gid.New(fx.scope.GetTenantID(), coredata.DetectedTrackerEntityType),
+				CookieBannerID:   fx.banner.ID,
+				TrackerPatternID: &patternID,
+				TrackerType:      coredata.TrackerTypeCookie,
+				Identifier:       identifier,
+				InitiatorDomain:  &sharedDomain,
+				LastDetectedAt:   now,
+				CreatedAt:        now,
+				UpdatedAt:        now,
+			}
+
+			if _, err := det.Upsert(ctx, tx, fx.scope); err != nil {
+				return err
+			}
+		}
+
+		return fx.trackerPattern.SetMappingRequested(ctx, tx)
+	}))
+
+	h := newMappingHandler(client)
+	require.NoError(t, h.Process(ctx, fx.trackerPattern))
+
+	var reloadedUnmapped coredata.TrackerPattern
+
+	require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
+		return reloadedUnmapped.LoadByID(ctx, conn, fx.scope, unmappedSibling.ID)
+	}))
+
+	assert.Nil(t, reloadedUnmapped.MappingRequestedAt, "re-trigger with a pre-existing common third party must not re-enqueue siblings")
+}
--- a/pkg/coredata/tracker_pattern.go
+++ b/pkg/coredata/tracker_pattern.go
@@ -1184,3 +1184,68 @@ WHERE id = @id

 	return nil
 }
+
+// RequestMappingForUnmappedSiblings re-arms mapping_requested_at on
+// sibling tracker patterns of the same banner that share an initiator
+// domain with the just-mapped pattern but are still unpromoted. It is
+// the backward-propagation counterpart to the mapping worker's
+// sibling-origin matching: when a pattern newly resolves a vendor, its
+// siblings that were processed earlier and left unmatched can now be
+// re-evaluated against it.
+//
+// Only unpromoted (third_party_id IS NULL), not-already-queued
+// (mapping_requested_at IS NULL), non-extension siblings are touched, so
+// a fully mapped banner re-enqueues nothing. detected_trackers is used
+// only as a filtering subquery. Returns the number of siblings
+// re-enqueued.
+func (tps *TrackerPatterns) RequestMappingForUnmappedSiblings(
+	ctx context.Context,
+	tx pg.Tx,
+	scope Scoper,
+	cookieBannerID gid.GID,
+	excludePatternID gid.GID,
+	domains []string,
+) (int64, error) {
+	if len(domains) == 0 {
+		return 0, nil
+	}
+
+	q := `
+UPDATE tracker_patterns
+SET
+	mapping_requested_at = NOW(),
+	updated_at = NOW()
+WHERE
+	%[1]s
+	AND cookie_banner_id = @cookie_banner_id
+	AND id != @exclude_pattern_id
+	AND third_party_id IS NULL
+	AND mapping_requested_at IS NULL
+	AND (source IS NULL OR source != @extension_source)
+	AND id IN (
+		SELECT DISTINCT tracker_pattern_id
+		FROM detected_trackers
+		WHERE %[1]s
+			AND cookie_banner_id = @cookie_banner_id
+			AND initiator_domain = ANY(@domains)
+			AND tracker_pattern_id IS NOT NULL
+	)
+`
+
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"cookie_banner_id":   cookieBannerID,
+		"exclude_pattern_id": excludePatternID,
+		"extension_source":   CookieSourceExtension,
+		"domains":            domains,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := tx.Exec(ctx, q, args)
+	if err != nil {
+		return 0, fmt.Errorf("cannot request mapping for unmapped siblings: %w", err)
+	}
+
+	return result.RowsAffected(), nil
+}