Files
Agent-System/internal/artifactvault/registry.go
T
weichen 25fc2d92e8 fix(artifactvault): derive staging/artifact UUIDs from idempotency key
Payload-sha-only UUIDs collided under artifact_payload_staging_pkey when
two task runs finalized identical deliverable bytes. Key already includes
source_event_id; hash it for stable unique art-/stg- ids.
2026-09-24 00:56:54 +00:00

475 lines
17 KiB
Go

package artifactvault
import (
"context"
"fmt"
"sync"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
type Registry interface {
Request(ctx context.Context, facts Facts, req PublishRequest) (*Artifact, *Staging, error)
LoadRequested(ctx context.Context) ([]Artifact, error)
LoadStaging(ctx context.Context, sourceEventID string) (*Staging, error)
MarkPublishing(ctx context.Context, uuid string) error
MarkVerified(ctx context.Context, uuid, manifestSHA string) error
MarkFailed(ctx context.Context, uuid, code string) error
GetByLogical(ctx context.Context, env, logical string) (*Artifact, error)
Tombstone(ctx context.Context, uuid, reason, authority string) error
IsDenied(ctx context.Context, uuid string) bool
RecordReference(ctx context.Context, uuid, projectKey, sha, keyRef string) error
ActiveRefs(ctx context.Context, projectKey, sha string) int
Quarantine(ctx context.Context, reason string, logical string) error
QuarantineCount(ctx context.Context) int
Coverage(ctx context.Context) Coverage
}
type Coverage struct {
Required int
Verified int
Quarantine int
Excluded int
}
type MemoryRegistry struct {
mu sync.Mutex
arts map[string]*Artifact
byKey map[string]string
staging map[string]*Staging
denied map[string]bool
refs map[string]int
quarantine int
required int
}
func NewMemoryRegistry() *MemoryRegistry {
return &MemoryRegistry{
arts: map[string]*Artifact{}, byKey: map[string]string{}, staging: map[string]*Staging{},
denied: map[string]bool{}, refs: map[string]int{},
}
}
func (m *MemoryRegistry) Request(_ context.Context, facts Facts, req PublishRequest) (*Artifact, *Staging, error) {
if err := EvaluateEligibility(facts); err != nil {
return nil, nil, err
}
if err := ValidateEnv(req.Environment); err != nil {
return nil, nil, err
}
sha := SHA256Hex(req.Payload)
key := IdempotencyKey(req, sha)
m.mu.Lock()
defer m.mu.Unlock()
if id, ok := m.byKey[key]; ok {
return m.arts[id], m.staging[req.SourceEventID], nil
}
uuid := "art-" + sha[:16] + "-" + fmt.Sprintf("%d", len(m.arts)+1)
art := &Artifact{
UUID: uuid, IdempotencyKey: key, ArtifactType: req.ArtifactType,
LogicalResourceID: req.LogicalResourceID, Environment: req.Environment,
ProjectID: req.ProjectID, ProjectExternalKey: req.ProjectExternalKey,
SourceEventID: req.SourceEventID, SourceEventKind: req.SourceEventKind,
SourceSnapshotID: req.SourceSnapshotID, SourceMessageID: req.SourceMessageID,
SourceRunID: req.SourceRunID, PayloadSHA256: sha, TerminalState: req.TerminalState,
PolicyVersion: nz(req.PolicyVersion, PolicyVersion), PublicationStatus: StatusRequested,
RetentionClass: nz(req.RetentionClass, "governance"), RequestedAt: time.Now().UTC(),
}
st := &Staging{
UUID: "stg-" + uuid, ProjectID: req.ProjectID, SourceEventID: req.SourceEventID,
PayloadSHA256: sha, Payload: append([]byte(nil), req.Payload...), Classification: ClassificationOK,
CreatedAt: time.Now().UTC(),
}
m.arts[uuid] = art
m.byKey[key] = uuid
m.staging[req.SourceEventID] = st
m.required++
return art, st, nil
}
func (m *MemoryRegistry) LoadRequested(_ context.Context) ([]Artifact, error) {
m.mu.Lock()
defer m.mu.Unlock()
out := make([]Artifact, 0)
for _, a := range m.arts {
if a.PublicationStatus == StatusRequested || a.PublicationStatus == StatusRetryableFailed {
out = append(out, *a)
}
}
return out, nil
}
func (m *MemoryRegistry) LoadStaging(_ context.Context, sourceEventID string) (*Staging, error) {
m.mu.Lock()
defer m.mu.Unlock()
st := m.staging[sourceEventID]
if st == nil {
return nil, fmt.Errorf("artifactvault: staging missing")
}
cp := *st
cp.Payload = append([]byte(nil), st.Payload...)
return &cp, nil
}
func (m *MemoryRegistry) MarkPublishing(_ context.Context, uuid string) error {
m.mu.Lock()
defer m.mu.Unlock()
if a := m.arts[uuid]; a != nil {
a.PublicationStatus = StatusPublishing
}
return nil
}
func (m *MemoryRegistry) MarkVerified(_ context.Context, uuid, manifestSHA string) error {
m.mu.Lock()
defer m.mu.Unlock()
if a := m.arts[uuid]; a != nil {
now := time.Now().UTC()
a.PublicationStatus = StatusVerified
a.ManifestSHA256 = manifestSHA
a.VerifiedAt = &now
}
return nil
}
func (m *MemoryRegistry) MarkFailed(_ context.Context, uuid, _ string) error {
m.mu.Lock()
defer m.mu.Unlock()
if a := m.arts[uuid]; a != nil {
a.PublicationStatus = StatusRetryableFailed
}
return nil
}
func (m *MemoryRegistry) GetByLogical(_ context.Context, env, logical string) (*Artifact, error) {
m.mu.Lock()
defer m.mu.Unlock()
for _, a := range m.arts {
if a.Environment == env && a.LogicalResourceID == logical && a.PublicationStatus == StatusVerified && !m.denied[a.UUID] {
cp := *a
return &cp, nil
}
}
return nil, fmt.Errorf("artifactvault: not found")
}
func (m *MemoryRegistry) Tombstone(_ context.Context, uuid, _, _ string) error {
m.mu.Lock()
defer m.mu.Unlock()
m.denied[uuid] = true
if a := m.arts[uuid]; a != nil {
now := time.Now().UTC()
a.PublicationStatus = StatusTombstoned
a.TombstonedAt = &now
}
return nil
}
func (m *MemoryRegistry) IsDenied(_ context.Context, uuid string) bool {
m.mu.Lock()
defer m.mu.Unlock()
return m.denied[uuid]
}
func (m *MemoryRegistry) RecordReference(_ context.Context, _, projectKey, sha, _ string) error {
m.mu.Lock()
defer m.mu.Unlock()
m.refs[projectKey+"|"+sha]++
return nil
}
func (m *MemoryRegistry) ActiveRefs(_ context.Context, projectKey, sha string) int {
m.mu.Lock()
defer m.mu.Unlock()
return m.refs[projectKey+"|"+sha]
}
func (m *MemoryRegistry) Quarantine(_ context.Context, _, _ string) error {
m.mu.Lock()
defer m.mu.Unlock()
m.quarantine++
return nil
}
func (m *MemoryRegistry) QuarantineCount(_ context.Context) int {
m.mu.Lock()
defer m.mu.Unlock()
return m.quarantine
}
func (m *MemoryRegistry) Coverage(_ context.Context) Coverage {
m.mu.Lock()
defer m.mu.Unlock()
v := 0
for _, a := range m.arts {
if a.PublicationStatus == StatusVerified {
v++
}
}
return Coverage{Required: m.required, Verified: v, Quarantine: m.quarantine}
}
// PGRegistry persists FAV rows. Request uses a dedicated transaction unless WithTx is used.
type PGRegistry struct {
pool *pgxpool.Pool
}
func NewPGRegistry(pool *pgxpool.Pool) *PGRegistry { return &PGRegistry{pool: pool} }
func (p *PGRegistry) Request(ctx context.Context, facts Facts, req PublishRequest) (*Artifact, *Staging, error) {
if p == nil || p.pool == nil {
return nil, nil, fmt.Errorf("artifactvault: no database")
}
tx, err := p.pool.BeginTx(ctx, pgx.TxOptions{})
if err != nil {
return nil, nil, err
}
defer func() { _ = tx.Rollback(ctx) }()
art, st, err := requestTX(ctx, tx, facts, req)
if err != nil {
return nil, nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, nil, err
}
return art, st, nil
}
func requestTX(ctx context.Context, tx pgx.Tx, facts Facts, req PublishRequest) (*Artifact, *Staging, error) {
if err := EvaluateEligibility(facts); err != nil {
return nil, nil, err
}
sha := SHA256Hex(req.Payload)
key := IdempotencyKey(req, sha)
var existing string
err := tx.QueryRow(ctx, `SELECT artifact_uuid FROM final_artifacts WHERE idempotency_key=$1`, key).Scan(&existing)
if err == nil && existing != "" {
st := &Staging{SourceEventID: req.SourceEventID, PayloadSHA256: sha, Payload: req.Payload}
return &Artifact{UUID: existing, IdempotencyKey: key, PublicationStatus: StatusRequested}, st, nil
}
// Derive UUIDs from the full idempotency key (includes source_event_id + sha).
// Using payload sha alone collides when two events share identical bytes.
uuid, stgUUID := artifactIDsFromKey(key)
_, err = tx.Exec(ctx, `
INSERT INTO artifact_payload_staging (staging_uuid, project_id, source_event_id, payload_sha256, encrypted_payload, classification)
VALUES ($1,$2,$3,$4,$5,$6)
ON CONFLICT (source_event_id, payload_sha256) DO NOTHING
`, stgUUID, req.ProjectID, req.SourceEventID, sha, req.Payload, ClassificationOK)
if err != nil {
return nil, nil, err
}
_, err = tx.Exec(ctx, `
INSERT INTO final_artifacts (
artifact_uuid, idempotency_key, artifact_type, logical_resource_id, environment, project_id,
project_external_key, source_event_id, source_event_kind, source_snapshot_id, source_message_id,
source_run_id, payload_sha256, terminal_state, policy_version, publication_status, retention_class
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)
ON CONFLICT (idempotency_key) DO NOTHING
`, uuid, key, req.ArtifactType, req.LogicalResourceID, req.Environment, req.ProjectID,
req.ProjectExternalKey, req.SourceEventID, req.SourceEventKind, nullInt(req.SourceSnapshotID),
nullInt(req.SourceMessageID), req.SourceRunID, sha, req.TerminalState, nz(req.PolicyVersion, PolicyVersion),
StatusRequested, nz(req.RetentionClass, "governance"))
if err != nil {
return nil, nil, err
}
return &Artifact{UUID: uuid, IdempotencyKey: key, ArtifactType: req.ArtifactType, LogicalResourceID: req.LogicalResourceID,
Environment: req.Environment, ProjectExternalKey: req.ProjectExternalKey, SourceEventID: req.SourceEventID,
PayloadSHA256: sha, PublicationStatus: StatusRequested, TerminalState: req.TerminalState}, &Staging{Payload: req.Payload, PayloadSHA256: sha, SourceEventID: req.SourceEventID}, nil
}
func artifactIDsFromKey(key string) (artUUID, stgUUID string) {
idHash := SHA256Hex([]byte(key))[:16]
return "art-" + idHash, "stg-" + idHash
}
func nullInt(v int64) any {
if v == 0 {
return nil
}
return v
}
func (p *PGRegistry) LoadRequested(ctx context.Context) ([]Artifact, error) {
if p == nil || p.pool == nil {
return nil, fmt.Errorf("artifactvault: no database")
}
rows, err := p.pool.Query(ctx, `
SELECT artifact_uuid, idempotency_key, artifact_type, logical_resource_id, environment,
COALESCE(project_id,0), project_external_key, source_event_id, source_event_kind,
COALESCE(source_snapshot_id,0), COALESCE(source_message_id,0), source_run_id,
payload_sha256, COALESCE(manifest_sha256,''), COALESCE(vault_storage_identity,''),
terminal_state, policy_version, publication_status, retention_class, requested_at
FROM final_artifacts
WHERE publication_status IN ('requested','retryable_failed')
ORDER BY requested_at ASC`)
if err != nil {
return nil, err
}
defer rows.Close()
out := make([]Artifact, 0)
for rows.Next() {
var a Artifact
if err := rows.Scan(&a.UUID, &a.IdempotencyKey, &a.ArtifactType, &a.LogicalResourceID, &a.Environment,
&a.ProjectID, &a.ProjectExternalKey, &a.SourceEventID, &a.SourceEventKind,
&a.SourceSnapshotID, &a.SourceMessageID, &a.SourceRunID, &a.PayloadSHA256, &a.ManifestSHA256,
&a.VaultStorageIdentity, &a.TerminalState, &a.PolicyVersion, &a.PublicationStatus, &a.RetentionClass, &a.RequestedAt); err != nil {
return nil, err
}
out = append(out, a)
}
return out, rows.Err()
}
func (p *PGRegistry) LoadStaging(ctx context.Context, sourceEventID string) (*Staging, error) {
if p == nil || p.pool == nil {
return nil, fmt.Errorf("artifactvault: no database")
}
var st Staging
err := p.pool.QueryRow(ctx, `
SELECT staging_uuid, COALESCE(project_id,0), source_event_id, payload_sha256, encrypted_payload, classification, created_at
FROM artifact_payload_staging WHERE source_event_id=$1 ORDER BY created_at DESC LIMIT 1`, sourceEventID).
Scan(&st.UUID, &st.ProjectID, &st.SourceEventID, &st.PayloadSHA256, &st.Payload, &st.Classification, &st.CreatedAt)
if err != nil {
return nil, err
}
return &st, nil
}
func (p *PGRegistry) MarkPublishing(ctx context.Context, uuid string) error {
if p == nil || p.pool == nil {
return nil
}
_, err := p.pool.Exec(ctx, `UPDATE final_artifacts SET publication_status=$2 WHERE artifact_uuid=$1`, uuid, StatusPublishing)
if err != nil {
return err
}
_, err = p.pool.Exec(ctx, `
INSERT INTO artifact_publication_attempts (artifact_uuid, idempotency_key, attempt_no)
SELECT artifact_uuid, idempotency_key, COALESCE((SELECT MAX(attempt_no) FROM artifact_publication_attempts a WHERE a.artifact_uuid=$1),0)+1
FROM final_artifacts WHERE artifact_uuid=$1`, uuid)
return err
}
func (p *PGRegistry) MarkVerified(ctx context.Context, uuid, manifestSHA string) error {
if p == nil || p.pool == nil {
return nil
}
_, err := p.pool.Exec(ctx, `UPDATE final_artifacts SET publication_status=$2, manifest_sha256=$3, verified_at=now() WHERE artifact_uuid=$1`, uuid, StatusVerified, manifestSHA)
return err
}
func (p *PGRegistry) MarkFailed(ctx context.Context, uuid, code string) error {
if p == nil || p.pool == nil {
return nil
}
_, err := p.pool.Exec(ctx, `UPDATE final_artifacts SET publication_status=$2 WHERE artifact_uuid=$1`, uuid, StatusRetryableFailed)
if err != nil {
return err
}
_, _ = p.pool.Exec(ctx, `
UPDATE artifact_publication_attempts SET error_code=$2
WHERE artifact_uuid=$1 AND attempt_no=(SELECT MAX(attempt_no) FROM artifact_publication_attempts WHERE artifact_uuid=$1)`, uuid, code)
return nil
}
func (p *PGRegistry) GetByLogical(ctx context.Context, env, logical string) (*Artifact, error) {
if p == nil || p.pool == nil {
return nil, fmt.Errorf("artifactvault: no database")
}
var a Artifact
err := p.pool.QueryRow(ctx, `
SELECT artifact_uuid, idempotency_key, artifact_type, logical_resource_id, environment,
COALESCE(project_id,0), project_external_key, source_event_id, source_event_kind,
COALESCE(source_snapshot_id,0), COALESCE(source_message_id,0), source_run_id,
payload_sha256, COALESCE(manifest_sha256,''), COALESCE(vault_storage_identity,''),
terminal_state, policy_version, publication_status
FROM final_artifacts
WHERE environment=$1 AND logical_resource_id=$2 AND publication_status='verified'
ORDER BY verified_at DESC NULLS LAST LIMIT 1`, env, logical).
Scan(&a.UUID, &a.IdempotencyKey, &a.ArtifactType, &a.LogicalResourceID, &a.Environment,
&a.ProjectID, &a.ProjectExternalKey, &a.SourceEventID, &a.SourceEventKind,
&a.SourceSnapshotID, &a.SourceMessageID, &a.SourceRunID, &a.PayloadSHA256, &a.ManifestSHA256,
&a.VaultStorageIdentity, &a.TerminalState, &a.PolicyVersion, &a.PublicationStatus)
if err != nil {
return nil, err
}
return &a, nil
}
func (p *PGRegistry) Tombstone(ctx context.Context, uuid, reason, authority string) error {
if p == nil || p.pool == nil {
return nil
}
_, err := p.pool.Exec(ctx, `UPDATE final_artifacts SET publication_status=$2, tombstoned_at=now() WHERE artifact_uuid=$1`, uuid, StatusTombstoned)
if err != nil {
return err
}
_, err = p.pool.Exec(ctx, `INSERT INTO artifact_deletion_ledger (artifact_uuid, reason, authority, effective_at) VALUES ($1,$2,$3,now())`, uuid, reason, authority)
return err
}
func (p *PGRegistry) IsDenied(ctx context.Context, uuid string) bool {
if p == nil || p.pool == nil {
return false
}
var n int
_ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_deletion_ledger WHERE artifact_uuid=$1`, uuid).Scan(&n)
return n > 0
}
func (p *PGRegistry) RecordReference(ctx context.Context, uuid, projectKey, sha, keyRef string) error {
if p == nil || p.pool == nil {
return nil
}
_, err := p.pool.Exec(ctx, `
INSERT INTO artifact_blob_references (artifact_uuid, project_scope, payload_sha256, envelope_key_ref, reference_status)
VALUES ($1,$2,$3,$4,'active')
ON CONFLICT (artifact_uuid, payload_sha256) DO NOTHING`, uuid, projectKey, sha, keyRef)
return err
}
func (p *PGRegistry) ActiveRefs(ctx context.Context, projectKey, sha string) int {
if p == nil || p.pool == nil {
return 0
}
var n int
_ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_blob_references WHERE project_scope=$1 AND payload_sha256=$2 AND reference_status='active'`, projectKey, sha).Scan(&n)
return n
}
func (p *PGRegistry) Quarantine(ctx context.Context, reason, logical string) error {
if p == nil || p.pool == nil {
return nil
}
_, err := p.pool.Exec(ctx, `INSERT INTO artifact_quarantine (logical_resource_id, reason) VALUES ($1,$2)`, logical, reason)
return err
}
func (p *PGRegistry) QuarantineCount(ctx context.Context) int {
if p == nil || p.pool == nil {
return 0
}
var n int
_ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_quarantine`).Scan(&n)
return n
}
func (p *PGRegistry) Coverage(ctx context.Context) Coverage {
var c Coverage
if p == nil || p.pool == nil {
return c
}
_ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM final_artifacts`).Scan(&c.Required)
_ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM final_artifacts WHERE publication_status='verified'`).Scan(&c.Verified)
_ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_quarantine`).Scan(&c.Quarantine)
_ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_backfill_exclusions WHERE expires_at > now()`).Scan(&c.Excluded)
return c
}
// RequestWithTx writes staging + unpublished outbox inside an existing domain transaction.
func RequestWithTx(ctx context.Context, tx pgx.Tx, facts Facts, req PublishRequest) (*Artifact, *Staging, error) {
return requestTX(ctx, tx, facts, req)
}