From 952c427d2a08310e2e95e53ca72b77800fbc4e37 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=89mile=20R=C3=A9?= Date: Mon, 1 Jun 2026 11:43:53 +0200 Subject: [PATCH] Add stale recovery to tracker mapping worker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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é --- .../charts/probo/templates/deployment.yaml | 4 + .../probo/values-production.yaml.example | 4 +- contrib/helm/charts/probo/values.yaml | 8 +- pkg/bootstrap/builder.go | 1 + pkg/bootstrap/builder_test.go | 3 + pkg/cookiebanner/tracker_mapping_worker.go | 31 ++++++++ pkg/coredata/tracker_pattern.go | 44 ++++++++++- pkg/coredata/tracker_pattern_test.go | 75 +++++++++++++++++++ pkg/probod/probod.go | 1 + pkg/probodconfig/llm_config.go | 1 + 10 files changed, 167 insertions(+), 5 deletions(-) diff --git a/contrib/helm/charts/probo/templates/deployment.yaml b/contrib/helm/charts/probo/templates/deployment.yaml index d62d64e1d..03ec276f5 100644 --- a/contrib/helm/charts/probo/templates/deployment.yaml +++ b/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 }} diff --git a/contrib/helm/charts/probo/values-production.yaml.example b/contrib/helm/charts/probo/values-production.yaml.example index 4d5622d4b..8b0d771c5 100644 --- a/contrib/helm/charts/probo/values-production.yaml.example +++ b/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 diff --git a/contrib/helm/charts/probo/values.yaml b/contrib/helm/charts/probo/values.yaml index 6f23205d4..a1b37ff95 100644 --- a/contrib/helm/charts/probo/values.yaml +++ b/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 diff --git a/pkg/bootstrap/builder.go b/pkg/bootstrap/builder.go index 70e03ab5d..01a4c6792 100644 --- a/pkg/bootstrap/builder.go +++ b/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), }, diff --git a/pkg/bootstrap/builder_test.go b/pkg/bootstrap/builder_test.go index 522e51fff..9eb03c50d 100644 --- a/pkg/bootstrap/builder_test.go +++ b/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) diff --git a/pkg/cookiebanner/tracker_mapping_worker.go b/pkg/cookiebanner/tracker_mapping_worker.go index 7a63f15ad..9c4f3936b 100644 --- a/pkg/cookiebanner/tracker_mapping_worker.go +++ b/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 diff --git a/pkg/coredata/tracker_pattern.go b/pkg/coredata/tracker_pattern.go index f806c1797..d18e57671 100644 --- a/pkg/coredata/tracker_pattern.go +++ b/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, diff --git a/pkg/coredata/tracker_pattern_test.go b/pkg/coredata/tracker_pattern_test.go index b22f51db8..88a2199c2 100644 --- a/pkg/coredata/tracker_pattern_test.go +++ b/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") +} diff --git a/pkg/probod/probod.go b/pkg/probod/probod.go index aa6918dd8..a18ff1a96 100644 --- a/pkg/probod/probod.go +++ b/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), ) diff --git a/pkg/probodconfig/llm_config.go b/pkg/probodconfig/llm_config.go index 70021d5af..b745bde31 100644 --- a/pkg/probodconfig/llm_config.go +++ b/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"` }