Design: CRIT Publisher

Overview

Drains CRIT candidate envelopes staged at s3://{bucket}/crit-candidates/pending/... by every CRIT-emitting processor — runs the publish-time validation suite, upserts the CritRecord row with newer-only conflict resolution, moves the S3 object to its terminal prefix (approved/inserted/ on success or rejected/spec-violation/ on failure), updates S3QueueObject status, and emits a sibling CycloneDX 1.6 VEX document when vex_status="fixed".

Two consumer entry points:

In-process drain (the common case): every Phase 1.x producer calls internal/critpublisher.DrainKeys at end-of-run with the keys it staged. Pending state becomes transient — by the time the producer exits, every staged envelope is either inserted or rejected.
cmd/crit-publisher binary (cross-source manual runs): reserved for replaying the rejected pile after a fix, or draining stragglers from a producer that crashed mid-run.

Source identifier: crit-publisher (writes only — not a producer) Data type: N/A (consumer of producer output) ECS task name: not scheduled; on-demand Phase: 0 (foundation) + Phase-1.x in-process drain

Data Source

Input — staged envelope

s3://{bucket}/crit-candidates/pending/{YYYY}/{MM}/{DD}/{cve}/{provider}/{service}/{rt}/{sha}.json

Each envelope is a JSON document conforming to internal/critutil.Envelope:

{
  "envelope_version": "2",
  "spec_version": "CRITv0.2.0",
  "producer": "salesforce-advisories-rss-processor/0.1.0",
  "generated_at": 1730000000000,
  "cve_id": "CVE-2025-26496",
  "natural_key": {"provider": "salesforce", "service": "tableau", "resource_type": "server"},
  "candidate": { ... CRIT v0.2.0 record ... },
  "validation": { ... producer-side validation results ... },
  "provenance": { ... primary_source + supporting_sources + evidence + confidence ... },
  "custom_dictionary": null,
  "diff": {"exists_in_db": false, "vector_changed": false, "fields_changed": []},
  "auto_approval_eligible": false,
  "auto_approval_reason": ""
}

Output

CritRecord row in PostgreSQL (UPSERT with newer-only WHERE clause keyed by (cveId, provider, service, resourceType)).
S3 object move from pending/... to approved/inserted/{YYYY}/{MM}/{DD}/{cve}/{provider}/{service}/{rt}/{sha}.json (success) or rejected/spec-violation/{YYYY}/{MM}/{DD}/{cve}/{provider}/{service}/{rt}/{sha}.json (failure).
S3QueueObject row update — status: pending → inserted/rejected, key remapped, errorCause populated on rejection.
VEX sibling at {movedKey}.vex.json when vex_status="fixed" and the upsert wrote a row (newer-only WHERE was satisfied). CycloneDX 1.6 with deterministic UUIDv5 serial number.

Architecture / Data Flow

sequenceDiagram participant Producer as Any Phase 1.x Producer participant Publisher as critpublisher.DrainKeys participant Worker as 4 worker goroutines participant S3 participant DB as PostgreSQL Producer->>Producer: stage CRIT envelopes Producer->>S3: PUT crit-candidates/pending/... Producer->>DB: RegisterS3QueueObject(status=pending) Producer->>Producer: collect stagedKeys Producer->>Publisher: DrainKeys(deps, keys, opts) Publisher->>Worker: jobs <- key (channel; 4 workers) par 4 workers in parallel Worker->>S3: GET pending/{key} Worker->>Publisher: ParseEnvelope (verifies envelope_version="2") Worker->>Publisher: Revalidate Note over Publisher: 1. ValidateCustomDictionary (if present)
2. ValidateRecord (schema + vector + slot ABNF + spec rules + dict resolution)
3. Vector recompute round-trip
4. SpecRules.Failed must be empty alt validation pass Worker->>DB: INSERT INTO CritRecord ... ON CONFLICT DO UPDATE WHERE older < newer Worker->>S3: CopyObject + DeleteObject (pending → approved/inserted/) Worker->>DB: RemapS3QueueObjectKey + MarkS3QueueObjectInserted opt vex_status=fixed AND wrote=true (rowsAffected>0) Worker->>Worker: vex.Build (CycloneDX 1.6, UUIDv5 over cveId|provider|service|rt) Worker->>S3: PUT {movedKey}.vex.json end else validation fail Worker->>S3: CopyObject + DeleteObject (pending → rejected/spec-violation/) Worker->>DB: RemapS3QueueObjectKey + MarkS3QueueObjectRejected(errorCause) end end Publisher-->>Producer: Stats{Processed, Inserted, Rejected, Errored}

Source → Database Mappings

`CritRecord` (UPSERT)

Envelope field	Column	Notes
`uuid.New()`	`uuid`	Generated per UPSERT call.
`candidate.vulnerability_id` (== `envelope.cve_id`)	`cveId`
`natural_key.provider`	`provider`
`natural_key.service`	`service`
`natural_key.resource_type`	`resourceType`
`candidate.vector_string`	`vectorString`
`JSON.Marshal(candidate)`	`critJSON`	Full candidate as JSON.
`time.Now().UnixMilli()`	`generatedAt`

UPSERT clause:

INSERT INTO "CritRecord" (...) VALUES (...)
ON CONFLICT ("cveId","provider","service","resourceType") DO UPDATE SET
  "vectorString" = EXCLUDED."vectorString",
  "critJSON"     = EXCLUDED."critJSON",
  "generatedAt"  = EXCLUDED."generatedAt"
WHERE "CritRecord"."generatedAt" < EXCLUDED."generatedAt"

The newer-only WHERE means: if a producer re-stages the same natural key with the same content, rowsAffected = 0 (the existing row’s generatedAt is not strictly less than the incoming one). VEX emission gates on rowsAffected > 0 so we don’t re-emit identical VEX docs.

`S3QueueObject` lifecycle

Stage	Function	Action
Stage time (producer)	`RegisterS3QueueObject`	INSERT row with `processingStatus="pending"`.
Insert success	`RemapS3QueueObjectKey` + `MarkS3QueueObjectInserted`	UPDATE key (pending→inserted prefix), set `processingStatus="inserted"`, stamp `mappedAt`.
Reject	`RemapS3QueueObjectKey` + `MarkS3QueueObjectRejected(reason)`	UPDATE key (pending→rejected prefix), set `processingStatus="rejected"`, set `errorCause=reason`.

S3 object lifecycle

Stage	Source key	Destination key
Stage	(none)	`crit-candidates/pending/{YYYY}/{MM}/{DD}/{cve}/{provider}/{service}/{rt}/{sha}.json`
Insert	`pending/...`	`crit-candidates/approved/inserted/{YYYY}/{MM}/{DD}/{cve}/{provider}/{service}/{rt}/{sha}.json`
Reject	`pending/...`	`crit-candidates/rejected/spec-violation/{YYYY}/{MM}/{DD}/{cve}/{provider}/{service}/{rt}/{sha}.json`
VEX	(companion to insert)	`{insertedKey}.vex.json`

YYYY/MM/DD are the publish run’s UTC date — not the envelope generation date — so the destination prefix groups envelopes by when they were published, not when they were staged.

Validation Suite (`Revalidate`)

Producer-side validation is best-effort; publisher-side Revalidate is authoritative.

Step	Function	Hard failure → reject
Custom dictionary schema	`ValidateCustomDictionary` (when `env.CustomDictionary != nil`)	Yes
Apply custom dictionary to registry	`WithCustomDictionary`	Yes
Record schema	`ValidateRecord` → loadRecordSchema → JSONSchema validation	Yes (schema rule)
Vector recompute	`ValidateVectorRoundTrip`	Yes (`VectorRecomputeOK = false` → spec rule failed)
Slot ABNF	`ValidateSlotABNF`	Yes (Phase-1.6 fix: regex `[A-Za-z0-9_-]+` matches the spec ABNF `1*( ALPHA / DIGIT / "-" / "_" )`)
Template format	`ValidateTemplate(provider, format, template)`	Yes
Dictionary resolution	`dictReg.Resolve(provider, service, resourceType)`	Warning only
Conditional rules	`vex_status == "fixed"` requires `provider_fix_date`; `shared_responsibility=="provider_only"` requires `existing_deployments_remain_vulnerable=false`	Yes

When Revalidate returns an error, the publisher records errorCause containing the joined SpecRules.Failed strings and moves the object to rejected/spec-violation/.

VEX Emission

Triggered by:

Successful UPSERT (rowsAffected > 0 from the newer-only WHERE).
env.Candidate.VEXStatus == "fixed".

Built via internal/critutil/vex.Build(env, time.Now()):

CycloneDX 1.6 schema.
Serial number = deterministic UUIDv5 over (cveId|provider|service|resource_type).
Components include the affected resource template and the producer/advisory metadata.
Uploaded to {insertedKey}.vex.json (the envelope’s destination key with .json swapped to .vex.json).

VEX failure (build / marshal / upload) is best-effort — logged but doesn’t roll back the upsert.

Business Rules

Drain dispatch (`publisher.go:DrainKeys`, `DrainPrefix`)

Rule	Conditional
R1 Skip drain on empty key list	`if len(keys) == 0 { return Stats{} }`
R2 Workers default to 4	`if opts.Workers <= 0 { opts.Workers = 4 }`
R3 Producer-side ops escape	`func DisabledByEnv() bool` checks `CRIT_DISABLE_INPROCESS_DRAIN ∈ {"1", "true", "yes"}`

Per-envelope processing (`publisher.go:ProcessOne`)

Rule	Conditional
R4 Get failure → outcomeError	`if err != nil { return OutcomeError }` — no S3 move, no DB update; the row stays `pending` and the operator must investigate.
R5 Parse failure → outcomeError	Same as R4 — preserves debugging surface.
R6 Validation hard-failure → outcomeRejected	`if vErr != nil { ... moveAndMark(rejected, reason) }`
R7 dryRun stops short of S3/DB writes	`if dryRun { logger.Info("would insert/reject"); return outcomeInserted/Rejected }`
R8 Upsert failure → outcomeError, NOT rejected	`if err != nil { return OutcomeError }` — DB error is transient; we want it back in pending for retry, not in rejected/.
R9 VEX emission gated on rowsAffected	`if wrote && env.Candidate.VEXStatus == "fixed" { emitVEX(...) }` — re-runs of unchanged content don’t re-emit VEX.

Envelope parsing (`publisher.go:ParseEnvelope`)

Rule	Conditional
R10 Reject mismatched envelope_version	`if env.EnvelopeVersion != critutil.EnvelopeVersion { return error }` — the publisher and producer agree on `"2"`; a mismatch indicates a schema migration that needs a publisher rebuild.

Validation (`publisher.go:Revalidate`)

Rule	Conditional
R11 Custom dict applied before validation	`if env.CustomDictionary != nil { ValidateCustomDictionary; WithCustomDictionary }` — custom dicts override extended which overrides spec.
R12 Vector mismatch is a hard failure	`if !val.VectorRecomputeOK { return val, fmt.Errorf("vector recompute mismatch") }` — vectors must round-trip identically; mismatch indicates the producer wrote a stale or fabricated vector.
R13 Spec rule failures concatenated as errorCause	`return val, fmt.Errorf("spec rules failed: %s", strings.Join(val.SpecRules.Failed, "; "))` — populates the `errorCause` field on the rejection row.

S3 move (`publisher.go:moveAndMark`)

Rule	Conditional
R14 S3 move failure aborts the mark	`if err := uploader.MoveObject(...); err != nil { return }` — best-effort; the row remains in the source state, retry on next run.
R15 RemapS3QueueObjectKey before status update	If `pending → approved/inserted` move succeeds but the rename in DB fails, the row is still pending in DB — the next publisher run will retry. Idempotent on natural key.

Insert SQL (`publisher.go:upsert`)

Rule	Conditional
R16 Newer-only WHERE	`WHERE "CritRecord"."generatedAt" < EXCLUDED."generatedAt"` — protects against producers re-staging out-of-order.
R17 Per-row UUID generation	`uuid.New().String()` — every UPSERT call generates a fresh UUID for the would-be insert; the conflict path uses EXCLUDED columns so the existing UUID is preserved.
R18 Returns `wrote = tag.RowsAffected() > 0`	Used by R9 (VEX emission gate).

Path construction

Rule	Function
R19 Inserted-prefix key includes today’s UTC date	`insertedKey(env, srcKey)` builds `approved/inserted/{YYYY}/{MM}/{DD}/{cve}/{provider}/{service}/{rt}/{sha}` from `time.Now().UTC()`. The original `pending/...` date is discarded.
R20 Rejected-prefix key fills “unknown” for empty natural-key fields	`rejectedKey(env, srcKey)` defensively replaces empty `provider`/`service`/`resourceType`/`cveId` with `"unknown"` so the path is well-formed even for malformed envelopes.
R21 VEX key derived by suffix swap	`vexKeyFor(envelopeKey)` returns `strings.TrimSuffix(envelopeKey, ".json") + ".vex.json"`.

CLI (`cmd/crit-publisher/main.go`)

The binary is a thin wrapper around critpublisher.DrainPrefix + UpsertEnvelope. Flags:

--envelope — fixture mode (one envelope from disk; bypasses S3, calls UpsertEnvelope directly).
--prefix — list mode (S3 prefix scan; calls DrainPrefix).
--bucket — overrides S3_BUCKET_NAME.
--workers — concurrency (default 4).
--limit — cap items processed in list mode.
--dry-run — validate without inserting / moving.

Mutual exclusion: exactly one of --envelope or --prefix.

Loaded dictionary stack: spec → extended (no custom layer at the binary level; custom dicts ride per-envelope and merge in Revalidate).

Verification Queries

-- CritRecord count by provider
SELECT provider, count(*) FROM "CritRecord" GROUP BY 1 ORDER BY 2 DESC;

-- Pending envelopes (should be empty after producers run with in-process drain)
SELECT source, count(*) FROM "S3QueueObject"
 WHERE "processingStatus"='pending' GROUP BY 1;

-- Rejection reasons (worth periodically auditing)
SELECT "errorCause", count(*) FROM "S3QueueObject"
 WHERE "processingStatus"='rejected' GROUP BY 1 ORDER BY 2 DESC LIMIT 10;

-- Verify newer-only enforcement: no two CritRecord rows have identical natural keys
SELECT "cveId", provider, service, "resourceType", count(*)
  FROM "CritRecord"
 GROUP BY 1, 2, 3, 4
HAVING count(*) > 1;
-- expected: 0 rows (UNIQUE constraint enforced by ON CONFLICT clause)

# VEX coverage — fixed CritRecords should each have a sibling .vex.json
aws s3 ls s3://vdb-manager-artifacts/crit-candidates/approved/inserted/ \
   --recursive | grep -c '\.vex\.json$'

Risk Surface

Risk	Guard
Validation rule drift between producer and publisher	`Revalidate` is authoritative — runs at publish time even if the producer’s `validation.spec_rules.failed` is empty.
S3 move-then-mark race	RemapS3QueueObjectKey is idempotent; if the move succeeds and the mark fails, next run re-marks (no orphan).
VEX sibling write failure	Best-effort — logged but never rolls back the upsert. Re-run with `--force` on the producer to re-emit.
`generated_at` clock skew across producers	Each producer uses `time.Now().UnixMilli()`; skew is bounded by ECS clock sync. Newer-only WHERE handles arbitrary clock order — older `generated_at` values just don’t UPDATE.
Schema enum drift (e.g. `template_format` not in allowed list)	Caught by `Revalidate` → `ValidateTemplate` → schema rule failure → rejected with errorCause.
Stale envelopes from a deprecated producer revision	`envelope_version` mismatch rejects at parse time (R10).
Custom-dictionary entries violate dict schema	`ValidateCustomDictionary` rejects before any record-validation step.

S3 Persistence

Not used. This processor does not currently archive payloads or quarantine failures to S3. Per the S3 Persistence Contract this is non-compliant — see the compliance matrix for the implementation roadmap.

⚠ Not in the compliance matrix — status needs verification.

Expected paths when implemented:

Archive: crit-publisher/files/{sha256}/{filename}
Quarantine: failed-feeds/crit-publisher/{YYYY-MM-DD}/{reason}/{filename}
Likely reasons: (none documented)