Skip to content

Distributed Workers Data Model

Status: proposed data model RFC v1.

This document defines the proposed persistence model for Distributed Workers. It is documentation only. No migrations have been created.

Status labels

  • CURRENT: implemented in the repository today.
  • ACCEPTED DESIGN: already agreed in the Distributed Workers ADRs.
  • PROPOSED: recommended design for future implementation.
  • UNRESOLVED: requires a later implementation or product decision.
  • FUTURE: outside initial scanner Worker implementation.

Verified existing schema

CURRENT: The repository already has durable scan, scan job, schedule, asset, threat, file, and history tables.

Important source references:

  • scans and scan_jobs: backend/migrations/001_initial_schema.sql
  • scheduled_scans: backend/migrations/002_scheduled_scans.sql
  • files: backend/migrations/007_files.sql
  • exposure_changes: backend/migrations/008_exposure_changes.sql
  • threats and vulnerability_details: backend/migrations/020_threat_model.sql
  • asset_scan_snapshots and asset_scan_changes: backend/migrations/025_asset_scan_history.sql
  • scans.asset_targets: backend/migrations/026_scan_asset_targets.sql
  • threat_detections: backend/migrations/028_threat_detections.sql
  • threat_exclusions: backend/migrations/029_threat_exclusions.sql and backend/migrations/030_threat_exclusion_identity.sql

CURRENT Go models:

  • Scan, ScanJob, ScheduledScan: backend/internal/scans/model.go
  • scan repository methods: backend/internal/scans/repository.go
  • plugin input and result types: backend/internal/plugins/model.go
  • file model and save input: backend/internal/files/model.go
  • threat service and repository: backend/internal/vulnerabilities
  • asset service and repository: backend/internal/assets

Existing tables that should be reused

Existing table Reuse in distributed architecture
organizations Source of organization ownership.
organization_members User access model; not Worker authorization.
scans User-visible scan lifecycle and aggregate state.
scan_jobs User-visible plugin/job execution state, inputs, outputs, errors.
scheduled_scans Existing recurring scan definition. Needs safer claiming, not replacement.
assets Existing asset canonical store and deduplication.
asset_edges Existing graph relation store.
threats Existing threat store and deduplication target.
vulnerability_details Existing vulnerability-specific detail table.
threat_detections Existing scan/job detection history.
threat_exclusions Existing suppression rules.
files Existing artifact metadata.
exposure_changes Existing exposure-change timeline.
asset_scan_snapshots / asset_scan_changes Existing asset observation history.

Existing schema insufficiencies

CURRENT schema does not include:

  • Worker identities;
  • Worker certificate identity or revocation state;
  • organization-to-Worker assignments;
  • task leases;
  • explicit attempts;
  • stale-attempt fencing;
  • result-batch idempotency;
  • artifact upload sessions for Workers without S3 credentials.

Therefore, the Distributed Worker implementation needs additive tables. Existing scan/job tables should remain the user-visible scan execution model.

Proposed entity relationship diagram

PROPOSED:

erDiagram
    ORGANIZATIONS ||--o{ SCANS : owns
    SCANS ||--o{ SCAN_JOBS : contains
    SCAN_JOBS ||--o{ WORKER_TASKS : executes_as

    WORKERS ||--o{ ORGANIZATION_WORKERS : assigned
    ORGANIZATIONS ||--o{ ORGANIZATION_WORKERS : uses

    WORKER_TASKS ||--o{ WORKER_TASK_ATTEMPTS : attempts
    WORKERS ||--o{ WORKER_TASK_ATTEMPTS : owns
    WORKER_TASKS ||--o{ WORKER_RESULT_BATCHES : accepts
    WORKER_TASK_ATTEMPTS ||--o{ WORKER_RESULT_BATCHES : submits
    WORKER_TASKS ||--o{ WORKER_ARTIFACT_UPLOADS : uploads

Worker identity

PROPOSED table: workers.

Purpose: persistent identity and policy state for a Worker. Connection state remains ephemeral and is derived from heartbeats and active leases.

Illustrative SQL:

CREATE TABLE workers (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    name TEXT NOT NULL,
    worker_type TEXT NOT NULL,
    enabled BOOLEAN NOT NULL DEFAULT true,
    revoked_at TIMESTAMP NULL,
    revocation_reason TEXT NOT NULL DEFAULT '',
    certificate_subject TEXT NOT NULL DEFAULT '',
    certificate_fingerprint TEXT NOT NULL DEFAULT '',
    software_version TEXT NOT NULL DEFAULT '',
    protocol_version TEXT NOT NULL DEFAULT '',
    capabilities JSONB NOT NULL DEFAULT '{}'::jsonb,
    max_concurrent_tasks INTEGER NOT NULL DEFAULT 1,
    last_seen_at TIMESTAMP NULL,
    metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
    created_by UUID NULL REFERENCES users(id) ON DELETE SET NULL,
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
    CONSTRAINT workers_type_check CHECK (worker_type IN ('scanner', 'ai')),
    CONSTRAINT workers_capacity_positive CHECK (max_concurrent_tasks > 0)
);

CREATE UNIQUE INDEX workers_certificate_fingerprint_unique
    ON workers(certificate_fingerprint)
    WHERE certificate_fingerprint <> '';

CREATE INDEX workers_enabled_type_idx
    ON workers(worker_type, enabled)
    WHERE revoked_at IS NULL;

CREATE INDEX workers_last_seen_idx
    ON workers(last_seen_at DESC);

Rationale:

  • enabled controls normal administrative availability.
  • revoked_at represents security revocation and should prevent all new leases.
  • capabilities stores last centrally accepted capability advertisement, not authorization.
  • last_seen_at is enough for v1 availability; a heartbeat history table is not required unless product needs historical availability analytics.
  • Private keys must never be stored in PostgreSQL.

Multiple Gateway replicas observe the same Worker state through PostgreSQL. They may cache capabilities for display, but task acquisition must re-check Worker and assignment state transactionally.

Organization-to-Worker assignments

ACCEPTED DESIGN: the relationship is many-to-many.

PROPOSED table: organization_workers.

Illustrative SQL:

CREATE TABLE organization_workers (
    organization_id UUID NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
    worker_id UUID NOT NULL REFERENCES workers(id) ON DELETE CASCADE,
    enabled BOOLEAN NOT NULL DEFAULT true,
    assigned_by UUID NULL REFERENCES users(id) ON DELETE SET NULL,
    assigned_at TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
    revoked_at TIMESTAMP NULL,
    revoked_by UUID NULL REFERENCES users(id) ON DELETE SET NULL,
    revocation_reason TEXT NOT NULL DEFAULT '',
    PRIMARY KEY (organization_id, worker_id)
);

CREATE INDEX organization_workers_worker_enabled_idx
    ON organization_workers(worker_id, enabled, organization_id)
    WHERE revoked_at IS NULL;

CREATE INDEX organization_workers_org_enabled_idx
    ON organization_workers(organization_id, enabled, worker_id)
    WHERE revoked_at IS NULL;

Soft revocation is recommended instead of hard delete for auditability and race-safe decisions. A hard delete can be allowed later only if audit logs capture the previous assignment.

Assignment scenarios:

Scenario Behavior
One organization assigned to one Worker Only that Worker is eligible if it has required capabilities and capacity.
One organization assigned to multiple Workers Gateway may lease to any eligible assigned Worker.
One Worker assigned to multiple organizations Worker may receive tasks for any assigned organization, one task at a time or up to capacity.
Worker with no assignments No organization-scoped scanner tasks are leased.
Disabled Worker No new tasks are leased; active tasks follow cancellation policy.
Assignment removed during execution New leases stop immediately. Active leases should be cancelled by default for security revocation, configurable for routine reassignment.
Temporarily offline Worker No new leases; expired leases become retryable.
Organization with no eligible online Workers Tasks remain pending or retry-wait; UI should show no eligible Worker.
Capability changes Future leases use newly advertised and persisted capabilities; active leases continue unless capability loss is marked critical.
Concurrent assignment change and acquisition Acquisition transaction must lock/check assignment before creating the attempt.

Assignment eligibility is not the same as target-scope authorization. Assignment says which organization a Worker may serve. Scan scope rules still come from existing scan, scope, asset approval, and plugin configuration logic.

Task granularity and scan relationships

PROPOSED: a durable Worker task represents one scanner plugin execution, backed by one scan_jobs row.

Why not an entire scan:

  • current processScan resolves scope, creates jobs, runs plugins sequentially, enriches scope, and persists results in one Worker process;
  • an entire-scan remote task would keep too much DB-backed orchestration and result ingestion on the Worker;
  • cancellation/retry is already exposed at job level for manual retries.

Why not only extend scan_jobs:

  • scan_jobs is user-visible execution history;
  • lease ownership, attempts, idempotency batches, Worker assignment, and future AI task support are infrastructure concerns;
  • adding all lease fields to scan_jobs would make scan job state and transport state hard to evolve.

Recommended hybrid:

  • scan_jobs remains the user-visible plugin job.
  • worker_tasks is the durable leasing envelope.
  • one scanner worker_tasks row references one scan_jobs row.
  • future AI tasks can use worker_tasks without pretending to be scan jobs.

Tasks should be created lazily as the scan orchestrator reaches each plugin. This preserves the current sequential profile behavior and runtime scope enrichment. Eager creation is only safe later if plugin dependencies are explicitly modeled.

Durable task schema

PROPOSED table: worker_tasks.

Illustrative SQL:

CREATE TABLE worker_tasks (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    task_type TEXT NOT NULL,
    organization_id UUID NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
    scan_id UUID NULL REFERENCES scans(id) ON DELETE CASCADE,
    scan_job_id UUID NULL REFERENCES scan_jobs(id) ON DELETE CASCADE,
    required_capabilities JSONB NOT NULL DEFAULT '{}'::jsonb,
    payload JSONB NOT NULL DEFAULT '{}'::jsonb,
    state TEXT NOT NULL DEFAULT 'pending',
    priority INTEGER NOT NULL DEFAULT 0,
    available_at TIMESTAMP NOT NULL DEFAULT NOW(),
    execution_deadline_at TIMESTAMP NULL,
    attempt_count INTEGER NOT NULL DEFAULT 0,
    max_attempts INTEGER NOT NULL DEFAULT 3,
    cancel_requested_at TIMESTAMP NULL,
    completed_at TIMESTAMP NULL,
    last_error TEXT NOT NULL DEFAULT '',
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
    CONSTRAINT worker_tasks_state_check CHECK (
        state IN ('pending', 'leased', 'running', 'cancelling', 'cancelled', 'completed', 'retry_wait', 'failed_terminal')
    ),
    CONSTRAINT worker_tasks_type_check CHECK (task_type IN ('scanner_plugin', 'ai')),
    CONSTRAINT worker_tasks_attempts_nonnegative CHECK (attempt_count >= 0),
    CONSTRAINT worker_tasks_max_attempts_positive CHECK (max_attempts > 0)
);

CREATE UNIQUE INDEX worker_tasks_scan_job_unique
    ON worker_tasks(scan_job_id)
    WHERE scan_job_id IS NOT NULL AND task_type = 'scanner_plugin';

CREATE INDEX worker_tasks_acquire_idx
    ON worker_tasks(state, available_at, priority DESC, created_at)
    WHERE state IN ('pending', 'retry_wait');

CREATE INDEX worker_tasks_org_state_idx
    ON worker_tasks(organization_id, state);

CREATE INDEX worker_tasks_scan_idx
    ON worker_tasks(scan_id);

Notes:

  • payload should contain the task input snapshot required by the Worker, such as plugin name, PluginInput, timeout, and rate limit.
  • For scanner tasks, the canonical user-visible job remains scan_jobs.
  • execution_deadline_at is a hard maximum for task runtime and is separate from the renewable lease.

Attempts and leases

PROPOSED table: worker_task_attempts.

Illustrative SQL:

CREATE TABLE worker_task_attempts (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    task_id UUID NOT NULL REFERENCES worker_tasks(id) ON DELETE CASCADE,
    worker_id UUID NOT NULL REFERENCES workers(id) ON DELETE RESTRICT,
    attempt_number INTEGER NOT NULL,
    lease_token TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'leased',
    lease_expires_at TIMESTAMP NOT NULL,
    started_at TIMESTAMP NOT NULL DEFAULT NOW(),
    finished_at TIMESTAMP NULL,
    error_code TEXT NOT NULL DEFAULT '',
    error_message TEXT NOT NULL DEFAULT '',
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
    CONSTRAINT worker_task_attempts_status_check CHECK (
        status IN ('leased', 'running', 'succeeded', 'failed', 'cancelled', 'expired', 'superseded')
    ),
    UNIQUE (task_id, attempt_number),
    UNIQUE (lease_token)
);

CREATE INDEX worker_task_attempts_active_idx
    ON worker_task_attempts(task_id, lease_expires_at)
    WHERE status IN ('leased', 'running');

CREATE INDEX worker_task_attempts_worker_active_idx
    ON worker_task_attempts(worker_id, status)
    WHERE status IN ('leased', 'running');

Rationale:

  • Separate attempts preserve execution history and make retries explicit.
  • The lease_token is a fencing token. Mutating RPCs must include it.
  • attempt_number gives a simple monotonic task-local generation.
  • Lease expiration and execution deadline are distinct.

Result batch acceptance

CURRENT idempotency is partial:

  • asset upsert is idempotent by (organization_id, type, normalized_value);
  • threat upsert is idempotent for non-empty external IDs by (organization_id, asset_id, external_id);
  • threat detections are idempotent per (threat_id, scan_job_id);
  • exposure changes use an application-level similarity check;
  • file metadata and artifact upload are not fully idempotent across retry boundaries.

PROPOSED table: worker_result_batches.

CREATE TABLE worker_result_batches (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    task_id UUID NOT NULL REFERENCES worker_tasks(id) ON DELETE CASCADE,
    attempt_id UUID NOT NULL REFERENCES worker_task_attempts(id) ON DELETE CASCADE,
    sequence_number INTEGER NOT NULL,
    idempotency_key TEXT NOT NULL,
    payload_sha256 TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'accepted',
    received_at TIMESTAMP NOT NULL DEFAULT NOW(),
    persisted_at TIMESTAMP NULL,
    error TEXT NOT NULL DEFAULT '',
    UNIQUE (task_id, idempotency_key),
    UNIQUE (attempt_id, sequence_number),
    CONSTRAINT worker_result_batches_status_check CHECK (
        status IN ('accepted', 'persisted', 'duplicate', 'failed')
    )
);

CREATE INDEX worker_result_batches_task_idx
    ON worker_result_batches(task_id, sequence_number);

Gateway transaction boundary:

  1. Validate task, attempt, Worker identity, and lease token.
  2. Insert result-batch acceptance row.
  3. If unique conflict with same checksum, return previous acknowledgement.
  4. If unique conflict with different checksum, reject as idempotency conflict.
  5. Persist batch through existing ingestion services.
  6. Mark batch persisted.

If Gateway crashes after persistence but before responding, Worker retry with the same idempotency key must observe the persisted batch and return acknowledgement instead of re-ingesting.

Artifact upload sessions

PROPOSED table: worker_artifact_uploads.

CREATE TABLE worker_artifact_uploads (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    task_id UUID NOT NULL REFERENCES worker_tasks(id) ON DELETE CASCADE,
    attempt_id UUID NOT NULL REFERENCES worker_task_attempts(id) ON DELETE CASCADE,
    organization_id UUID NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
    scan_id UUID NULL REFERENCES scans(id) ON DELETE CASCADE,
    scan_job_id UUID NULL REFERENCES scan_jobs(id) ON DELETE CASCADE,
    object_key TEXT NOT NULL UNIQUE,
    original_name TEXT NOT NULL,
    content_type TEXT NOT NULL DEFAULT 'application/octet-stream',
    size_bytes BIGINT NULL,
    sha256 TEXT NOT NULL DEFAULT '',
    status TEXT NOT NULL DEFAULT 'prepared',
    expires_at TIMESTAMP NOT NULL,
    committed_at TIMESTAMP NULL,
    file_id UUID NULL REFERENCES files(id) ON DELETE SET NULL,
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    CONSTRAINT worker_artifact_uploads_status_check CHECK (
        status IN ('prepared', 'uploaded', 'committed', 'expired', 'failed')
    )
);

CREATE INDEX worker_artifact_uploads_task_idx
    ON worker_artifact_uploads(task_id);

CREATE INDEX worker_artifact_uploads_expiry_idx
    ON worker_artifact_uploads(expires_at)
    WHERE status IN ('prepared', 'uploaded');

This table is needed because remote Workers must not receive permanent platform S3 credentials, and files should only represent committed application artifacts.

Migration considerations

PROPOSED migration order:

  1. Add Worker identity and organization assignment tables.
  2. Add task, attempt, result-batch, and artifact-upload tables.
  3. Add repositories without changing the legacy Worker path.
  4. Write new Gateway-connected code against the new repositories.
  5. Migrate task creation from Redis queue to durable tasks behind a feature flag.
  6. Retire Redis BRPop only after Gateway-connected Workers handle production scans.

Existing queued Redis messages cannot be recovered after process restart because the current queue is Redis-list based. During migration, deployments should drain or disable legacy Workers before switching a scan profile or organization to Gateway execution.