Keep attribution consistent with vendor links
Three paths could leave a catalog row's attribution out of step with its common_third_party_id. A FIRST_PARTY reclassification in the mapping worker kept a stale org ThirdParty link instead of clearing it. The upsert requeued terminal FIRST_PARTY rows for enrichment on a vendor they never adopt, since the vendor-preservation clause nulls it. And the proboctl upsert command did not normalize the verdict when an operator linked or unlinked a vendor without passing --attribution. Clear the org link on a first-party verdict, exclude FIRST_PARTY rows from the enrichment requeue, and have the CLI downgrade THIRD_PARTY to UNDETERMINED on unlink and promote UNDETERMINED to THIRD_PARTY on link. Signed-off-by: Émile Ré <emile@probo.com>
This commit is contained in:
@@ -215,6 +215,7 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
|
|||||||
commonPatternID := det.commonPatternID
|
commonPatternID := det.commonPatternID
|
||||||
commonThirdPartyID := det.commonThirdPartyID
|
commonThirdPartyID := det.commonThirdPartyID
|
||||||
directThirdPartyID := det.directThirdPartyID
|
directThirdPartyID := det.directThirdPartyID
|
||||||
|
firstParty := det.firstParty
|
||||||
|
|
||||||
// Phase 2: tracker-mapping agent (no transaction). It runs only when
|
// Phase 2: tracker-mapping agent (no transaction). It runs only when
|
||||||
// the deterministic signals could not resolve a catalog third party.
|
// the deterministic signals could not resolve a catalog third party.
|
||||||
@@ -251,6 +252,7 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
|
|||||||
|
|
||||||
commonPatternID = firstNonNil(commonPatternID, match.commonPatternID)
|
commonPatternID = firstNonNil(commonPatternID, match.commonPatternID)
|
||||||
commonThirdPartyID = match.commonThirdPartyID
|
commonThirdPartyID = match.commonThirdPartyID
|
||||||
|
firstParty = match.firstParty
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
@@ -265,7 +267,12 @@ func (h *trackerMappingHandler) Process(ctx context.Context, tp coredata.Tracker
|
|||||||
// touches the database (in a short transaction).
|
// touches the database (in a short transaction).
|
||||||
thirdPartyID := tp.ThirdPartyID
|
thirdPartyID := tp.ThirdPartyID
|
||||||
|
|
||||||
if thirdPartyID == nil {
|
// A first-party verdict is terminal: the artifact has no vendor, so
|
||||||
|
// any org ThirdParty link a prior mapping run left on the pattern is
|
||||||
|
// stale and must be cleared.
|
||||||
|
if firstParty {
|
||||||
|
thirdPartyID = nil
|
||||||
|
} else if thirdPartyID == nil {
|
||||||
switch {
|
switch {
|
||||||
case directThirdPartyID != nil:
|
case directThirdPartyID != nil:
|
||||||
thirdPartyID = directThirdPartyID
|
thirdPartyID = directThirdPartyID
|
||||||
|
|||||||
@@ -309,16 +309,20 @@ SET
|
|||||||
-- re-queued for enrichment: the enrichment agent leaves descriptions
|
-- re-queued for enrichment: the enrichment agent leaves descriptions
|
||||||
-- blank when it cannot substantiate a purpose, and knowing the vendor
|
-- blank when it cannot substantiate a purpose, and knowing the vendor
|
||||||
-- gives it a second, better-informed attempt. The attempt counter is
|
-- gives it a second, better-informed attempt. The attempt counter is
|
||||||
-- reset so the re-armed row gets a fresh retry budget.
|
-- reset so the re-armed row gets a fresh retry budget. A terminal
|
||||||
|
-- FIRST_PARTY row never gains a vendor (the clause above discards the
|
||||||
|
-- incoming one), so it must never re-arm on a vendor it did not adopt.
|
||||||
enrichment_requested_at = CASE
|
enrichment_requested_at = CASE
|
||||||
WHEN common_tracker_patterns.description = ''
|
WHEN common_tracker_patterns.attribution <> 'FIRST_PARTY'
|
||||||
|
AND common_tracker_patterns.description = ''
|
||||||
AND common_tracker_patterns.common_third_party_id IS NULL
|
AND common_tracker_patterns.common_third_party_id IS NULL
|
||||||
AND EXCLUDED.common_third_party_id IS NOT NULL
|
AND EXCLUDED.common_third_party_id IS NOT NULL
|
||||||
THEN NOW()
|
THEN NOW()
|
||||||
ELSE common_tracker_patterns.enrichment_requested_at
|
ELSE common_tracker_patterns.enrichment_requested_at
|
||||||
END,
|
END,
|
||||||
enrichment_attempts = CASE
|
enrichment_attempts = CASE
|
||||||
WHEN common_tracker_patterns.description = ''
|
WHEN common_tracker_patterns.attribution <> 'FIRST_PARTY'
|
||||||
|
AND common_tracker_patterns.description = ''
|
||||||
AND common_tracker_patterns.common_third_party_id IS NULL
|
AND common_tracker_patterns.common_third_party_id IS NULL
|
||||||
AND EXCLUDED.common_third_party_id IS NOT NULL
|
AND EXCLUDED.common_third_party_id IS NOT NULL
|
||||||
THEN 0
|
THEN 0
|
||||||
@@ -330,7 +334,8 @@ SET
|
|||||||
-- unrecoverable: the stale-recovery sweep skips any row whose payload
|
-- unrecoverable: the stale-recovery sweep skips any row whose payload
|
||||||
-- is non-null, so the re-claimed-but-never-finished row never requeues.
|
-- is non-null, so the re-claimed-but-never-finished row never requeues.
|
||||||
enrichment = CASE
|
enrichment = CASE
|
||||||
WHEN common_tracker_patterns.description = ''
|
WHEN common_tracker_patterns.attribution <> 'FIRST_PARTY'
|
||||||
|
AND common_tracker_patterns.description = ''
|
||||||
AND common_tracker_patterns.common_third_party_id IS NULL
|
AND common_tracker_patterns.common_third_party_id IS NULL
|
||||||
AND EXCLUDED.common_third_party_id IS NOT NULL
|
AND EXCLUDED.common_third_party_id IS NOT NULL
|
||||||
THEN NULL
|
THEN NULL
|
||||||
|
|||||||
@@ -158,6 +158,18 @@ func newCmdUpsert(f *cmdutil.Factory) *cobra.Command {
|
|||||||
if cmd.Flags().Changed("common-third-party") {
|
if cmd.Flags().Changed("common-third-party") {
|
||||||
if flagCommonThirdParty == "" {
|
if flagCommonThirdParty == "" {
|
||||||
pattern.CommonThirdPartyID = nil
|
pattern.CommonThirdPartyID = nil
|
||||||
|
|
||||||
|
// Removing the vendor invalidates a THIRD_PARTY
|
||||||
|
// verdict, which by definition carries one. When the
|
||||||
|
// operator did not set an explicit --attribution,
|
||||||
|
// downgrade the now-stale verdict to UNDETERMINED so
|
||||||
|
// the mapping pipeline probes the vendor-free row
|
||||||
|
// again. A FIRST_PARTY row is already vendor-free and
|
||||||
|
// terminal, so it is left untouched.
|
||||||
|
if !cmd.Flags().Changed("attribution") &&
|
||||||
|
pattern.Attribution == coredata.CommonTrackerPatternAttributionThirdParty {
|
||||||
|
pattern.Attribution = coredata.CommonTrackerPatternAttributionUndetermined
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
thirdPartyID, err := resolveCommonThirdPartyID(ctx, tx, flagCommonThirdParty)
|
thirdPartyID, err := resolveCommonThirdPartyID(ctx, tx, flagCommonThirdParty)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -165,6 +177,19 @@ func newCmdUpsert(f *cmdutil.Factory) *cobra.Command {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pattern.CommonThirdPartyID = &thirdPartyID
|
pattern.CommonThirdPartyID = &thirdPartyID
|
||||||
|
|
||||||
|
// A vendor-linked row is, by definition, attributed to
|
||||||
|
// a third party. When the operator did not set an
|
||||||
|
// explicit --attribution, normalize the verdict to
|
||||||
|
// THIRD_PARTY so the row never persists with an
|
||||||
|
// UNDETERMINED (or unset) verdict. A FIRST_PARTY row is
|
||||||
|
// terminal and stays vendor-free — the upsert discards
|
||||||
|
// the incoming vendor and keeps the verdict — so it is
|
||||||
|
// left untouched.
|
||||||
|
if !cmd.Flags().Changed("attribution") &&
|
||||||
|
pattern.Attribution != coredata.CommonTrackerPatternAttributionFirstParty {
|
||||||
|
pattern.Attribution = coredata.CommonTrackerPatternAttributionThirdParty
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user