Add stale recovery to tracker mapping worker

The tracker-mapping worker clears mapping_requested_at at claim time, so
a crash or hard failure between Process phases left the pattern dequeued,
unmapped, and with nothing to re-trigger it. Only an incidental sibling
remap could rescue it, so a lone pattern could stay stranded forever.

Implement the worker.StaleRecoverer interface, mirroring the enrichment
worker. ResetStaleMappings re-arms rows that were claimed but never
assigned a catalog row (common_tracker_pattern_id IS NULL) once idle past
a configurable window; a successful Process always assigns one via the
unmatched fallback, so the predicate cleanly detects interrupted runs and
self-heals after a single pass. ClearMappingRequestedAt now bumps
updated_at so the stale clock starts at claim time and the sweep never
recycles an in-flight claim.

Plumb a StaleAfter knob (default 600s) through the config struct, builder
env var, probod wiring, and Helm templates.

Signed-off-by: Émile Ré <emile@probo.com>

contrib/helm/charts/probo/templates/deployment.yaml

@@ -273,6 +273,10 @@ spec:
             - name: TRACKER_MAPPING_MAX_CONCURRENCY
               value: {{ .Values.probo.trackerMappingWorker.maxConcurrency | quote }}
             {{- end }}
+            {{- if .Values.probo.trackerMappingWorker.staleAfter }}
+            - name: TRACKER_MAPPING_STALE_AFTER
+              value: {{ .Values.probo.trackerMappingWorker.staleAfter | quote }}
+            {{- end }}
             {{- if .Values.probo.trackerMappingWorker.agentTimeout }}
             - name: TRACKER_MAPPING_AGENT_TIMEOUT
               value: {{ .Values.probo.trackerMappingWorker.agentTimeout | quote }}

contrib/helm/charts/probo/values-production.yaml.example

@@ -176,11 +176,13 @@ probo:
   #   temperature: "0.1"
   #   maxTokens: "4096"
 
-  # Tracker mapping worker tuning (optional; seconds for interval/agentTimeout).
+  # Tracker mapping worker tuning (optional; seconds for
+  # interval/staleAfter/agentTimeout).
   # Keep concurrency modest to stay under OpenAI/Firecrawl limits and the DB pool.
   # trackerMappingWorker:
   #   interval: 10
   #   maxConcurrency: 3
+  #   staleAfter: 600
   #   agentTimeout: 45
   #   agentMaxTurns: 4
 

contrib/helm/charts/probo/values.yaml

@@ -276,12 +276,14 @@ probo:
     temperature: ""
     maxTokens: ""
 
-  # Tracker mapping background worker tuning (optional). interval and
-  # agentTimeout are in seconds. Keep concurrency modest to stay under
-  # OpenAI/Firecrawl rate limits and the database connection pool.
+  # Tracker mapping background worker tuning (optional). interval,
+  # staleAfter, and agentTimeout are in seconds. Keep concurrency modest
+  # to stay under OpenAI/Firecrawl rate limits and the database
+  # connection pool.
   trackerMappingWorker:
     interval: 10
     maxConcurrency: 3
+    staleAfter: 600
     agentTimeout: 45
     agentMaxTurns: 4
 

pkg/bootstrap/builder.go

@@ -249,6 +249,7 @@ func (b *Builder) Build() (*probodconfig.FullConfig, error) {
 			TrackerMappingWorker: probodconfig.TrackerMappingWorkerConfig{
 				Interval:       b.getEnvIntOrDefault("TRACKER_MAPPING_INTERVAL", 10),
 				MaxConcurrency: b.getEnvIntOrDefault("TRACKER_MAPPING_MAX_CONCURRENCY", 3),
+				StaleAfter:     b.getEnvIntOrDefault("TRACKER_MAPPING_STALE_AFTER", 600),
 				AgentTimeout:   b.getEnvIntOrDefault("TRACKER_MAPPING_AGENT_TIMEOUT", 45),
 				AgentMaxTurns:  b.getEnvIntOrDefault("TRACKER_MAPPING_AGENT_MAX_TURNS", 4),
 			},

pkg/bootstrap/builder_test.go

@@ -223,6 +223,7 @@ func TestBuilder_Build_Defaults(t *testing.T) {
 	// Tracker worker tuning — defaults
 	assert.Equal(t, 10, cfg.Probod.TrackerMappingWorker.Interval)
 	assert.Equal(t, 3, cfg.Probod.TrackerMappingWorker.MaxConcurrency)
+	assert.Equal(t, 600, cfg.Probod.TrackerMappingWorker.StaleAfter)
 	assert.Equal(t, 45, cfg.Probod.TrackerMappingWorker.AgentTimeout)
 	assert.Equal(t, 4, cfg.Probod.TrackerMappingWorker.AgentMaxTurns)
 	assert.Equal(t, 10, cfg.Probod.CommonPatternEnrichmentWorker.Interval)
@@ -327,6 +328,7 @@ func TestBuilder_Build_CustomValues(t *testing.T) {
 	// Tracker worker tuning override
 	env["TRACKER_MAPPING_INTERVAL"] = "20"
 	env["TRACKER_MAPPING_MAX_CONCURRENCY"] = "5"
+	env["TRACKER_MAPPING_STALE_AFTER"] = "1200"
 	env["TRACKER_MAPPING_AGENT_TIMEOUT"] = "30"
 	env["TRACKER_MAPPING_AGENT_MAX_TURNS"] = "6"
 	env["COMMON_PATTERN_ENRICHMENT_INTERVAL"] = "15"
@@ -428,6 +430,7 @@ func TestBuilder_Build_CustomValues(t *testing.T) {
 	// Tracker worker tuning — overrides
 	assert.Equal(t, 20, cfg.Probod.TrackerMappingWorker.Interval)
 	assert.Equal(t, 5, cfg.Probod.TrackerMappingWorker.MaxConcurrency)
+	assert.Equal(t, 1200, cfg.Probod.TrackerMappingWorker.StaleAfter)
 	assert.Equal(t, 30, cfg.Probod.TrackerMappingWorker.AgentTimeout)
 	assert.Equal(t, 6, cfg.Probod.TrackerMappingWorker.AgentMaxTurns)
 	assert.Equal(t, 15, cfg.Probod.CommonPatternEnrichmentWorker.Interval)

pkg/cookiebanner/tracker_mapping_worker.go

@@ -31,6 +31,12 @@ import (
 	"go.probo.inc/probo/pkg/uri"
 )
 
+// defaultMappingStaleAfter is the fallback idle window after which a
+// claimed-but-unfinished tracker pattern mapping is re-armed. It is
+// generous relative to a single Process run (deterministic SQL plus up
+// to two bounded agent runs) so an in-flight mapping is never recycled.
+const defaultMappingStaleAfter = 10 * time.Minute
+
 type trackerMappingHandler struct {
 	pg                    *pg.Client
 	logger                *log.Logger
@@ -38,6 +44,7 @@ type trackerMappingHandler struct {
 	disambiguationAgent   *agent.Agent
 	agentTimeout          time.Duration
 	disambiguationTimeout time.Duration
+	staleAfter            time.Duration
 }
 
 func NewTrackerMappingWorker(
@@ -45,6 +52,7 @@ func NewTrackerMappingWorker(
 	logger *log.Logger,
 	mappingCfg TrackerAgentsConfig,
 	disambiguationCfg thirdparty.DisambiguationConfig,
+	staleAfter time.Duration,
 	opts ...worker.Option,
 ) *worker.Worker[coredata.TrackerPattern] {
 	agentTimeout := mappingCfg.AgentTimeout
@@ -52,11 +60,16 @@ func NewTrackerMappingWorker(
 		agentTimeout = defaultAgentTimeout
 	}
 
+	if staleAfter <= 0 {
+		staleAfter = defaultMappingStaleAfter
+	}
+
 	h := &trackerMappingHandler{
 		pg:                    pgClient,
 		logger:                logger,
 		agentTimeout:          agentTimeout,
 		disambiguationTimeout: disambiguationCfg.Timeout,
+		staleAfter:            staleAfter,
 	}
 
 	if mappingCfg.LLMClient != nil {
@@ -98,6 +111,24 @@ func (h *trackerMappingHandler) Claim(ctx context.Context) (coredata.TrackerPatt
 	return tp, nil
 }
 
+// RecoverStale re-arms tracker patterns whose mapping was claimed but
+// never finished. Claim clears mapping_requested_at up front, so a crash
+// or hard failure between phases would otherwise strand the pattern
+// unmapped with nothing to re-trigger it. ResetStaleMappings re-queues
+// those rows once they have been idle past staleAfter.
+func (h *trackerMappingHandler) RecoverStale(ctx context.Context) error {
+	return h.pg.WithConn(
+		ctx,
+		func(ctx context.Context, conn pg.Querier) error {
+			if err := coredata.ResetStaleMappings(ctx, conn, h.staleAfter); err != nil {
+				return fmt.Errorf("cannot reset stale tracker pattern mappings: %w", err)
+			}
+
+			return nil
+		},
+	)
+}
+
 // catalogMatch is the result of a single catalog signal. commonPatternID
 // is the catalog row the signal resolved (or backfilled); commonThirdPartyID
 // is the catalog third party the signal discovered, when any; thirdPartyID

pkg/coredata/tracker_pattern.go

@@ -1196,13 +1196,19 @@ LIMIT 1;
 	return nil
 }
 
+// ClearMappingRequestedAt removes the row from the mapping queue. It
+// bumps updated_at so the stale-recovery clock starts at claim time,
+// keeping ResetStaleMappings from re-arming a row that is still being
+// processed.
 func (tp *TrackerPattern) ClearMappingRequestedAt(
 	ctx context.Context,
 	tx pg.Tx,
 ) error {
 	q := `
 UPDATE tracker_patterns
-SET mapping_requested_at = NULL
+SET
+    mapping_requested_at = NULL,
+    updated_at = NOW()
 WHERE id = @id
 `
 
@@ -1218,6 +1224,42 @@ WHERE id = @id
 	return nil
 }
 
+// ResetStaleMappings re-arms mapping_requested_at on rows whose mapping
+// was claimed but never completed (no common_tracker_pattern_id) and
+// have been idle longer than staleAfter, so a crashed or timed-out
+// mapping run is retried. A successful Process always assigns a catalog
+// row (the unmatched fallback in createUnmatchedPattern), so a missing
+// common_tracker_pattern_id on a dequeued row marks an interrupted run.
+//
+// Like the claim query, this sweep is intentionally cross-tenant: the
+// mapping worker is a system worker that drains the queue regardless of
+// tenant.
+func ResetStaleMappings(
+	ctx context.Context,
+	conn pg.Querier,
+	staleAfter time.Duration,
+) error {
+	q := `
+UPDATE tracker_patterns
+SET
+    mapping_requested_at = NOW(),
+    updated_at = NOW()
+WHERE
+    mapping_requested_at IS NULL
+    AND common_tracker_pattern_id IS NULL
+    AND updated_at < @stale_before
+`
+
+	args := pgx.StrictNamedArgs{"stale_before": time.Now().Add(-staleAfter)}
+
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot reset stale tracker pattern mappings: %w", err)
+	}
+
+	return nil
+}
+
 func (tp *TrackerPattern) SetMappingRequested(
 	ctx context.Context,
 	tx pg.Tx,

pkg/coredata/tracker_pattern_test.go

@@ -249,3 +249,78 @@ func TestTrackerPattern_Update_NotFoundForMissingRow(t *testing.T) {
 
 	assert.ErrorIs(t, err, coredata.ErrResourceNotFound)
 }
+
+// TestResetStaleMappings pins the mapping stale-recovery contract: a row
+// dequeued (mapping_requested_at IS NULL) but never finished (no
+// common_tracker_pattern_id) and idle past the window is re-armed, while
+// a recently claimed row (clock not yet elapsed) and a completed row
+// (catalog row assigned) are left untouched. Without this sweep a crash
+// between Process phases would strand the pattern unmapped forever.
+func TestResetStaleMappings(t *testing.T) {
+	t.Parallel()
+
+	client := newTestPgClient(t)
+	ctx := context.Background()
+	fx := seedTrackerPatternFixture(t, ctx, client)
+
+	now := time.Now().UTC().Truncate(time.Microsecond)
+	old := now.Add(-time.Hour)
+	maxAge := 3600
+	source := coredata.CookieSourceScript
+
+	newPattern := func(pattern string, updatedAt time.Time, commonID *gid.GID) *coredata.TrackerPattern {
+		tp := &coredata.TrackerPattern{
+			ID:                     gid.New(fx.scope.GetTenantID(), coredata.TrackerPatternEntityType),
+			OrganizationID:         fx.organizationID,
+			CookieBannerID:         fx.cookieBannerID,
+			CookieCategoryID:       fx.cookieCategoryID,
+			CommonTrackerPatternID: commonID,
+			TrackerType:            coredata.TrackerTypeCookie,
+			Pattern:                pattern,
+			MatchType:              coredata.TrackerPatternMatchTypeExact,
+			DisplayName:            pattern,
+			MaxAgeSeconds:          &maxAge,
+			Source:                 &source,
+			CreatedAt:              old,
+			UpdatedAt:              updatedAt,
+		}
+
+		require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
+			return tp.Insert(ctx, tx, fx.scope)
+		}))
+
+		return tp
+	}
+
+	commonPattern := coredata.CommonTrackerPattern{
+		ID:          gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType),
+		TrackerType: coredata.TrackerTypeCookie,
+		Pattern:     "mapped_catalog_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String(),
+		MatchType:   coredata.TrackerPatternMatchTypeExact,
+		Confidence:  0.5,
+		CreatedAt:   old,
+		UpdatedAt:   old,
+	}
+	insertCommonTrackerPattern(t, ctx, client, commonPattern)
+
+	stale := newPattern("stale_unfinished", old, nil)
+	fresh := newPattern("fresh_unfinished", now, nil)
+	completed := newPattern("completed_mapping", old, &commonPattern.ID)
+
+	require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
+		return coredata.ResetStaleMappings(ctx, conn, 10*time.Minute)
+	}))
+
+	load := func(id gid.GID) coredata.TrackerPattern {
+		var reloaded coredata.TrackerPattern
+		require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
+			return reloaded.LoadByID(ctx, conn, fx.scope, id)
+		}))
+
+		return reloaded
+	}
+
+	assert.NotNil(t, load(stale.ID).MappingRequestedAt, "claimed-but-unfinished idle row must be re-armed")
+	assert.Nil(t, load(fresh.ID).MappingRequestedAt, "recently claimed row must not be re-armed before the window elapses")
+	assert.Nil(t, load(completed.ID).MappingRequestedAt, "completed mapping (catalog row assigned) must never be re-armed")
+}

pkg/probod/probod.go

@@ -730,6 +730,7 @@ func (impl *Implm) Run(
 		l,
 		trackerAgentsCfg,
 		thirdPartyDisambiguationCfg,
+		time.Duration(impl.cfg.TrackerMappingWorker.StaleAfter)*time.Second,
 		worker.WithInterval(time.Duration(impl.cfg.TrackerMappingWorker.Interval)*time.Second),
 		worker.WithMaxConcurrency(impl.cfg.TrackerMappingWorker.MaxConcurrency),
 	)

pkg/probodconfig/llm_config.go

@@ -48,6 +48,7 @@ type (
 	TrackerMappingWorkerConfig struct {
 		Interval       int `json:"interval"` // seconds between polls
 		MaxConcurrency int `json:"max-concurrency"`
+		StaleAfter     int `json:"stale-after"`   // seconds before a claim is recycled
 		AgentTimeout   int `json:"agent-timeout"` // seconds, single agent run
 		AgentMaxTurns  int `json:"agent-max-turns"`
 	}