Work Queue Service

Generic work queue service for Hummingbird. Manages work items through type-specific lifecycles with a shared processing framework. Currently handles MR creation and updates; designed to support CVE remediation, advisory tracking, and other work item types.

For design decisions and architectural rationale, see Work Queue Service Design.

Architecture

The service runs as a Kubernetes deployment sharing the same PostgreSQL database as hummingbird-status and hummingbird-dashboard (with a service-specific Alembic migration chain).

flowchart TD
    clients["Internal services"] -- "POST /work-items/mr" --> svc["Work Queue Service"]
    gl_events["GitLab\n(events)"] -- webhook --> fwd["gitlab-event-forwarder"]
    fwd --> sns["SNS"]
    sns -->|"filter: source=gitlab,\nevent_type=[merge_request, build]"| sqs["SQS Queue"]
    sqs --> svc
    svc -- "Commits API\n(create + update)\nJobs API\n(trace + retry)" --> gl["GitLab\n(API)"]
    svc -- "shared DB" --> pg[("PostgreSQL")]

The SQS consumer thread receives GitLab merge_request and pipeline events from the SNS topic via a filtered SQS subscription. Events are stored in the pending_events table and processed by existing work-item threads, which map MR state changes to phase transitions and trigger CI job retries on pipeline failure.

Design

Generic + detail tables

The schema separates queue machinery from type-specific data:

• work_items — generic: id, type, phase, actionable, attempts, leased_until, error, timestamps. Shared by all work item types.
• mr_details — MR-specific: package, project, file_actions, commit_message, content_hash, mr_iid, etc. Joined to work_items via foreign key.
• work_item_statuses — independent readiness dimensions with composite key (work_item_id, dimension, variant). Includes pending_until for deferred re-evaluation.

New work item types add their own detail table (e.g. cve_details) — no changes to the generic layer. SQLAlchemy joined table inheritance loads the correct subclass automatically via polymorphic_load="selectin".

Phases

Four generic phases track coarse lifecycle status:

Items start as active. Only active items accept event-driven phase transitions and executor actions. The three terminal phases (completed, failed, cancelled) are immutable to events — recovery requires explicit action via the retry API. GitLab merge_request events drive transitions from active: merge -> completed, close -> cancelled.

Dimensions

Independent readiness signals stored in work_item_statuses. All dimensions follow a uniform needs_* trigger model: status is "true" (action needed) or "false" (handled). Absent dimensions are treated as actionable by the decider (!= "false"). Each dimension has an optional variant for multi-instance signals (e.g. per-component Konflux builds) and an optional pending_until timestamp for deferred re-evaluation (see Deferred dimensions below).

Dimensions are set by executors or event handlers via EventResult.dimensions; the generic handle_event layer applies event-driven dimensions and emits audit entries.

Dimension schema

work_item_statuses:
  work_item_id: UUID  (PK, FK)
  dimension:    Text  (PK)
  variant:      Text  (PK, default="")
  status:       Text  (NOT NULL)       -- "true" or "false"
  updated_at:   DateTime(tz)
  detail:       JSONB (nullable)
  pending_until: DateTime(tz, nullable)

Unified state diagram

All dimensions share the same state machine:

stateDiagram-v2
    [*] --> absent : item created
    absent --> true : decider treats absent as actionable
    true --> true_deferred : executor (defer for later)
    true --> false : executor (done, no further checks)
    true_deferred --> true : sweep notifies (pending_until expired)
    true_deferred --> true : webhook/handler (clears pending_until)
    false --> true : webhook/handler (re-activates)
    note right of absent : no dimension row exists
    note right of true : status="true", pending_until=NULL
    note right of true_deferred : status="true", pending_until=future
    note right of false : status="false", decider ignores

Per-dimension transitions

Detail JSONB varies by dimension: needs_dashboard stores {"note_id": int, "url": str}, needs_ci_retry stores a jobs list, others use null.

Deferred dimensions

When an executor or handler sets pending_until to a future timestamp alongside status="true", the decider treats the dimension as "false" (not actionable) until that time. The periodic sweep finds items with expired pending_until and notifies them — no synthetic events are inserted.

Decider

A pure function registered per type that derives the next action from the work item’s current state. The decider reads the selectin-loaded item.statuses list (in-memory, no DB call) to check dimensions, pre-filtering deferred dimensions (where pending_until is in the future). See workqueue_service/mr/decider.py for the MR decider implementation.

The decider scans dimensions in priority order (DDR-45, DDR-48). Absent dimensions are treated as actionable (!= "false"). Deferred dimensions (pending_until in the future) are excluded from the scan. Adding a new concern = new dimension + handler, no decider changes.

The decider determines the next action to execute — it does not set phases or modify state.

MR upsert semantics

The MR submission endpoint uses upsert logic keyed on (project, source_branch). Callers always submit the current desired state; the service decides what to do based on existing work items:

stateDiagram-v2
    [*] --> created : no active item
    [*] --> unchanged : active, same hash
    [*] --> updated : active, different hash
    [*] --> retried : failed item

    created --> active : new work item
    unchanged --> active : no-op
    updated --> active : refresh fields
    retried --> active : reset phase

Content hash. A SHA-256 of the normalized file_actions JSON (sorted by file_path + action, compact separators, sorted keys). Enables cheap same-content detection without comparing the full payload.

Idempotency key. Derived server-side as mr:{project}:{source_branch}:{generation} where generation is a counter per (project, source_branch) pair. A partial unique index ensures uniqueness only among non-terminal phases — completed and cancelled items can share a key prefix without blocking new items for the same branch.

Race handling. If two concurrent inserts race for the same idempotency key, the loser catches the IntegrityError, rolls back, and retries the full upsert — the winner is now visible and the correct action (unchanged/updated) is taken.

MR action chain

The MR work item lifecycle chains executor actions. After upsert, the processing loop calls the decider to select the first action; subsequent actions are also selected by the decider after each executor completes:

• ensure_mr — creates or updates the MR. If mr_iid is not set, creates the branch, commit, and MR via the GitLab Commits API and applies the managed-by::workqueue scoped label. If mr_iid is already set, force-pushes updated content to the existing MR branch and syncs MR metadata (title, description). Detects empty diffs (before creation) or closes an empty MR after force-push. Sets needs_mr to "false" on success.
• ensure_dashboard_note — posts an internal note with the Hummingbird dashboard link on the MR and stores the needs_dashboard dimension as "false" with detail {"note_id": int, "url": str}. Selected by the decider after create_mr or update_mr completes when needs_dashboard != "false".
• check_empty_diff — checks whether an active MR’s diff became empty (target branch moved). Triggered by the needs_empty_check dimension via pending_until expiry (~every 30 minutes per item). If the diff is empty, closes the MR with an explanatory note, deletes the branch, and cancels the item. Sets needs_empty_check to "false" on completion. Transient GitLab API errors are logged but do not fail the item.
• reconcile_mr — reconciles desired vs actual MR state. Triggered by the needs_reconcile dimension, set by webhooks (approved, unapproved, update, pipeline success/failure) and pending_until expiry. Merges directly when detailed_merge_status is mergeable and auto_merge is enabled. Syncs title and description if drifted. Sets needs_reconcile to "false" on completion (with pending_until for the next cycle).

Draft state. The draft submission option controls initial MR creation only. After creation, GitLab owns Ready/Draft state; update_mr and reconcile_mr preserve the current GitLab state while synchronizing title text. The details.draft response field reports the initial creation setting, not the live GitLab status.

After all actions complete, the decider returns None and the item sits in active phase waiting for event-driven lifecycle tracking (merge → completed, close → cancelled).

The ensure_dashboard_note executor sets the needs_dashboard dimension to "false" with detail {"note_id": int, "url": str}. The note_id is the GitLab note ID used for future status updates.

Label coordination with dashboard-mr-linker. The managed-by::workqueue label causes the dashboard-mr-linker Lambda to skip the MR, so the workqueue-service creates the dashboard note for its MRs. When dashboard_base_url is not configured, ensure_dashboard_note returns immediately without posting a note.

head_pipeline_id on mr_details

The current head pipeline ID is stored as a typed integer column (head_pipeline_id) on the mr_details table rather than as a dimension. Build events and the reconciler update it directly. The CI retry executor uses it to filter stale jobs from previous pipelines.

Processing loop

The ProcessingManager uses a bounded ThreadPoolExecutor (10 workers):

• Claims items where actionable IS TRUE using SELECT ... FOR UPDATE SKIP LOCKED
• Serial per item — one action at a time per work item
• Parallel across items — independent items processed concurrently
• Per-item deadline — processing loops exit after 60 seconds to prevent stuck items from holding pool workers
• Heartbeat context manager extends leased_until during execution
• Executors apply state updates directly (phase, MR fields) and must not call session.commit() or session.rollback()
• After execution, the loop re-derives the next action via the decider
• Periodic sweep reclaims items with expired leases and notifies items with expired pending_until dimensions (no synthetic events)

MR labeling

The executor always adds the managed-by::workqueue scoped label to every MR it creates, merged with any caller-provided labels. This label is a coordination signal: other services (e.g. dashboard-mr-linker) check for this label and skip managed MRs, since the workqueue-service handles dashboard note creation itself via the ensure_dashboard_note action.

Event processing

The SQS consumer thread receives merge_request events from the SNS topic and stores them in the pending_events table. Work-item threads drain and process these events as part of their main loop.

Consumer thread. A single daemon thread (poll_loop) long-polls SQS with MaxNumberOfMessages=10 and WaitTimeSeconds=20. For each message it decodes the SNS envelope (handles gzip+base64), resolves matching work items via the resolver registry, INSERTs into pending_events, commits, deletes the SQS message, and calls manager.notify() for each affected work item. The consumer is started only when WORKQUEUE_SERVICE_EVENTS_QUEUE_URL is set.

State mapping. MR state changes are mapped to phases:

Only items in active phase accept event-driven transitions. Events for terminal items (completed, failed, cancelled) are silently discarded.

Out-of-order guard. The last_event_at column on work_items rejects events with updated_at older than the last processed event. Same-timestamp events pass the <= guard (processing is idempotent).

Advisory lock. A per-work-item pg_advisory_xact_lock coordinates the consumer’s INSERT with the work-item thread’s exit decision, preventing events from being orphaned between the final drain check and thread exit.

Sweep extension. The periodic sweep (every 5 minutes) also checks for work items with unprocessed pending_events, providing startup recovery and a safety net behind the advisory lock.

Adding a new work item type

1. Create a detail table and SQLAlchemy model (e.g. CveWorkItem)
2. Register a decider function: decider.register("cve", _cve_decide)
3. Register executor functions for each action
4. Add a submission endpoint: POST /api/work-items/cve

No changes needed to the processing loop, claiming, or generic API.

API

Endpoints

Responses include a nested details dict with type-specific fields:

id: "..."
type: mr
phase: active
details:
  source: rpms-ci
  mr_iid: 42
  mr_url: https://gitlab.com/org/group/project/-/merge_requests/42

Authentication

All callers authenticate via Authorization: Bearer tokens. The service validates each token in one of two ways:

X-Forwarded-* headers are never trusted on API paths. The proxy protects only the OAuth login flow (/oauth/start, /oauth/callback) and the POST /oauth/token endpoint, which bridges cookie sessions to Bearer tokens for browser and CLI users. The /oauth/token endpoint uses POST (not GET) to prevent CSRF — browser-initiated GETs via <img> or <script> tags cannot trigger it.

OAuth proxy deployment requirements

The oauth-proxy sidecar handles only the OAuth login flow. All /api/ paths bypass the proxy — the app handles auth itself. The proxy must be configured with these flags:

Why --cookie-refresh is critical

The proxy manages two independent lifetimes that are easy to confuse:

• Session cookie — controlled by the proxy (--cookie-expire, defaults to browser session). Contains the encrypted OCP access token.
• OCP access token — controlled by OCP (default 24h). The actual credential the app uses to call users/~.

The proxy does not know when the OCP token inside its cookie expires. It only checks whether the cookie is valid (signature, expiry). Without --cookie-refresh, the cookie outlives the token:

--cookie-refresh=1h tells the proxy to silently refresh the OCP token inside the cookie every hour, keeping it well within the 24h expiry. The workqueue-service also uses a TTL-bounded cache (5 minutes) for users/~ results, so expired-token failures are never permanent.

Authorization

Access rules are defined in the config file. Each rule maps a remote identity to permissions (read, retry, submit), optionally scoped to work item types. Rules use either groups (OCP group membership) or issuer+claims (JWT bearer). Claim matching follows Vault’s bound_claims pattern: all conditions AND, list values OR.

All permissions (including read) must be granted by a matching rule. Authenticated callers with no matching rules get 403. Multiple matching rules are merged (union of permissions).

See Configuration for the config file format.

Whoami introspection

GET /api/whoami returns the caller’s identity and effective permissions. Example response:

authentication:
  tier: ocp-user
  identity: admin-user
  groups:
    - konflux-hummingbird-admin-access
authorization:
  effective_permissions:
    mr:
      - read
      - retry
      - submit
  matching_rules:
    - groups:
        - konflux-hummingbird-admin-access
      permissions:
        - read
        - submit
        - retry
      item_types: []

API Documentation

Interactive API documentation available without authentication: /docs (Swagger UI), /redoc (ReDoc), /openapi.json.

Audit Log

State-changing operations (processing, API calls, auth decisions) emit audit entries. Every entry is logged at INFO level. When WORKQUEUE_SERVICE_AUDIT_S3_BUCKET is set, entries are additionally archived to S3 as gzipped JSON objects for long-term querying via Athena.

Audited events

S3 key format

When S3 archival is enabled, entries are stored as gzipped JSON with an Athena-partitionable key layout:

{prefix}/YYYY/MM/DD/HH/MM/{timestamp}#{work_item_id}::{event_type}.json.gz

The ContentEncoding: gzip header is set so S3 API consumers get transparent decompression.

Field conventions

Follow these conventions when adding new audit calls:

• actor: caller identity for API/auth events; thread name for worker events; "system" for decider/system-triggered transitions.
• old_state: state before the change. Include phase and action when both are relevant. Omit on creation (no prior state) and on auth events (no work item state).
• new_state: state after the change. Same keys as old_state. Omit fields that are None (filtered by to_dict()).
• trigger: what caused this event. Always include source ("api", "worker", or "decider"). Add action for deliberate operations, reason for error/system transitions, error for failure details.
• metadata: context orthogonal to the state transition — request info on auth events, type-specific data on submit. Not for duplicating state already in old_state/new_state.

CI Job Retry

When a CI pipeline on a managed MR fails, the service can automatically retry jobs that match known transient failure patterns. This is ported from the CKI pipeline-herder’s proven regex-based job log matching engine.

How it works

1. GitLab sends a build (job) webhook when a CI job finishes
2. The event handler appends the failed job to the needs_ci_retry dimension’s jobs list and sets status to "true"
3. The decider selects the retry_ci_jobs executor
4. The executor processes all jobs in the list, matching each against the config tree and retrying matching jobs via the GitLab Jobs API
5. Jobs with non-zero delay set pending_until to the earliest per-job not_before timestamp — the sweep notifies when it expires
6. Multiple failed jobs in the same pipeline accumulate in the list

needs_ci_retry dimension

The needs_ci_retry dimension follows the unified state machine (see Unified state diagram). Jobs with non-zero delay use the deferred path: status="true" with pending_until set to the earliest per-job not_before timestamp. The sweep notifies when any job’s delay elapses.

The true → true self-loop fires when another job in the same pipeline fails — the build event handler appends the new job to the existing list (clearing pending_until so the decider picks it up immediately).

The detail JSONB stores a jobs list. Each entry carries build_id, build_name, build_failure_reason, build_finished_at, retries_count, and pipeline_id — all from the build webhook payload. The executor deduplicates by build_id and skips jobs whose pipeline_id does not match the MR’s current head_pipeline_id. When a job has a non-zero delay, the executor stamps a per-job not_before ISO timestamp based on build_finished_at + retry_delays[retries_count]. The dimension’s pending_until is set to the earliest per-job not_before so the sweep fires as soon as any job is ready.

Matcher configuration

CI job retry is configured per type under handled_types.mr.job_retry. The config uses a recursive matcher tree where parent fields are inherited by children:

handled_types:
  mr:
    job_retry:
      retry_delays: ["0", "5m", "10m"]
      retry_limit: 5
      matchers:
        - name: runner-system-failure
          description: Job failed with runner_system_failure
          failure_reason: runner_system_failure
        - name: git-clone
          description: Git clone failure
          messages:
            - "fatal: unable to access"
            - '/error: RPC failed; curl 92 HTTP\/2 stream \d+/'

Matcher fields

Message matching

• Literal: "fatal: unable to access" → substring match per line
• Regex: '/error: RPC failed; curl \d+/' → re.search on joined lines (slash-delimited)

Retry safety

Before retrying a job, the executor checks:

1. No newer job with the same name exists (prevents double retries)
2. Retry limit not exceeded (count of jobs with the same name)

Retry timing

The retry_delays list is indexed by the job’s attempt count (0-based). When the index exceeds the list length, the last entry repeats. A delay of "0" means immediate retry; non-zero delays set pending_until to the computed not_before timestamp. The sweep notifies the item when pending_until expires. Due to sweep cycle alignment, actual delay precision is ±5 minutes.

Prometheus Metrics

The service exposes Prometheus metrics at /metrics on the same port as the API (no sidecar). Metrics follow the repo naming convention: generic operational metrics use the shared hummingbird_ prefix, service-specific metrics use hummingbird_workqueue_.

Generic operational metrics

Service-specific metrics

UUID path segments in HTTP metrics are normalized to /{id} to limit label cardinality. The /healthz and /metrics paths are excluded from HTTP metrics.

CLI

export WORKQUEUE_SERVICE_URL=https://mr.apps.cluster.example.com
export WORKQUEUE_SERVICE_TOKEN=$(oc whoami -t)

# Update an existing file (local_path:repo_path)
workqueue-service submit \
  --package kernel \
  --project org/group/project \
  --change-kind sync \
  --file /path/to/sources.spec:sources.spec \
  --commit-message "Sync sources from upstream" \
  --source-branch mr-service/sync-1 --title "Sync sources"

# Create a new file
workqueue-service submit \
  --package kernel \
  --project org/group/project \
  --change-kind sync \
  --create /path/to/new-file.txt:new-file.txt \
  --commit-message "Add new file" \
  --source-branch mr-service/add-1 --title "Add file"

# Check status
workqueue-service status --id <uuid>

# List active MR items
workqueue-service status --phase active

# Show identity and permissions
workqueue-service whoami

# Raw JSON output
workqueue-service whoami --json

--url overrides WORKQUEUE_SERVICE_URL. --no-auth skips all credential loading and sends unauthenticated requests (like aws --no-sign-request). Useful with whoami to verify anonymous access.

CLI authentication

The CLI resolves credentials in this order:

1. WORKQUEUE_SERVICE_TOKEN env var — used directly as a Bearer token. Recommended for CI (OIDC/projected SA tokens).
2. Cached token — if a valid token exists in the per-server cache, it is reused automatically.
3. OAuth/Kerberos login — when neither env var nor cache is available, the CLI performs a synchronous headless SPNEGO login against the oauth-proxy, exchanges the session cookie for an OCP Bearer token via POST /oauth/token, and caches it for subsequent calls. Requires requests-gssapi and a valid Kerberos ticket (kinit).

Tokens are cached per-server under $XDG_CACHE_HOME/workqueue-service/tokens/ (defaults to ~/.cache/workqueue-service/tokens/) with 0600 permissions and a 20-hour TTL. If a cached token is rejected (HTTP 401), the CLI invalidates the cache and retries once with a fresh OAuth login.

Configuration

Config file

The service reads its configuration from a YAML file at startup. The path defaults to /etc/workqueue/config.yml and can be overridden with WORKQUEUE_SERVICE_CONFIG. The service refuses to start if the file is missing or invalid (fail-closed).

authorization:
  rules:
    # Any in-cluster SA with the right audience can read
    - issuer: k8s
      claims:
        aud: workqueue-service
      permissions: [read]

    # Human users -- OCP group membership
    - groups:
        - konflux-hummingbird-admin-access
      permissions: [read, submit, retry]

    # Specific K8s SA for CI automation
    - issuer: k8s
      claims:
        aud: workqueue-service
        sub: system:serviceaccount:hummingbird--internal:rpms-ci
      permissions: [submit, retry]
      item_types: [mr]

    # GitLab CI OIDC
    - issuer: https://gitlab.com
      claims:
        aud: workqueue-service
        project_path: redhat/hummingbird/rpms
      permissions: [read, submit, retry]
      item_types: [mr]

handled_types:
  mr:
    gitlab_url: https://gitlab.com
    projects:
      redhat/hummingbird/rpms:
        token_env: GITLAB_TOKEN_RPMS
      redhat/hummingbird/containers:
        token_env: GITLAB_TOKEN_CONTAINERS
      redhat/rhel/src:
        token_env: GITLAB_TOKEN_RHEL

Authorization rules

Each rule has either groups (OCP group membership from users/~ response) or issuer+claims (JWT bearer), not both.

• issuer determines which JWKS keys validate the token signature. Use k8s as shorthand for the auto-detected cluster OIDC issuer.
• claims matches JWT payload fields. All conditions must match (AND); list values match any (OR). Values are compared case-insensitively with string coercion (handles numeric project_id, boolean ref_protected).
• permissions grants read, retry, and/or submit.
• item_types optionally restricts the rule to specific work item types. Omit to apply to all types.
• Every issuer rule must include aud in claims to prevent cross-service token reuse.

Handled types

handled_types declares which work item types this instance serves. Only listed types get their routers and executors registered. The processing sweep only picks up items matching these types. Empty or omitted = handle nothing.

Each type can have per-type operational config:

Per-project token configuration

The projects map under handled_types.mr controls which GitLab API token the executor uses for each target project. Keys are GitLab project paths or group prefixes. Every project submitted to the service must match an entry – there is no default fallback token.

Matching order:

1. Exact match on the full project path
2. Longest group-prefix match on a / boundary (e.g. redhat/rhel/src matches redhat/rhel/src/kernel)

Each entry requires token_env (the name of the environment variable holding the GitLab API token) and optionally overrides gitlab_url:

handled_types:
  mr:
    gitlab_url: https://gitlab.com
    projects:
      redhat/hummingbird/rpms:
        token_env: GITLAB_TOKEN_RPMS
      redhat/hummingbird/containers:
        token_env: GITLAB_TOKEN_CONTAINERS
        gitlab_url: https://gitlab.example.com
      redhat/rhel/src:
        token_env: GITLAB_TOKEN_RHEL

At startup, the service validates that all referenced token_env variables exist in the process environment and refuses to start if any are missing.

Adding a new issuer

Add a rule with the issuer’s OIDC discovery URL. The service discovers the JWKS endpoint lazily on the first token from that issuer:

- issuer: https://token.actions.githubusercontent.com
  claims:
    aud: workqueue-service
    repository: org/repo
  permissions: [submit]
  types: [mr]

Environment variables

Development

Local Development

cd workqueue-service
./dev.sh db-start    # Start local PostgreSQL
./dev.sh db-migrate  # Run Alembic migrations
./dev.sh start       # Start the service
./dev.sh db-shell    # PostgreSQL shell
./dev.sh db-stop     # Stop PostgreSQL
./dev.sh db-reset    # Stop and delete volume

Running Tests

cd workqueue-service
pip install -e ".[dev]"
python -m unittest discover tests

Tests use testcontainers[postgres] locally (auto-detects Podman) or a CI-provided PostgreSQL sidecar via DATABASE_URL.

License

This project is licensed under the GNU General Public License v3.0 or later — see the LICENSE file for details.