From ef6cead482c3b3b58eae5abda038964e0b035839 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=89mile=20R=C3=A9?= Date: Mon, 15 Jun 2026 16:25:53 +0200 Subject: [PATCH] Unify enrichment tracking and outcome-based status MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make common_tracker_patterns and common_third_parties share one enrichment-tracking model and fix the misleading proboctl status. Both tables now carry the enrichment JSONB provenance payload, an enrichment_attempts counter, and a last_enrichment_attempt_at clock. On common_tracker_patterns the enriched_at done-flag is renamed to last_enrichment_attempt_at and stamped at claim time, so it is truthful to "attempt" rather than "success". A row is considered to have been through the workflow when it carries an enrichment payload, not when a timestamp is set, which lets stale recovery key off the payload being absent with budget remaining, exactly like common_third_parties. The claim path reads the attempt counter and timestamp back via RETURNING so the in-memory receiver matches the database clock instead of a separate app-side time.Now. The enricher builds a per-field provenance payload (description and third-party outcomes plus the mapping attribution) and persists it via UpdateEnrichment, named to mirror the common-third-party sibling. The common pattern enrichment worker gains a max-attempts ceiling so a permanently failing row stops looping. proboctl now shows "enriched" only when every field the last run recorded an outcome for resolved a value, otherwise "partial (X/Y)", replacing the misleading "enriched (no description)" label. Signed-off-by: Émile Ré --- pkg/cookiebanner/common_pattern_enricher.go | 24 ++- .../common_pattern_enrichment_metadata.go | 163 ++++++++++++++ .../common_pattern_enrichment_worker.go | 34 ++- pkg/coredata/common_third_party.go | 33 ++- pkg/coredata/common_third_party_filter.go | 7 +- pkg/coredata/common_tracker_pattern.go | 201 ++++++++++++------ pkg/coredata/common_tracker_pattern_filter.go | 10 +- .../common_tracker_pattern_order_field.go | 18 +- pkg/coredata/common_tracker_pattern_test.go | 170 +++++++++++---- pkg/coredata/migrations/20260615T161615Z.sql | 37 ++++ .../commonthirdparty/commonthirdparty.go | 50 ++++- pkg/proboctl/commonthirdparty/show.go | 6 +- .../commontrackerpattern.go | 53 ++++- pkg/proboctl/commontrackerpattern/list.go | 8 +- .../commontrackerpattern/set_description.go | 39 +++- pkg/proboctl/commontrackerpattern/show.go | 6 +- pkg/probod/probod.go | 1 + 17 files changed, 705 insertions(+), 155 deletions(-) create mode 100644 pkg/cookiebanner/common_pattern_enrichment_metadata.go create mode 100644 pkg/coredata/migrations/20260615T161615Z.sql diff --git a/pkg/cookiebanner/common_pattern_enricher.go b/pkg/cookiebanner/common_pattern_enricher.go index 9d1ba5d33..3e755d46c 100644 --- a/pkg/cookiebanner/common_pattern_enricher.go +++ b/pkg/cookiebanner/common_pattern_enricher.go @@ -16,6 +16,7 @@ package cookiebanner import ( "context" + "encoding/json" "fmt" "strings" "time" @@ -140,6 +141,8 @@ func (e *CommonPatternEnricher) EnrichPattern(ctx context.Context, cp coredata.C return fmt.Errorf("cannot research tracker description: %w", err) } + alreadyLinked := cp.CommonThirdPartyID != nil + return e.pg.WithTx( ctx, func(ctx context.Context, tx pg.Tx) error { @@ -156,7 +159,23 @@ func (e *CommonPatternEnricher) EnrichPattern(ctx context.Context, cp coredata.C } } - if err := cp.SetEnriched(ctx, tx, description, thirdPartyID); err != nil { + linked := alreadyLinked || thirdPartyID != nil + + meta := buildCommonPatternEnrichmentMetadata( + e.enrichmentCfg.Model, + description, + attribution, + alreadyLinked, + linked, + time.Now(), + ) + + payload, err := json.Marshal(meta) + if err != nil { + return fmt.Errorf("cannot marshal common tracker pattern enrichment metadata: %w", err) + } + + if err := cp.UpdateEnrichment(ctx, tx, description, thirdPartyID, payload); err != nil { return fmt.Errorf("cannot set common tracker pattern enriched: %w", err) } @@ -177,7 +196,8 @@ func (e *CommonPatternEnricher) EnrichPattern(ctx context.Context, cp coredata.C log.String("common_tracker_pattern_id", cp.ID.String()), log.String("pattern", cp.Pattern), log.Bool("described", description != ""), - log.Bool("third_party_linked", thirdPartyID != nil), + log.Bool("third_party_linked", linked), + log.Int("enrichment_attempts", cp.EnrichmentAttempts), log.Int64("backfilled_tracker_patterns", backfilled), ) diff --git a/pkg/cookiebanner/common_pattern_enrichment_metadata.go b/pkg/cookiebanner/common_pattern_enrichment_metadata.go new file mode 100644 index 000000000..7b756a3db --- /dev/null +++ b/pkg/cookiebanner/common_pattern_enrichment_metadata.go @@ -0,0 +1,163 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +package cookiebanner + +import ( + "strings" + "time" +) + +// Per-field outcomes recorded in the common tracker pattern enrichment +// payload. They mirror the common-third-party enrichment provenance so the +// proboctl display can compute the same "X/Y resolved" completeness across +// both catalogs. +const ( + commonPatternFieldStatusFound = "found" + commonPatternFieldStatusNotFound = "not_found" + commonPatternFieldStatusExternal = "exists_external" + + // Run-level status recorded at the top of the enrichment payload. + commonPatternStatusDone = "done" + commonPatternStatusPartial = "partial" + commonPatternStatusNoResult = "no_result" + + // Field keys recorded in the payload (the enrichment targets). + commonPatternFieldDescription = "description" + commonPatternFieldThirdParty = "third_party" +) + +type ( + // CommonPatternFieldMeta is the per-field provenance recorded in the + // common_tracker_patterns.enrichment JSON column. + CommonPatternFieldMeta struct { + Status string `json:"status"` + UpdatedAt time.Time `json:"updated_at"` + } + + // CommonPatternAttributionMeta records the mapping-agent result the + // enricher used to attribute (and possibly link) a vendor, so the + // "mapping" decision is auditable from the enrichment payload. + CommonPatternAttributionMeta struct { + ThirdPartyName string `json:"third_party_name,omitempty"` + Category string `json:"category,omitempty"` + Confidence float64 `json:"confidence"` + Linked bool `json:"linked"` + } + + // CommonPatternEnrichmentMetadata is the full payload stored in the + // enrichment JSON column: run-level bookkeeping, per-field provenance + // keyed by the enrichment target, and the vendor attribution. + CommonPatternEnrichmentMetadata struct { + Model string `json:"model,omitempty"` + AttemptedAt time.Time `json:"attempted_at"` + Status string `json:"status"` + Error string `json:"error,omitempty"` + Fields map[string]CommonPatternFieldMeta `json:"fields"` + Attribution *CommonPatternAttributionMeta `json:"attribution,omitempty"` + } +) + +// buildCommonPatternEnrichmentMetadata assembles the per-run provenance for +// one common tracker pattern. It records an outcome for both enrichment +// targets — the description and the third-party attribution — so the +// display can report "X/Y resolved". alreadyLinked marks a vendor the +// mapping pipeline resolved before this run (recorded as exists_external), +// versus a vendor this enrichment run resolved (found). attribution, when +// non-nil, carries the mapping-agent decision this run made. +func buildCommonPatternEnrichmentMetadata( + model string, + description string, + attribution *TrackerMappingAgentResult, + alreadyLinked bool, + linked bool, + now time.Time, +) CommonPatternEnrichmentMetadata { + fields := make(map[string]CommonPatternFieldMeta, 2) + + descStatus := commonPatternFieldStatusNotFound + if strings.TrimSpace(description) != "" { + descStatus = commonPatternFieldStatusFound + } + + fields[commonPatternFieldDescription] = CommonPatternFieldMeta{ + Status: descStatus, + UpdatedAt: now, + } + + thirdPartyStatus := commonPatternFieldStatusNotFound + switch { + case alreadyLinked: + thirdPartyStatus = commonPatternFieldStatusExternal + case linked: + thirdPartyStatus = commonPatternFieldStatusFound + } + + fields[commonPatternFieldThirdParty] = CommonPatternFieldMeta{ + Status: thirdPartyStatus, + UpdatedAt: now, + } + + meta := CommonPatternEnrichmentMetadata{ + Model: model, + AttemptedAt: now, + Status: commonPatternRunStatus(fields), + Fields: fields, + } + + if attribution != nil { + meta.Attribution = &CommonPatternAttributionMeta{ + ThirdPartyName: attribution.ThirdPartyName, + Category: string(attribution.Category), + Confidence: attribution.ThirdPartyConfidence, + Linked: linked, + } + } + + return meta +} + +// commonPatternRunStatus classifies the run from its per-field outcomes: +// done when every field resolved a value, no_result when none did, partial +// otherwise. +func commonPatternRunStatus(fields map[string]CommonPatternFieldMeta) string { + var resolved int + + for _, f := range fields { + if commonPatternFieldResolved(f.Status) { + resolved++ + } + } + + switch { + case resolved == 0: + return commonPatternStatusNoResult + case resolved == len(fields): + return commonPatternStatusDone + default: + return commonPatternStatusPartial + } +} + +// commonPatternFieldResolved reports whether a field status carries a +// resolved value (found or already present externally) versus an absent +// one. +func commonPatternFieldResolved(status string) bool { + switch status { + case commonPatternFieldStatusFound, commonPatternFieldStatusExternal: + return true + default: + return false + } +} diff --git a/pkg/cookiebanner/common_pattern_enrichment_worker.go b/pkg/cookiebanner/common_pattern_enrichment_worker.go index d65b15d88..3df8847b0 100644 --- a/pkg/cookiebanner/common_pattern_enrichment_worker.go +++ b/pkg/cookiebanner/common_pattern_enrichment_worker.go @@ -26,17 +26,25 @@ import ( "go.probo.inc/probo/pkg/coredata" ) -const defaultEnrichmentStaleAfter = 10 * time.Minute +const ( + defaultEnrichmentStaleAfter = 10 * time.Minute + + // defaultEnrichmentMaxAttempts caps how many times a row is retried + // before stale recovery leaves it alone, so a permanently failing row + // does not loop forever. + defaultEnrichmentMaxAttempts = 3 +) // commonPatternEnrichmentHandler is the queue poller for common tracker // pattern enrichment. It owns only the claim/dequeue and stale-recovery // mechanics; the enrichment work itself lives in CommonPatternEnricher so // it can also run synchronously from operator tooling. type commonPatternEnrichmentHandler struct { - pg *pg.Client - logger *log.Logger - enricher *CommonPatternEnricher - staleAfter time.Duration + pg *pg.Client + logger *log.Logger + enricher *CommonPatternEnricher + staleAfter time.Duration + maxAttempts int } // NewCommonPatternEnrichmentWorker builds the worker that fills @@ -52,17 +60,23 @@ func NewCommonPatternEnrichmentWorker( enrichmentCfg TrackerEnrichmentAgentConfig, mappingCfg TrackerMappingAgentConfig, staleAfter time.Duration, + maxAttempts int, opts ...worker.Option, ) *worker.Worker[coredata.CommonTrackerPattern] { if staleAfter <= 0 { staleAfter = defaultEnrichmentStaleAfter } + if maxAttempts <= 0 { + maxAttempts = defaultEnrichmentMaxAttempts + } + h := &commonPatternEnrichmentHandler{ - pg: pgClient, - logger: logger, - enricher: NewCommonPatternEnricher(pgClient, logger, enrichmentCfg, mappingCfg), - staleAfter: staleAfter, + pg: pgClient, + logger: logger, + enricher: NewCommonPatternEnricher(pgClient, logger, enrichmentCfg, mappingCfg), + staleAfter: staleAfter, + maxAttempts: maxAttempts, } return worker.New( @@ -108,7 +122,7 @@ func (h *commonPatternEnrichmentHandler) RecoverStale(ctx context.Context) error return h.pg.WithConn( ctx, func(ctx context.Context, conn pg.Querier) error { - if err := coredata.ResetStaleEnrichments(ctx, conn, h.staleAfter); err != nil { + if err := coredata.ResetStaleEnrichments(ctx, conn, h.staleAfter, h.maxAttempts); err != nil { return fmt.Errorf("cannot reset stale common tracker pattern enrichments: %w", err) } diff --git a/pkg/coredata/common_third_party.go b/pkg/coredata/common_third_party.go index dd358c15d..b58e947e4 100644 --- a/pkg/coredata/common_third_party.go +++ b/pkg/coredata/common_third_party.go @@ -53,6 +53,7 @@ type ( EnrichmentRequestedAt *time.Time `db:"enrichment_requested_at"` Enrichment json.RawMessage `db:"enrichment"` EnrichmentAttempts int `db:"enrichment_attempts"` + LastEnrichmentAttemptAt *time.Time `db:"last_enrichment_attempt_at"` CreatedAt time.Time `db:"created_at"` UpdatedAt time.Time `db:"updated_at"` } @@ -131,6 +132,7 @@ SELECT enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -191,6 +193,7 @@ SELECT enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -251,6 +254,7 @@ SELECT enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -310,6 +314,7 @@ INSERT INTO common_third_parties ( enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at ) VALUES ( @@ -335,6 +340,7 @@ INSERT INTO common_third_parties ( @enrichment_requested_at, @enrichment, @enrichment_attempts, + @last_enrichment_attempt_at, @created_at, @updated_at ) @@ -363,6 +369,7 @@ INSERT INTO common_third_parties ( "enrichment_requested_at": t.EnrichmentRequestedAt, "enrichment": t.Enrichment, "enrichment_attempts": t.EnrichmentAttempts, + "last_enrichment_attempt_at": t.LastEnrichmentAttemptAt, "created_at": t.CreatedAt, "updated_at": t.UpdatedAt, } @@ -406,6 +413,7 @@ INSERT INTO common_third_parties ( enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at ) VALUES ( @@ -431,6 +439,7 @@ INSERT INTO common_third_parties ( @enrichment_requested_at, @enrichment, @enrichment_attempts, + @last_enrichment_attempt_at, @created_at, @updated_at ) @@ -476,6 +485,7 @@ RETURNING enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at ` @@ -505,6 +515,7 @@ RETURNING "enrichment_requested_at": t.EnrichmentRequestedAt, "enrichment": t.Enrichment, "enrichment_attempts": t.EnrichmentAttempts, + "last_enrichment_attempt_at": t.LastEnrichmentAttemptAt, "created_at": t.CreatedAt, "updated_at": t.UpdatedAt, } @@ -571,6 +582,7 @@ SELECT enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -625,6 +637,7 @@ SELECT enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -769,6 +782,7 @@ SELECT enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -862,6 +876,7 @@ SELECT enrichment_requested_at, enrichment, enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -895,8 +910,8 @@ LIMIT 1; } // ClearEnrichmentRequestedAt removes the row from the enrichment queue -// and bumps the attempt counter. It bumps updated_at so the -// stale-recovery clock starts at claim time, keeping +// and bumps the attempt counter. It stamps last_enrichment_attempt_at so +// the stale-recovery clock starts at claim time, keeping // ResetStaleCommonThirdPartyEnrichments from re-arming a row that is // still being processed. The attempt counter is incremented up front so // a crash between claim and persist still counts against the retry @@ -910,19 +925,27 @@ UPDATE common_third_parties SET enrichment_requested_at = NULL, enrichment_attempts = enrichment_attempts + 1, + last_enrichment_attempt_at = NOW(), updated_at = NOW() WHERE id = @id +RETURNING enrichment_attempts, last_enrichment_attempt_at ` args := pgx.StrictNamedArgs{"id": t.ID} - _, err := tx.Exec(ctx, q, args) + var ( + attempts int + lastAttempt *time.Time + ) + + err := tx.QueryRow(ctx, q, args).Scan(&attempts, &lastAttempt) if err != nil { return fmt.Errorf("cannot clear enrichment requested at: %w", err) } t.EnrichmentRequestedAt = nil - t.EnrichmentAttempts++ + t.EnrichmentAttempts = attempts + t.LastEnrichmentAttemptAt = lastAttempt return nil } @@ -1024,7 +1047,7 @@ WHERE AND enrichment IS NULL AND enrichment_attempts > 0 AND enrichment_attempts < @max_attempts - AND updated_at < @stale_before + AND last_enrichment_attempt_at < @stale_before ` args := pgx.StrictNamedArgs{ diff --git a/pkg/coredata/common_third_party_filter.go b/pkg/coredata/common_third_party_filter.go index c5f637a20..ae0e382e3 100644 --- a/pkg/coredata/common_third_party_filter.go +++ b/pkg/coredata/common_third_party_filter.go @@ -23,10 +23,9 @@ import ( // CommonThirdPartyEnrichmentState is a synthetic filter over the // enrichment_requested_at / enrichment columns. It is not a stored -// column; it classifies a row's position in the enrichment lifecycle. -// Unlike common tracker patterns, a common third party has no -// enriched_at column: a row is "enriched" once it carries an enrichment -// payload (Process always writes one, even on a no-result run). +// column; it classifies a row's position in the enrichment lifecycle. A +// row is "enriched" once it carries an enrichment payload (Process always +// writes one, even on a no-result run). type CommonThirdPartyEnrichmentState string const ( diff --git a/pkg/coredata/common_tracker_pattern.go b/pkg/coredata/common_tracker_pattern.go index da894b657..25ffc5c2e 100644 --- a/pkg/coredata/common_tracker_pattern.go +++ b/pkg/coredata/common_tracker_pattern.go @@ -16,6 +16,7 @@ package coredata import ( "context" + "encoding/json" "errors" "fmt" "maps" @@ -29,18 +30,20 @@ import ( type ( CommonTrackerPattern struct { - ID gid.GID `db:"id"` - CommonThirdPartyID *gid.GID `db:"common_third_party_id"` - TrackerType TrackerType `db:"tracker_type"` - Pattern string `db:"pattern"` - MatchType TrackerPatternMatchType `db:"match_type"` - Description string `db:"description"` - MaxAgeSeconds *int `db:"max_age_seconds"` - Confidence float32 `db:"confidence"` - EnrichmentRequestedAt *time.Time `db:"enrichment_requested_at"` - EnrichedAt *time.Time `db:"enriched_at"` - CreatedAt time.Time `db:"created_at"` - UpdatedAt time.Time `db:"updated_at"` + ID gid.GID `db:"id"` + CommonThirdPartyID *gid.GID `db:"common_third_party_id"` + TrackerType TrackerType `db:"tracker_type"` + Pattern string `db:"pattern"` + MatchType TrackerPatternMatchType `db:"match_type"` + Description string `db:"description"` + MaxAgeSeconds *int `db:"max_age_seconds"` + Confidence float32 `db:"confidence"` + EnrichmentRequestedAt *time.Time `db:"enrichment_requested_at"` + Enrichment json.RawMessage `db:"enrichment"` + EnrichmentAttempts int `db:"enrichment_attempts"` + LastEnrichmentAttemptAt *time.Time `db:"last_enrichment_attempt_at"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` } CommonTrackerPatterns []*CommonTrackerPattern @@ -62,7 +65,9 @@ SELECT max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -111,7 +116,9 @@ SELECT max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -163,7 +170,9 @@ INSERT INTO common_tracker_patterns ( max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at ) VALUES ( @@ -176,25 +185,29 @@ INSERT INTO common_tracker_patterns ( @max_age_seconds, @confidence, @enrichment_requested_at, - @enriched_at, + @enrichment, + @enrichment_attempts, + @last_enrichment_attempt_at, @created_at, @updated_at ) ` args := pgx.StrictNamedArgs{ - "id": p.ID, - "common_third_party_id": p.CommonThirdPartyID, - "tracker_type": p.TrackerType, - "pattern": p.Pattern, - "match_type": p.MatchType, - "description": p.Description, - "max_age_seconds": p.MaxAgeSeconds, - "confidence": p.Confidence, - "enrichment_requested_at": p.EnrichmentRequestedAt, - "enriched_at": p.EnrichedAt, - "created_at": p.CreatedAt, - "updated_at": p.UpdatedAt, + "id": p.ID, + "common_third_party_id": p.CommonThirdPartyID, + "tracker_type": p.TrackerType, + "pattern": p.Pattern, + "match_type": p.MatchType, + "description": p.Description, + "max_age_seconds": p.MaxAgeSeconds, + "confidence": p.Confidence, + "enrichment_requested_at": p.EnrichmentRequestedAt, + "enrichment": p.Enrichment, + "enrichment_attempts": p.EnrichmentAttempts, + "last_enrichment_attempt_at": p.LastEnrichmentAttemptAt, + "created_at": p.CreatedAt, + "updated_at": p.UpdatedAt, } _, err := conn.Exec(ctx, q, args) @@ -226,7 +239,9 @@ INSERT INTO common_tracker_patterns ( max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at ) VALUES ( @@ -240,6 +255,8 @@ INSERT INTO common_tracker_patterns ( @confidence, CASE WHEN @description = '' THEN NOW() ELSE NULL END, NULL, + 0, + NULL, @created_at, @updated_at ) @@ -255,8 +272,8 @@ SET -- A blank, unlinked catalog row that now gains a third party is -- re-queued for enrichment: the enrichment agent leaves descriptions -- blank when it cannot substantiate a purpose, and knowing the vendor - -- gives it a second, better-informed attempt. enriched_at is cleared - -- so the row is no longer terminal. + -- gives it a second, better-informed attempt. The attempt counter is + -- reset so the re-armed row gets a fresh retry budget. enrichment_requested_at = CASE WHEN common_tracker_patterns.description = '' AND common_tracker_patterns.common_third_party_id IS NULL @@ -264,12 +281,12 @@ SET THEN NOW() ELSE common_tracker_patterns.enrichment_requested_at END, - enriched_at = CASE + enrichment_attempts = CASE WHEN common_tracker_patterns.description = '' AND common_tracker_patterns.common_third_party_id IS NULL AND EXCLUDED.common_third_party_id IS NOT NULL - THEN NULL - ELSE common_tracker_patterns.enriched_at + THEN 0 + ELSE common_tracker_patterns.enrichment_attempts END, updated_at = EXCLUDED.updated_at RETURNING @@ -282,7 +299,9 @@ RETURNING max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at ` @@ -352,7 +371,9 @@ SELECT max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -469,7 +490,9 @@ SELECT max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -516,7 +539,9 @@ SELECT max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -548,8 +573,11 @@ LIMIT 1; return nil } -// ClearEnrichmentRequestedAt removes the row from the enrichment queue. It -// bumps updated_at so the stale-recovery clock starts at claim time. +// ClearEnrichmentRequestedAt removes the row from the enrichment queue and +// records the attempt: it bumps the attempt counter and stamps +// last_enrichment_attempt_at, which is the stale-recovery clock. The +// attempt counter is incremented up front so a crash between claim and +// persist still counts against the retry budget. It bumps updated_at too. func (p *CommonTrackerPattern) ClearEnrichmentRequestedAt( ctx context.Context, tx pg.Tx, @@ -558,42 +586,57 @@ func (p *CommonTrackerPattern) ClearEnrichmentRequestedAt( UPDATE common_tracker_patterns SET enrichment_requested_at = NULL, + enrichment_attempts = enrichment_attempts + 1, + last_enrichment_attempt_at = NOW(), updated_at = NOW() WHERE id = @id +RETURNING enrichment_attempts, last_enrichment_attempt_at ` args := pgx.StrictNamedArgs{"id": p.ID} - _, err := tx.Exec(ctx, q, args) + var ( + attempts int + lastAttempt *time.Time + ) + + err := tx.QueryRow(ctx, q, args).Scan(&attempts, &lastAttempt) if err != nil { return fmt.Errorf("cannot clear enrichment requested at: %w", err) } p.EnrichmentRequestedAt = nil + p.EnrichmentAttempts = attempts + p.LastEnrichmentAttemptAt = lastAttempt return nil } -// SetEnriched records the researched description and marks the row -// enriched so the stale-recovery loop never re-queues it. An empty -// description is allowed: the enrichment agent leaves it blank when it -// cannot substantiate a purpose, and a later third-party link re-arms -// enrichment for a second attempt. When thirdPartyID is non-nil it links -// the row to that third party, but only when none is set yet -// (COALESCE) — the enrichment worker links, it never overrides an -// attribution the mapping pipeline already resolved. -func (p *CommonTrackerPattern) SetEnriched( +// UpdateEnrichment records the researched description and the per-run +// enrichment provenance payload (named to mirror +// CommonThirdParty.UpdateEnrichment, the sibling persist step). The +// payload presence is what marks a row as having been through the +// workflow, so the stale-recovery loop never re-queues it +// (last_enrichment_attempt_at, the attempt clock, is stamped separately at +// claim time). An empty description is allowed: the enrichment agent +// leaves it blank when it cannot substantiate a purpose, and a later +// third-party link re-arms enrichment for a second attempt. When +// thirdPartyID is non-nil it links the row to that third party, but only +// when none is set yet (COALESCE) — the enrichment worker links, it never +// overrides an attribution the mapping pipeline already resolved. +func (p *CommonTrackerPattern) UpdateEnrichment( ctx context.Context, tx pg.Tx, description string, thirdPartyID *gid.GID, + enrichment json.RawMessage, ) error { q := ` UPDATE common_tracker_patterns SET description = @description, common_third_party_id = COALESCE(common_third_party_id, @third_party_id), - enriched_at = NOW(), + enrichment = @enrichment, enrichment_requested_at = NULL, updated_at = NOW() WHERE id = @id @@ -603,6 +646,7 @@ WHERE id = @id "id": p.ID, "description": description, "third_party_id": thirdPartyID, + "enrichment": enrichment, } result, err := tx.Exec(ctx, q, args) @@ -615,6 +659,8 @@ WHERE id = @id } p.Description = description + p.Enrichment = enrichment + p.EnrichmentRequestedAt = nil if p.CommonThirdPartyID == nil { p.CommonThirdPartyID = thirdPartyID @@ -624,13 +670,21 @@ WHERE id = @id } // ResetStaleEnrichments re-queues rows whose enrichment was claimed but -// never completed (no enriched_at, still description-less) and have been -// idle longer than staleAfter, so a crashed or timed-out enrichment is -// retried. +// never completed and have been idle longer than staleAfter, so a crashed +// or timed-out enrichment is retried. +// +// A claimed row has enrichment_attempts > 0 (the claim increments it) and +// a completed row carries a non-null enrichment payload (UpdateEnrichment +// always writes it, even on a blank-description run), so the sweep targets +// rows that were claimed but carry no payload yet. Curated rows that were +// never enqueued keep enrichment_attempts = 0 and are left untouched. The +// max-attempts ceiling stops permanently failing rows from looping +// forever. last_enrichment_attempt_at, stamped at claim, is the idle clock. func ResetStaleEnrichments( ctx context.Context, conn pg.Querier, staleAfter time.Duration, + maxAttempts int, ) error { q := ` UPDATE common_tracker_patterns @@ -639,12 +693,16 @@ SET updated_at = NOW() WHERE enrichment_requested_at IS NULL - AND enriched_at IS NULL - AND description = '' - AND updated_at < @stale_before + AND enrichment IS NULL + AND enrichment_attempts > 0 + AND enrichment_attempts < @max_attempts + AND last_enrichment_attempt_at < @stale_before ` - args := pgx.StrictNamedArgs{"stale_before": time.Now().Add(-staleAfter)} + args := pgx.StrictNamedArgs{ + "max_attempts": maxAttempts, + "stale_before": time.Now().Add(-staleAfter), + } _, err := conn.Exec(ctx, q, args) if err != nil { @@ -670,7 +728,9 @@ SELECT max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -706,12 +766,12 @@ func (p *CommonTrackerPattern) CursorKey(field CommonTrackerPatternOrderField) p return page.NewCursorKey(p.ID, p.CreatedAt) case CommonTrackerPatternOrderFieldUpdatedAt: return page.NewCursorKey(p.ID, p.UpdatedAt) - case CommonTrackerPatternOrderFieldEnrichedAt: - if p.EnrichedAt == nil { + case CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt: + if p.LastEnrichmentAttemptAt == nil { return page.NewCursorKey(p.ID, time.Time{}) } - return page.NewCursorKey(p.ID, *p.EnrichedAt) + return page.NewCursorKey(p.ID, *p.LastEnrichmentAttemptAt) } panic(fmt.Sprintf("unsupported order by: %s", field)) @@ -737,7 +797,9 @@ SELECT max_age_seconds, confidence, enrichment_requested_at, - enriched_at, + enrichment, + enrichment_attempts, + last_enrichment_attempt_at, created_at, updated_at FROM @@ -874,10 +936,14 @@ WHERE // RequestEnrichmentByIDs arms enrichment on the given common tracker // patterns by stamping enrichment_requested_at, which is the only column -// the enrichment worker claims on. Already-enriched rows are re-processed -// too: the worker overwrites enriched_at and the description when it runs. -// Returns the number of rows re-queued. This is the async fallback path; -// the synchronous enricher service is preferred. +// the enrichment worker claims on. It resets enrichment_attempts to 0 so +// the re-queued rows get a fresh retry budget: the claim path bumps the +// counter on every run, and without a reset a row near the max-attempts +// ceiling would not be re-armed by stale recovery if a re-run crashed. +// Already-enriched rows are re-processed too: the worker overwrites the +// description and enrichment payload when it runs. Returns the number of +// rows re-queued. This is the async fallback path; the synchronous +// enricher service is preferred. func (ps *CommonTrackerPatterns) RequestEnrichmentByIDs( ctx context.Context, tx pg.Tx, @@ -887,6 +953,7 @@ func (ps *CommonTrackerPatterns) RequestEnrichmentByIDs( UPDATE common_tracker_patterns SET enrichment_requested_at = NOW(), + enrichment_attempts = 0, updated_at = NOW() WHERE id = ANY(@ids) diff --git a/pkg/coredata/common_tracker_pattern_filter.go b/pkg/coredata/common_tracker_pattern_filter.go index 2f8632c03..9e2e1ff13 100644 --- a/pkg/coredata/common_tracker_pattern_filter.go +++ b/pkg/coredata/common_tracker_pattern_filter.go @@ -22,7 +22,7 @@ import ( ) // CommonTrackerPatternEnrichmentState is a synthetic filter over the -// enrichment_requested_at / enriched_at columns. It is not a stored +// enrichment_requested_at / enrichment columns. It is not a stored // column; it classifies a row's position in the enrichment lifecycle. type CommonTrackerPatternEnrichmentState string @@ -30,8 +30,8 @@ const ( // CommonTrackerPatternEnrichmentStateQueued: a row armed for the // enrichment worker (enrichment_requested_at IS NOT NULL). CommonTrackerPatternEnrichmentStateQueued CommonTrackerPatternEnrichmentState = "QUEUED" - // CommonTrackerPatternEnrichmentStateEnriched: a row whose - // enrichment has completed (enriched_at IS NOT NULL) and is not + // CommonTrackerPatternEnrichmentStateEnriched: a row that has been + // through the enrichment workflow (enrichment IS NOT NULL) and is not // re-queued. CommonTrackerPatternEnrichmentStateEnriched CommonTrackerPatternEnrichmentState = "ENRICHED" // CommonTrackerPatternEnrichmentStateUnenriched: a row never enriched @@ -183,9 +183,9 @@ func (f *CommonTrackerPatternFilter) SQLFragment() string { CASE WHEN @filter_state_queued::boolean THEN enrichment_requested_at IS NOT NULL WHEN @filter_state_enriched::boolean THEN - enrichment_requested_at IS NULL AND enriched_at IS NOT NULL + enrichment_requested_at IS NULL AND enrichment IS NOT NULL WHEN @filter_state_unenriched::boolean THEN - enrichment_requested_at IS NULL AND enriched_at IS NULL + enrichment_requested_at IS NULL AND enrichment IS NULL ELSE TRUE END )` diff --git a/pkg/coredata/common_tracker_pattern_order_field.go b/pkg/coredata/common_tracker_pattern_order_field.go index 7dec07a07..fc9a79d30 100644 --- a/pkg/coredata/common_tracker_pattern_order_field.go +++ b/pkg/coredata/common_tracker_pattern_order_field.go @@ -24,11 +24,11 @@ import ( type CommonTrackerPatternOrderField string const ( - CommonTrackerPatternOrderFieldPattern CommonTrackerPatternOrderField = "PATTERN" - CommonTrackerPatternOrderFieldConfidence CommonTrackerPatternOrderField = "CONFIDENCE" - CommonTrackerPatternOrderFieldCreatedAt CommonTrackerPatternOrderField = "CREATED_AT" - CommonTrackerPatternOrderFieldUpdatedAt CommonTrackerPatternOrderField = "UPDATED_AT" - CommonTrackerPatternOrderFieldEnrichedAt CommonTrackerPatternOrderField = "ENRICHED_AT" + CommonTrackerPatternOrderFieldPattern CommonTrackerPatternOrderField = "PATTERN" + CommonTrackerPatternOrderFieldConfidence CommonTrackerPatternOrderField = "CONFIDENCE" + CommonTrackerPatternOrderFieldCreatedAt CommonTrackerPatternOrderField = "CREATED_AT" + CommonTrackerPatternOrderFieldUpdatedAt CommonTrackerPatternOrderField = "UPDATED_AT" + CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt CommonTrackerPatternOrderField = "LAST_ENRICHMENT_ATTEMPT_AT" ) var ( @@ -44,7 +44,7 @@ func CommonTrackerPatternOrderFields() []CommonTrackerPatternOrderField { CommonTrackerPatternOrderFieldConfidence, CommonTrackerPatternOrderFieldCreatedAt, CommonTrackerPatternOrderFieldUpdatedAt, - CommonTrackerPatternOrderFieldEnrichedAt, + CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt, } } @@ -55,7 +55,7 @@ func (v CommonTrackerPatternOrderField) IsValid() bool { CommonTrackerPatternOrderFieldConfidence, CommonTrackerPatternOrderFieldCreatedAt, CommonTrackerPatternOrderFieldUpdatedAt, - CommonTrackerPatternOrderFieldEnrichedAt: + CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt: return true } @@ -91,8 +91,8 @@ func (v CommonTrackerPatternOrderField) Column() string { return "created_at" case CommonTrackerPatternOrderFieldUpdatedAt: return "updated_at" - case CommonTrackerPatternOrderFieldEnrichedAt: - return "COALESCE(enriched_at, '0001-01-01T00:00:00Z'::timestamptz)" + case CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt: + return "COALESCE(last_enrichment_attempt_at, '0001-01-01T00:00:00Z'::timestamptz)" } panic(fmt.Sprintf("unsupported order by: %s", v)) diff --git a/pkg/coredata/common_tracker_pattern_test.go b/pkg/coredata/common_tracker_pattern_test.go index 896590489..128132829 100644 --- a/pkg/coredata/common_tracker_pattern_test.go +++ b/pkg/coredata/common_tracker_pattern_test.go @@ -16,6 +16,7 @@ package coredata_test import ( "context" + "encoding/json" "testing" "time" @@ -100,12 +101,12 @@ func loadCommonTrackerPattern( return reloaded } -// TestCommonTrackerPattern_SetEnriched_AllowsEmptyDescription pins the -// no-fabrication contract: the enrichment worker records an empty -// description when it cannot substantiate a purpose, and the row is -// still marked terminally enriched so the stale-recovery loop never -// re-queues it. -func TestCommonTrackerPattern_SetEnriched_AllowsEmptyDescription(t *testing.T) { +// TestCommonTrackerPattern_UpdateEnrichment_AllowsEmptyDescription pins +// the no-fabrication contract: the enrichment worker records an empty +// description when it cannot substantiate a purpose, but still writes an +// enrichment payload so the row reads as having been through the workflow +// and the stale-recovery loop never re-queues it. +func TestCommonTrackerPattern_UpdateEnrichment_AllowsEmptyDescription(t *testing.T) { t.Parallel() client := test.PGClient(t) @@ -126,30 +127,32 @@ func TestCommonTrackerPattern_SetEnriched_AllowsEmptyDescription(t *testing.T) { } insertCommonTrackerPattern(t, ctx, client, cp) + payload := json.RawMessage(`{"status":"no_result","fields":{"description":{"status":"not_found"}}}`) + require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error { - return cp.SetEnriched(ctx, tx, "", nil) + return cp.UpdateEnrichment(ctx, tx, "", nil, payload) })) reloaded := loadCommonTrackerPattern(t, ctx, client, cp.ID) assert.Equal(t, "", reloaded.Description, "blank description must stay blank") - assert.NotNil(t, reloaded.EnrichedAt, "blank row must be marked enriched (terminal-for-now)") + assert.NotEmpty(t, reloaded.Enrichment, "blank row must still record an enrichment payload") assert.Nil(t, reloaded.EnrichmentRequestedAt, "enriched row must leave the queue") // A blank but enriched row must NOT be re-queued by stale recovery: - // enriched_at is set, so the stale sweep skips it. + // the enrichment payload is present, so the stale sweep skips it. require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error { - return coredata.ResetStaleEnrichments(ctx, conn, 0) + return coredata.ResetStaleEnrichments(ctx, conn, 0, 3) })) afterSweep := loadCommonTrackerPattern(t, ctx, client, cp.ID) assert.Nil(t, afterSweep.EnrichmentRequestedAt, "stale recovery must not re-queue an enriched blank row") } -// TestCommonTrackerPattern_SetEnriched_LinksThirdPartyWithoutOverride +// TestCommonTrackerPattern_UpdateEnrichment_LinksThirdPartyWithoutOverride // pins the link-no-override contract: the enrichment worker links a // resolved third party only when the row has none, and never clobbers an // attribution the mapping pipeline already resolved. -func TestCommonTrackerPattern_SetEnriched_LinksThirdPartyWithoutOverride(t *testing.T) { +func TestCommonTrackerPattern_UpdateEnrichment_LinksThirdPartyWithoutOverride(t *testing.T) { t.Parallel() client := test.PGClient(t) @@ -173,7 +176,7 @@ func TestCommonTrackerPattern_SetEnriched_LinksThirdPartyWithoutOverride(t *test insertCommonTrackerPattern(t, ctx, client, cp) require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error { - return cp.SetEnriched(ctx, tx, "Analytics tracker.", &party.ID) + return cp.UpdateEnrichment(ctx, tx, "Analytics tracker.", &party.ID, json.RawMessage(`{"status":"done"}`)) })) reloaded := loadCommonTrackerPattern(t, ctx, client, cp.ID) @@ -195,7 +198,7 @@ func TestCommonTrackerPattern_SetEnriched_LinksThirdPartyWithoutOverride(t *test insertCommonTrackerPattern(t, ctx, client, cp) require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error { - return cp.SetEnriched(ctx, tx, "Analytics tracker.", &other.ID) + return cp.UpdateEnrichment(ctx, tx, "Analytics tracker.", &other.ID, json.RawMessage(`{"status":"done"}`)) })) reloaded := loadCommonTrackerPattern(t, ctx, client, cp.ID) @@ -218,20 +221,23 @@ func TestCommonTrackerPattern_Upsert_RequeuesBlankRowOnThirdPartyLink(t *testing party := seedCommonThirdParty(t, ctx, client) now := time.Now().UTC().Truncate(time.Microsecond) - enrichedAt := now.Add(-time.Hour) + attemptAt := now.Add(-time.Hour) pattern := "requeue_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String() - // Stage a terminal blank row: enriched, no description, no vendor. + // Stage a completed blank row: enriched (carries a payload), no + // description, no vendor, with prior attempts spent. blank := coredata.CommonTrackerPattern{ - ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType), - TrackerType: coredata.TrackerTypeCookie, - Pattern: pattern, - MatchType: coredata.TrackerPatternMatchTypeExact, - Description: "", - Confidence: 0.5, - EnrichedAt: &enrichedAt, - CreatedAt: now, - UpdatedAt: now, + ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType), + TrackerType: coredata.TrackerTypeCookie, + Pattern: pattern, + MatchType: coredata.TrackerPatternMatchTypeExact, + Description: "", + Confidence: 0.5, + Enrichment: json.RawMessage(`{"status":"no_result"}`), + EnrichmentAttempts: 2, + LastEnrichmentAttemptAt: &attemptAt, + CreatedAt: now, + UpdatedAt: now, } insertCommonTrackerPattern(t, ctx, client, blank) @@ -263,7 +269,7 @@ func TestCommonTrackerPattern_Upsert_RequeuesBlankRowOnThirdPartyLink(t *testing require.NotNil(t, reloaded.CommonThirdPartyID) assert.Equal(t, party.ID, *reloaded.CommonThirdPartyID, "blank row must gain the linked third party") assert.NotNil(t, reloaded.EnrichmentRequestedAt, "linking a vendor must re-queue the blank row for enrichment") - assert.Nil(t, reloaded.EnrichedAt, "re-queued row must no longer be terminal") + assert.Equal(t, 0, reloaded.EnrichmentAttempts, "re-queued row must get a fresh retry budget") } // TestCommonTrackerPattern_Upsert_KeepsDescribedRowTerminal pins the @@ -279,19 +285,21 @@ func TestCommonTrackerPattern_Upsert_KeepsDescribedRowTerminal(t *testing.T) { party := seedCommonThirdParty(t, ctx, client) now := time.Now().UTC().Truncate(time.Microsecond) - enrichedAt := now.Add(-time.Hour) + attemptAt := now.Add(-time.Hour) pattern := "described_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String() described := coredata.CommonTrackerPattern{ - ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType), - TrackerType: coredata.TrackerTypeCookie, - Pattern: pattern, - MatchType: coredata.TrackerPatternMatchTypeExact, - Description: "An established analytics cookie.", - Confidence: 0.9, - EnrichedAt: &enrichedAt, - CreatedAt: now, - UpdatedAt: now, + ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType), + TrackerType: coredata.TrackerTypeCookie, + Pattern: pattern, + MatchType: coredata.TrackerPatternMatchTypeExact, + Description: "An established analytics cookie.", + Confidence: 0.9, + Enrichment: json.RawMessage(`{"status":"done"}`), + EnrichmentAttempts: 1, + LastEnrichmentAttemptAt: &attemptAt, + CreatedAt: now, + UpdatedAt: now, } insertCommonTrackerPattern(t, ctx, client, described) @@ -315,5 +323,93 @@ func TestCommonTrackerPattern_Upsert_KeepsDescribedRowTerminal(t *testing.T) { reloaded := loadCommonTrackerPattern(t, ctx, client, described.ID) assert.Equal(t, "An established analytics cookie.", reloaded.Description, "existing description must be preserved") assert.Nil(t, reloaded.EnrichmentRequestedAt, "described row must not be re-queued") - assert.NotNil(t, reloaded.EnrichedAt, "described row must stay terminal") + assert.NotEmpty(t, reloaded.Enrichment, "described row must keep its enrichment payload") + assert.Equal(t, 1, reloaded.EnrichmentAttempts, "described row's retry budget must be untouched") +} + +// TestCommonTrackerPattern_ClearEnrichmentRequestedAt_CountsAttempt pins +// the claim contract: dequeuing a row counts the attempt and stamps the +// idle clock, so a crash before UpdateEnrichment still spends part of the +// retry budget and the stale clock starts at claim time. +func TestCommonTrackerPattern_ClearEnrichmentRequestedAt_CountsAttempt(t *testing.T) { + t.Parallel() + + client := test.PGClient(t) + ctx := context.Background() + + now := time.Now().UTC().Truncate(time.Microsecond) + requestedAt := now.Add(-time.Minute) + cp := coredata.CommonTrackerPattern{ + ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType), + TrackerType: coredata.TrackerTypeCookie, + Pattern: "claim_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String(), + MatchType: coredata.TrackerPatternMatchTypeExact, + Confidence: 0.5, + EnrichmentRequestedAt: &requestedAt, + CreatedAt: now, + UpdatedAt: now, + } + insertCommonTrackerPattern(t, ctx, client, cp) + + require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error { + return cp.ClearEnrichmentRequestedAt(ctx, tx) + })) + + reloaded := loadCommonTrackerPattern(t, ctx, client, cp.ID) + assert.Nil(t, reloaded.EnrichmentRequestedAt, "claim must remove the row from the queue") + assert.Equal(t, 1, reloaded.EnrichmentAttempts, "claim must count the attempt") + assert.NotNil(t, reloaded.LastEnrichmentAttemptAt, "claim must stamp the idle clock") +} + +// TestCommonTrackerPattern_ResetStaleEnrichments_RespectsMaxAttempts pins +// the retry-budget contract: stale recovery re-queues a claimed-but- +// incomplete row that still has budget, but leaves a row at the +// max-attempts ceiling alone so a permanently failing row does not loop. +func TestCommonTrackerPattern_ResetStaleEnrichments_RespectsMaxAttempts(t *testing.T) { + t.Parallel() + + client := test.PGClient(t) + ctx := context.Background() + + now := time.Now().UTC().Truncate(time.Microsecond) + staleAttempt := now.Add(-time.Hour) + + // A claimed-but-incomplete row with budget left: no payload yet, one + // attempt spent, idle past the threshold. + eligible := coredata.CommonTrackerPattern{ + ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType), + TrackerType: coredata.TrackerTypeCookie, + Pattern: "stale_eligible_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String(), + MatchType: coredata.TrackerPatternMatchTypeExact, + Confidence: 0.5, + EnrichmentAttempts: 1, + LastEnrichmentAttemptAt: &staleAttempt, + CreatedAt: now, + UpdatedAt: now, + } + insertCommonTrackerPattern(t, ctx, client, eligible) + + // A row that has exhausted its retry budget must be left alone. + exhausted := coredata.CommonTrackerPattern{ + ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType), + TrackerType: coredata.TrackerTypeCookie, + Pattern: "stale_exhausted_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String(), + MatchType: coredata.TrackerPatternMatchTypeExact, + Confidence: 0.5, + EnrichmentAttempts: 3, + LastEnrichmentAttemptAt: &staleAttempt, + CreatedAt: now, + UpdatedAt: now, + } + insertCommonTrackerPattern(t, ctx, client, exhausted) + + require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error { + return coredata.ResetStaleEnrichments(ctx, conn, time.Minute, 3) + })) + + reloadedEligible := loadCommonTrackerPattern(t, ctx, client, eligible.ID) + assert.NotNil(t, reloadedEligible.EnrichmentRequestedAt, "stale row with budget must be re-queued") + + reloadedExhausted := loadCommonTrackerPattern(t, ctx, client, exhausted.ID) + assert.Nil(t, reloadedExhausted.EnrichmentRequestedAt, "row at the max-attempts ceiling must not be re-queued") } diff --git a/pkg/coredata/migrations/20260615T161615Z.sql b/pkg/coredata/migrations/20260615T161615Z.sql new file mode 100644 index 000000000..d88d062c1 --- /dev/null +++ b/pkg/coredata/migrations/20260615T161615Z.sql @@ -0,0 +1,37 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission to use, copy, modify, and/or distribute this software for any +-- purpose with or without fee is hereby granted, provided that the above +-- copyright notice and this permission notice appear in all copies. +-- +-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +-- PERFORMANCE OF THIS SOFTWARE. + +-- Unify the enrichment-tracking model across common_tracker_patterns and +-- common_third_parties. common_tracker_patterns gains a JSONB provenance +-- payload and an attempt counter (mirroring common_third_parties), and its +-- enriched_at done-flag is renamed to last_enrichment_attempt_at, stamped at +-- claim time. "Enriched" is now detected via the enrichment payload presence, +-- not the timestamp. +ALTER TABLE common_tracker_patterns + ADD COLUMN enrichment JSONB, + ADD COLUMN enrichment_attempts INTEGER NOT NULL DEFAULT 0; + +-- The DEFAULT only backfills existing rows; drop it so inserts must supply +-- the value explicitly. +ALTER TABLE common_tracker_patterns + ALTER COLUMN enrichment_attempts DROP DEFAULT; + +ALTER TABLE common_tracker_patterns + RENAME COLUMN enriched_at TO last_enrichment_attempt_at; + +-- common_third_parties already carries enrichment/enrichment_attempts; give +-- it the same explicit last-attempt timestamp (previously only recorded in +-- the enrichment JSON) so the stale-recovery clock has a dedicated column. +ALTER TABLE common_third_parties + ADD COLUMN last_enrichment_attempt_at TIMESTAMP WITH TIME ZONE; diff --git a/pkg/proboctl/commonthirdparty/commonthirdparty.go b/pkg/proboctl/commonthirdparty/commonthirdparty.go index f6e7fd38f..b1450ae2f 100644 --- a/pkg/proboctl/commonthirdparty/commonthirdparty.go +++ b/pkg/proboctl/commonthirdparty/commonthirdparty.go @@ -48,19 +48,63 @@ func NewCmdCommonThirdParty(f *cmdutil.Factory) *cobra.Command { } // enrichmentState classifies a common third party's position in the -// enrichment lifecycle for display. A row is "enriched" once it carries -// an enrichment payload; there is no enriched_at column. +// enrichment lifecycle for display. A row that has been through the +// workflow (it carries an enrichment payload) reads "enriched" only when +// every field the last run recorded an outcome for resolved a value; +// otherwise it reads "partial (X/Y)". func enrichmentState(p *coredata.CommonThirdParty) string { switch { case p.EnrichmentRequestedAt != nil: return "queued" case len(p.Enrichment) > 0: - return "enriched" + resolved, total := enrichmentCompleteness(p) + if total == 0 || resolved == total { + return "enriched" + } + + return fmt.Sprintf("partial (%d/%d)", resolved, total) default: return "unenriched" } } +// resolvedFieldStatuses are the per-field enrichment statuses that carry a +// value, as opposed to not_found / low_confidence. +var resolvedFieldStatuses = map[string]struct{}{ + "found": {}, + "exists_external": {}, + "fallback_display_name": {}, +} + +// enrichmentCompleteness counts how many of the fields the last enrichment +// run recorded an outcome for resolved a value (X) versus the total it +// recorded (Y), parsed from the enrichment payload's per-field provenance. +func enrichmentCompleteness(p *coredata.CommonThirdParty) (resolved, total int) { + if len(p.Enrichment) == 0 { + return 0, 0 + } + + var meta struct { + Fields map[string]struct { + Status string `json:"status"` + } `json:"fields"` + } + + if err := json.Unmarshal(p.Enrichment, &meta); err != nil { + return 0, 0 + } + + for _, f := range meta.Fields { + total++ + + if _, ok := resolvedFieldStatuses[f.Status]; ok { + resolved++ + } + } + + return resolved, total +} + // enrichmentStatus returns the run-level status recorded in the // enrichment payload (done, partial, failed), or an empty string when // the row has never been enriched or the payload is malformed. diff --git a/pkg/proboctl/commonthirdparty/show.go b/pkg/proboctl/commonthirdparty/show.go index 08ea00736..612a667e7 100644 --- a/pkg/proboctl/commonthirdparty/show.go +++ b/pkg/proboctl/commonthirdparty/show.go @@ -147,6 +147,10 @@ func newCmdShow(f *cmdutil.Factory) *cobra.Command { row("Enrichment state:", enrichmentState(&party)) row("Enrichment attempts:", fmt.Sprintf("%d", party.EnrichmentAttempts)) + if party.LastEnrichmentAttemptAt != nil { + row("Last attempt:", party.LastEnrichmentAttemptAt.Format("2006-01-02 15:04:05")) + } + if party.EnrichmentRequestedAt != nil { row("Queued at:", party.EnrichmentRequestedAt.Format("2006-01-02 15:04:05")) } @@ -180,7 +184,7 @@ func printEnrichmentDetails(out io.Writer, label lipgloss.Style, party coredata. } if !meta.AttemptedAt.IsZero() { - row("Last attempt:", meta.AttemptedAt.Format("2006-01-02 15:04:05")) + row("Last run recorded:", meta.AttemptedAt.Format("2006-01-02 15:04:05")) } if meta.Model != "" { diff --git a/pkg/proboctl/commontrackerpattern/commontrackerpattern.go b/pkg/proboctl/commontrackerpattern/commontrackerpattern.go index c90357045..f5679272a 100644 --- a/pkg/proboctl/commontrackerpattern/commontrackerpattern.go +++ b/pkg/proboctl/commontrackerpattern/commontrackerpattern.go @@ -16,6 +16,7 @@ package commontrackerpattern import ( "context" + "encoding/json" "errors" "fmt" @@ -47,20 +48,62 @@ func NewCmdCommonTrackerPattern(f *cmdutil.Factory) *cobra.Command { } // enrichmentState classifies a pattern's position in the enrichment -// lifecycle for display. +// lifecycle for display. A row that has been through the workflow (it +// carries an enrichment payload) reads "enriched" only when every field +// the last run recorded an outcome for resolved a value; otherwise it +// reads "partial (X/Y)". func enrichmentState(p *coredata.CommonTrackerPattern) string { switch { case p.EnrichmentRequestedAt != nil: return "queued" - case p.EnrichedAt != nil && p.Description == "": - return "enriched (no description)" - case p.EnrichedAt != nil: - return "enriched" + case len(p.Enrichment) > 0: + resolved, total := enrichmentCompleteness(p) + if total == 0 || resolved == total { + return "enriched" + } + + return fmt.Sprintf("partial (%d/%d)", resolved, total) default: return "unenriched" } } +// resolvedFieldStatuses are the per-field enrichment statuses that carry a +// value, as opposed to not_found. +var resolvedFieldStatuses = map[string]struct{}{ + "found": {}, + "exists_external": {}, +} + +// enrichmentCompleteness counts how many of the fields the last enrichment +// run recorded an outcome for resolved a value (X) versus the total it +// recorded (Y), parsed from the enrichment payload's per-field provenance. +func enrichmentCompleteness(p *coredata.CommonTrackerPattern) (resolved, total int) { + if len(p.Enrichment) == 0 { + return 0, 0 + } + + var meta struct { + Fields map[string]struct { + Status string `json:"status"` + } `json:"fields"` + } + + if err := json.Unmarshal(p.Enrichment, &meta); err != nil { + return 0, 0 + } + + for _, f := range meta.Fields { + total++ + + if _, ok := resolvedFieldStatuses[f.Status]; ok { + resolved++ + } + } + + return resolved, total +} + // resolveCommonThirdPartyID accepts either a common third party GID or a // slug and returns the corresponding id. func resolveCommonThirdPartyID(ctx context.Context, conn pg.Querier, value string) (gid.GID, error) { diff --git a/pkg/proboctl/commontrackerpattern/list.go b/pkg/proboctl/commontrackerpattern/list.go index ccc8edf62..42a961ff6 100644 --- a/pkg/proboctl/commontrackerpattern/list.go +++ b/pkg/proboctl/commontrackerpattern/list.go @@ -59,7 +59,7 @@ func newCmdList(f *cmdutil.Factory) *cobra.Command { cmd.Flags().StringVar(&flagState, "state", "", "Filter by enrichment state (queued, enriched, unenriched)") cmd.Flags().BoolVar(&flagWithCommonThirdParty, "with-common-third-party", false, "Filter by whether the pattern is linked to a common third party (true/false); ignored when not set") cmd.Flags().BoolVar(&flagWithoutDescription, "without-description", false, "Only patterns with a blank description") - cmd.Flags().StringVar(&flagSort, "sort", "confidence", "Sort field: pattern, confidence, created, updated, enriched") + cmd.Flags().StringVar(&flagSort, "sort", "confidence", "Sort field: pattern, confidence, created, updated, attempted") cmd.Flags().StringVar(&flagOrder, "order", "", "Sort order: asc, desc (default depends on field)") pageFlags := cmdutil.AddPageFlags(cmd) @@ -277,10 +277,10 @@ func parseOrderBy(sort, order string) (page.OrderBy[coredata.CommonTrackerPatter field, defaultDesc = coredata.CommonTrackerPatternOrderFieldCreatedAt, true case "updated": field, defaultDesc = coredata.CommonTrackerPatternOrderFieldUpdatedAt, true - case "enriched": - field, defaultDesc = coredata.CommonTrackerPatternOrderFieldEnrichedAt, true + case "attempted": + field, defaultDesc = coredata.CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt, true default: - return zeroOrderBy, fmt.Errorf("invalid --sort value %q: valid values are pattern, confidence, created, updated, enriched", sort) + return zeroOrderBy, fmt.Errorf("invalid --sort value %q: valid values are pattern, confidence, created, updated, attempted", sort) } direction := page.OrderDirectionAsc diff --git a/pkg/proboctl/commontrackerpattern/set_description.go b/pkg/proboctl/commontrackerpattern/set_description.go index cfef1a3af..fa931042f 100644 --- a/pkg/proboctl/commontrackerpattern/set_description.go +++ b/pkg/proboctl/commontrackerpattern/set_description.go @@ -16,8 +16,10 @@ package commontrackerpattern import ( "context" + "encoding/json" "errors" "fmt" + "time" "github.com/spf13/cobra" "go.gearno.de/kit/pg" @@ -26,6 +28,41 @@ import ( "go.probo.inc/probo/pkg/proboctl/cmdutil" ) +// manualEnrichmentPayload builds the enrichment provenance written when an +// operator sets a description by hand. It records only the description +// outcome (resolved), so the row reads "enriched" and the enrichment +// worker leaves it alone, while marking the source as manual for audit. +func manualEnrichmentPayload() json.RawMessage { + now := time.Now() + + payload := struct { + Status string `json:"status"` + Source string `json:"source"` + AttemptedAt time.Time `json:"attempted_at"` + Fields map[string]struct { + Status string `json:"status"` + UpdatedAt time.Time `json:"updated_at"` + } `json:"fields"` + }{ + Status: "manual", + Source: "manual", + AttemptedAt: now, + Fields: map[string]struct { + Status string `json:"status"` + UpdatedAt time.Time `json:"updated_at"` + }{ + "description": {Status: "found", UpdatedAt: now}, + }, + } + + raw, err := json.Marshal(payload) + if err != nil { + return nil + } + + return raw +} + func newCmdSetDescription(f *cmdutil.Factory) *cobra.Command { var ( flagDescription string @@ -83,7 +120,7 @@ func newCmdSetDescription(f *cmdutil.Factory) *cobra.Command { return fmt.Errorf("cannot load common tracker pattern: %w", err) } - if err := pattern.SetEnriched(ctx, tx, flagDescription, nil); err != nil { + if err := pattern.UpdateEnrichment(ctx, tx, flagDescription, nil, manualEnrichmentPayload()); err != nil { return fmt.Errorf("cannot set common tracker pattern description: %w", err) } diff --git a/pkg/proboctl/commontrackerpattern/show.go b/pkg/proboctl/commontrackerpattern/show.go index 456216d71..78ba80122 100644 --- a/pkg/proboctl/commontrackerpattern/show.go +++ b/pkg/proboctl/commontrackerpattern/show.go @@ -127,12 +127,14 @@ func renderPatternDetail(f *cmdutil.Factory, p coredata.CommonTrackerPattern, th row("Description:", description) + row("Enrichment attempts:", fmt.Sprintf("%d", p.EnrichmentAttempts)) + if p.EnrichmentRequestedAt != nil { row("Enrichment queued:", p.EnrichmentRequestedAt.Format("2006-01-02 15:04:05")) } - if p.EnrichedAt != nil { - row("Enriched at:", p.EnrichedAt.Format("2006-01-02 15:04:05")) + if p.LastEnrichmentAttemptAt != nil { + row("Last attempt:", p.LastEnrichmentAttemptAt.Format("2006-01-02 15:04:05")) } row("Created:", p.CreatedAt.Format("2006-01-02 15:04:05")) diff --git a/pkg/probod/probod.go b/pkg/probod/probod.go index 9feb3abf2..879c0745b 100644 --- a/pkg/probod/probod.go +++ b/pkg/probod/probod.go @@ -817,6 +817,7 @@ func (impl *Implm) Run( trackerEnrichmentCfg, trackerMappingCfg, time.Duration(impl.cfg.CommonPatternEnrichmentWorker.StaleAfter)*time.Second, + 0, worker.WithInterval(time.Duration(impl.cfg.CommonPatternEnrichmentWorker.Interval)*time.Second), worker.WithMaxConcurrency(impl.cfg.CommonPatternEnrichmentWorker.MaxConcurrency), )