Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions workers/publisher/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,24 @@ async function scheduleValidatedCandidates(
ON promotion.candidate_id = candidate.candidate_id
WHERE candidate.candidate_status = 'validated'
AND candidate.gate_status IN ('rule-pass', 'dual-pass')
AND json_array_length(candidate.facts_json) > 0
AND NOT EXISTS (
SELECT 1
FROM json_each(candidate.facts_json) fact
WHERE NOT EXISTS (
SELECT 1
FROM promotion_field_mappings mapping
JOIN records target ON target.id = mapping.subject_record_id
JOIN field_definitions definition
ON definition.record_kind = target.kind
AND definition.field_path = mapping.canonical_field_path
WHERE mapping.source_id = candidate.source_id
AND mapping.enabled = 1
AND mapping.candidate_field_path = CASE
WHEN fact.type = 'object' THEN json_extract(fact.value, '$.fieldPath')
END
)
)
AND (
promotion.candidate_id IS NULL
OR (
Expand Down Expand Up @@ -94,6 +112,8 @@ async function handleQueue(
if (result.status === 'busy') {
message.retry({ delaySeconds: 60 })
} else {
// A deferred dependency cannot be repaired by queue retries. The
// scheduler will rediscover it once every exact mapping is available.
message.ack()
}
} catch (error) {
Expand Down
19 changes: 18 additions & 1 deletion workers/publisher/src/promoter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,13 @@ class UnsafeCandidateError extends Error {
}
}

class DeferredCandidateError extends Error {
constructor(readonly code: 'field_mapping_missing', message: string) {
super(message)
this.name = 'DeferredCandidateError'
}
}

function unsafe(code: string, issue: string | string[]): never {
throw new UnsafeCandidateError(code, Array.isArray(issue) ? issue : [issue])
}
Expand Down Expand Up @@ -503,7 +510,12 @@ async function buildPlan(
const facts: PlannedFact[] = []
for (const fact of candidate.facts) {
const mapping = byField.get(fact.fieldPath)
if (!mapping) unsafe('field_mapping_missing', `No exact promotion mapping for ${fact.fieldPath}`)
if (!mapping) {
throw new DeferredCandidateError(
'field_mapping_missing',
`No exact promotion mapping for ${fact.fieldPath}`,
)
}
if (['quarantined', 'archived', 'rejected'].includes(mapping.workflow_status)) {
unsafe('target_record_blocked', `Target record ${mapping.subject_record_id} is ${mapping.workflow_status}`)
}
Expand Down Expand Up @@ -1063,6 +1075,11 @@ export async function promoteCandidate(
const candidate = await validateCandidate(row)
plan = await buildPlan(database, candidate)
} catch (error) {
if (error instanceof DeferredCandidateError) {
// Mapping is an operator-owned dependency, not an evidence failure.
// Keep the validated candidate untouched so a later poll can retry it.
return { candidateId, status: 'deferred', reasonCode: error.code }
}
if (!(error instanceof UnsafeCandidateError)) throw error
await isolateCandidate(database, candidateId, error.code, error.issues, now)
return { candidateId, status: 'quarantined', reasonCode: error.code }
Expand Down
2 changes: 1 addition & 1 deletion workers/publisher/src/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ export interface PublisherEnv {

export type PromotionResult = {
candidateId: string
status: 'applied' | 'already-applied' | 'quarantined' | 'busy'
status: 'applied' | 'already-applied' | 'quarantined' | 'busy' | 'deferred'
publicationJobId?: string
reasonCode?: string
}
Loading
Loading