Fix enrichment re-arm and migration backfill gaps
Three crash- and migration-recovery gaps in the unified enrichment model left rows stuck or misclassified: - Upsert re-armed a blank, newly-linked tracker pattern without clearing its prior enrichment payload. A crash between the worker's claim and persist then left the row with a stale payload, so the stale-recovery sweep (which only catches rows with a null payload) skipped it forever. Clear enrichment on re-arm so the row reads as not-yet-completed again, and pin the behavior with a test. - The migration added last_enrichment_attempt_at to common_third_parties without seeding it. Rows with prior attempts kept a NULL clock and could never satisfy the stale-reset predicate. Backfill from updated_at, the historical claim-time proxy. - The migration switched the tracker-pattern enriched-state source to the enrichment payload without backfilling rows previously marked by enriched_at, making already-enriched rows read as unenriched. Seed a provenance sentinel for rows that carried the old done-flag. Signed-off-by: Émile Ré <emile@probo.com>
This commit is contained in:
@@ -224,10 +224,14 @@ func (p *CommonTrackerPattern) Upsert(
|
|||||||
) (inserted bool, err error) {
|
) (inserted bool, err error) {
|
||||||
// On insert, a description-less row is immediately queued for the
|
// On insert, a description-less row is immediately queued for the
|
||||||
// enrichment worker (enrichment_requested_at = NOW()). On conflict the
|
// enrichment worker (enrichment_requested_at = NOW()). On conflict the
|
||||||
// enrichment columns are left untouched, and an empty incoming
|
// enrichment columns are otherwise left untouched, and an empty incoming
|
||||||
// description never overwrites an existing one — descriptions are owned
|
// description never overwrites an existing one — descriptions are owned
|
||||||
// by the enrichment worker, so mapping-side upserts must not clobber a
|
// by the enrichment worker, so mapping-side upserts must not clobber a
|
||||||
// researched description with an empty string.
|
// researched description with an empty string. The one exception is a
|
||||||
|
// blank, unlinked row that gains a third party: it is re-armed for
|
||||||
|
// enrichment, and re-arming resets the attempt counter and drops the
|
||||||
|
// prior payload so the row reads as not-yet-completed again (see the
|
||||||
|
// enrichment CASE below).
|
||||||
q := `
|
q := `
|
||||||
INSERT INTO common_tracker_patterns (
|
INSERT INTO common_tracker_patterns (
|
||||||
id,
|
id,
|
||||||
@@ -288,6 +292,18 @@ SET
|
|||||||
THEN 0
|
THEN 0
|
||||||
ELSE common_tracker_patterns.enrichment_attempts
|
ELSE common_tracker_patterns.enrichment_attempts
|
||||||
END,
|
END,
|
||||||
|
-- The prior attempt's payload is dropped so the re-armed row matches
|
||||||
|
-- the "not yet completed" predicate (enrichment IS NULL) again. Left
|
||||||
|
-- in place, a stale payload makes a crash between claim and persist
|
||||||
|
-- unrecoverable: the stale-recovery sweep skips any row whose payload
|
||||||
|
-- is non-null, so the re-claimed-but-never-finished row never requeues.
|
||||||
|
enrichment = 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.enrichment
|
||||||
|
END,
|
||||||
updated_at = EXCLUDED.updated_at
|
updated_at = EXCLUDED.updated_at
|
||||||
RETURNING
|
RETURNING
|
||||||
id,
|
id,
|
||||||
|
|||||||
@@ -270,6 +270,23 @@ func TestCommonTrackerPattern_Upsert_RequeuesBlankRowOnThirdPartyLink(t *testing
|
|||||||
assert.Equal(t, party.ID, *reloaded.CommonThirdPartyID, "blank row must gain the linked third party")
|
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.NotNil(t, reloaded.EnrichmentRequestedAt, "linking a vendor must re-queue the blank row for enrichment")
|
||||||
assert.Equal(t, 0, reloaded.EnrichmentAttempts, "re-queued row must get a fresh retry budget")
|
assert.Equal(t, 0, reloaded.EnrichmentAttempts, "re-queued row must get a fresh retry budget")
|
||||||
|
assert.Empty(t, reloaded.Enrichment, "re-armed row must drop the prior payload so it reads as not yet completed")
|
||||||
|
|
||||||
|
// The prior payload must be cleared so a crash between the worker's
|
||||||
|
// claim and persist stays recoverable. Simulate the claim (which bumps
|
||||||
|
// attempts past zero and stamps the idle clock) without completing, then
|
||||||
|
// confirm the stale sweep re-queues the row — it only catches rows whose
|
||||||
|
// payload is still null.
|
||||||
|
require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
|
||||||
|
return reloaded.ClearEnrichmentRequestedAt(ctx, tx)
|
||||||
|
}))
|
||||||
|
|
||||||
|
require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
|
||||||
|
return coredata.ResetStaleEnrichments(ctx, conn, 0, 3)
|
||||||
|
}))
|
||||||
|
|
||||||
|
afterSweep := loadCommonTrackerPattern(t, ctx, client, blank.ID)
|
||||||
|
assert.NotNil(t, afterSweep.EnrichmentRequestedAt, "stale recovery must re-queue a re-armed row claimed but never completed")
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestCommonTrackerPattern_Upsert_KeepsDescribedRowTerminal pins the
|
// TestCommonTrackerPattern_Upsert_KeepsDescribedRowTerminal pins the
|
||||||
|
|||||||
@@ -30,8 +30,30 @@ ALTER TABLE common_tracker_patterns
|
|||||||
ALTER TABLE common_tracker_patterns
|
ALTER TABLE common_tracker_patterns
|
||||||
RENAME COLUMN enriched_at TO last_enrichment_attempt_at;
|
RENAME COLUMN enriched_at TO last_enrichment_attempt_at;
|
||||||
|
|
||||||
|
-- Backfill the enriched-state marker for rows that were terminally enriched
|
||||||
|
-- under the old model. "Enriched" previously meant enriched_at IS NOT NULL (a
|
||||||
|
-- terminal done-flag); it now means a non-null enrichment payload. The rename
|
||||||
|
-- above moved the old done-flag into last_enrichment_attempt_at, so seed a
|
||||||
|
-- provenance payload for every row that carried it. Without this, rows already
|
||||||
|
-- enriched before the deploy would read as permanently unenriched. The payload
|
||||||
|
-- carries no per-field provenance, so completeness checks treat it as fully
|
||||||
|
-- enriched.
|
||||||
|
UPDATE common_tracker_patterns
|
||||||
|
SET enrichment = '{"status": "migrated"}'::jsonb
|
||||||
|
WHERE last_enrichment_attempt_at IS NOT NULL;
|
||||||
|
|
||||||
-- common_third_parties already carries enrichment/enrichment_attempts; give
|
-- common_third_parties already carries enrichment/enrichment_attempts; give
|
||||||
-- it the same explicit last-attempt timestamp (previously only recorded in
|
-- it the same explicit last-attempt timestamp (previously only recorded in
|
||||||
-- the enrichment JSON) so the stale-recovery clock has a dedicated column.
|
-- the enrichment JSON) so the stale-recovery clock has a dedicated column.
|
||||||
ALTER TABLE common_third_parties
|
ALTER TABLE common_third_parties
|
||||||
ADD COLUMN last_enrichment_attempt_at TIMESTAMP WITH TIME ZONE;
|
ADD COLUMN last_enrichment_attempt_at TIMESTAMP WITH TIME ZONE;
|
||||||
|
|
||||||
|
-- Seed the new clock for rows that already spent an attempt. The claim path
|
||||||
|
-- stamps last_enrichment_attempt_at and updated_at together, so updated_at is
|
||||||
|
-- the historical proxy for the last attempt. Without this, a row claimed but
|
||||||
|
-- never completed before the migration keeps a NULL clock, and the stale-reset
|
||||||
|
-- predicate (last_enrichment_attempt_at < stale_before) can never be true for
|
||||||
|
-- NULL, so it would never be re-queued. Never-attempted rows keep a NULL clock.
|
||||||
|
UPDATE common_third_parties
|
||||||
|
SET last_enrichment_attempt_at = updated_at
|
||||||
|
WHERE enrichment_attempts > 0;
|
||||||
|
|||||||
Reference in New Issue
Block a user