Skip to content

Event pub/sub architecture

How Waldur delivers events to external consumers: one RabbitMQ queue per consumer, scoped by entity bindings, with authorization re-checked at publish time.

This describes the system as it is built. Runtime mechanics for site agents (registration calls, STOMP connection, payload examples, migration runbook) live in the unified agent queue guide.

Overview

flowchart TD
    subgraph sources["Event sources (signals)"]
        ev_market["Marketplace: orders, resources, offering users"]
        ev_user["Core: user profile, SSH keys, lifecycle, roles"]
        ev_proposal["Proposal: call and proposal state changes"]
    end

    ev_market --> disp_keys
    ev_proposal --> disp_keys
    ev_user --> disp_global

    subgraph dispatch["Dispatch"]
        direction TB
        disp_keys["1. Resolve the event scope-keys<br/>offering chain plus project chain"]
        disp_keys --> disp_match["2. Match consumer bindings<br/>set intersection on an indexed table"]
        disp_match --> disp_auth["3. Re-authorize each matched owner<br/>ONE batched role query"]
        disp_global["Global consumers<br/>no bindings, staff or support only"] --> disp_auth
        disp_auth --> disp_emit["4. Enrich once, emit one message per consumer"]
    end

    disp_emit --> publish["publish_messages then STOMP<br/>circuit breaker · rate limit · dedup · DLQ"]
    publish --> q1[(consumer_uuid1)]
    publish --> q2[(consumer_uuid2)]
    q1 --> c1["Consumer 1"]
    q2 --> c2["Consumer 2"]

    subgraph legacy["Legacy path — deprecated"]
        legacy_queue["EventSubscription plus EventSubscriptionQueue<br/>one queue per offering and object_type"]
    end
    disp_match -.->|suppressed for users already on a unified queue| legacy
    style legacy stroke-dasharray: 5 5

A signal fires once. The dispatcher resolves which scope-keys the event belongs to, matches consumers whose bindings intersect those keys, re-authorizes each matched owner, enriches the payload once, and emits one message per consumer. Each consumer drains exactly one queue and demultiplexes on the envelope's object_type.

The consumer model

waldur_core.logging.EventConsumer owns all unified-queue state:

Field Meaning
user Owner/registrant; the RabbitMQ vhost is user.uuid.hex
rmq_username RMQ login (random UUID hex), blank until provisioned
queue_created Whether the queue exists in RMQ
object_types Allow-list of event types (empty = all)
payload_filters Allow-list over envelope fields, {field: [values]} (empty = none)
delivery Transport: rmq (queue, the default) or webhook
webhook_url Destination of a webhook consumer; a credential, never returned
last_delivered_at, last_error, last_error_at Webhook delivery health
scopes Entity bindings — see below

The queue is consumer_{uuid}. A consumer is generic: a site agent, an IdM/IGA sync, the keycloak operator, or a reporting script can each own one. Site agents link theirs from AgentIdentity.event_consumer (a nullable OneToOne); AgentIdentity itself is purely site-agent registration and telemetry — AgentService, AgentProcessor and SiteAgentLog keep their names and their home in the marketplace_site_agent plugin.

Scope bindings

EventConsumerScope(consumer, content_type, object_id) binds a consumer to a list of entities — an offering, several projects, a customer — using GenericForeignKeys, which is what keeps waldur_core free of marketplace imports.

Bindings Meaning
one offering the classic site-agent case
projects, a customer, or a mix a slice the owner has access to
your own user ({"type": "user", "uuid": <own>}) your own identity events — profile, SSH keys, lifecycle, role changes. Authorized by identity, not by role; binding another user is staff/support only.
empty global: everything platform-wide — user-centric events (all-user PII) AND marketplace events. Staff/support only.

Bindings use the same vocabulary as Personal Access Tokens (TYPE_MAP, AllowedScopeInputSerializer), so {"type": "project", "uuid": "..."} means the same thing in both features. The one exception is user: it exists only for event-consumer bindings (EVENT_CONSUMER_TYPE_MAP), deliberately kept out of TYPE_MAP so PAT scopes, invitation targets, role content types, and the RoleType schema enum are unaffected.

An empty binding set means global. A consumer left without bindings by a partial write would therefore fail open as an all-user PII firehose. Both registration paths create the consumer and its bindings in one transaction.

Dispatch and delivery authorization

An event's scope-keys are the (content_type, object_id) pairs it belongs to: the offering's own chain (offering, its project, that project's customer, the offering's customer) plus the consumer-side project chain derived from the order or resource. Matching is a set intersection against the indexed EventConsumerScope table, so a consumer bound anywhere in that chain matches — binding to a customer picks up its projects' events.

Call and proposal events (proposal/event_publishing.py) carry their own chains instead. A call event belongs to the call, its managing organisation and that organisation's customer; a proposal event to the proposal, its call and the call's managing organisation — not the organiser's customer. The API admits customer-level roles to proposals only through PROPOSAL.LIST, which no customer role carries, while delivery re-checks only that some role exists in the chain; a customer key would therefore send proposal names and states to organisation owners, support and readers the API hides them from. get_scope_ancestors would yield the applicant's project for a proposal, whose members cannot see it through the API, so the applicant side is not part of the chain either.

The proposal app registers both chains with event_dispatch.register_event_chain, and registration (holds_role_on_event_chain) and delivery_blocked_reason resolve call and proposal bindings through the registered chain rather than get_scope_ancestors, so all three agree. Both directions matter: a call manager may bind to one proposal of their call, and a role on the applicant's project does not admit a binding to the proposal it applied for. Were the generic walker used, the first would be refused and the second accepted and then silently starved — the failure the serviceprovider rewrite in the same function already exists to prevent.

Authorization is enforced twice, and stays dynamic:

  1. At registration — you may only bind to an entity you hold an active role on, or on one of its ancestors (holds_any_role_on_scope_or_ancestor; for a call or proposal, on its event chain instead, as above). Staff/support may bind to anything, and are the only ones allowed to request the empty (global) binding set.
  2. At delivery — the same rule is re-checked for every matched consumer, in one batched query for the whole fan-out (users_with_role_on_any_scope_key). Global consumers are re-checked for staff/support.

Registration is therefore not a permanent grant: revoking someone's role stops their delivery immediately.

Two properties make this work:

  • get_scope_ancestors(offering) yields exactly what OfferingQuerySet.filter_for_user ORs over, so the binding predicate is the same rule the rest of the platform uses for offering visibility. The unified path is therefore never narrower than the legacy path it replaces — a plain project member can subscribe to their project's events.
  • Because it is one batched query rather than a permission query per recipient, the dynamic check is also cheaper than the per-recipient filter_for_user it replaced.

Client-side filtering (object_types) is a convenience knob, never a security boundary.

A deliberate divergence from the legacy rule

The legacy consumer-access check only resolved a customer when the project still had a non-terminated resource on the offering. Scope-keys are derived straight from the order or resource, so that narrowing is gone: a project-bound consumer now also receives the event for its last resource being terminated, which the legacy path dropped. Treated as a bug fix; see _resolve_event_project.

Event types and payloads

Coverage Trigger object_type
Orders, resources, offering users, service and course accounts, importable resources, periodic limits marketplace signals order, resource, offering_user, …
Role changes on any scope role_granted / role_revoked user_role
Profile fields User post_save diff user_profile
SSH keys SshPublicKey post_save / post_delete user_ssh_key
Account lifecycle user create, activate, deactivate, delete user_lifecycle
Call state (draft, active, archived) Call post_save state diff call
Proposal state (submitted, in review, accepted, …) Proposal post_save state diff; drafts cancelled on round close proposal

The user-centric events originate in waldur_core (User, SSH keys, roles), so they are dispatched from waldur_core.logging.event_dispatch and delivered to global consumers and to the affected user's self-bound consumers (dispatch_user_event passes the affected user's (content_type, id) as the scope-key; delivery to a self-bound consumer is authorized by identity — consumer.user == affected user — rather than by role). A user-scoped user_role event on the core path cannot double-deliver with the marketplace path below: the marketplace path matches only offering/project/customer keys, never a user key. A global (empty-scope) consumer means "everything": it also receives the marketplace events (orders, resources, offering users, …) through the marketplace dispatcher, which opts globals in via build_messages(include_global=True). The single care point is user_role, which BOTH dispatchers can emit — the core path on the role_granted/role_revoked signal, the marketplace path on a manual project re-sync. To avoid delivering it to a global twice, the core path owns user_role for globals and the marketplace path passes include_global=False for user_role only; every other object type is emitted by exactly one path, so a global receives it once. Suppression of a user's legacy subscription is keyed on messages actually delivered, so a global that genuinely receives a marketplace event correctly supersedes its own legacy queue for it. Payloads are built lazily — construction runs only when a consumer actually exists — and every emitter is guarded by get_skip_side_effects() so imports, migrations and bulk loads stay silent.

Profile events are data-minimized. The broadcast field set is derived from ENABLED_USER_PROFILE_ATTRIBUTES (plus always-on core fields such as email and name), so an attribute an operator disabled never leaves the platform, and newly-enabled ones are picked up automatically. is_active is deliberately excluded — it belongs to the lifecycle event. Auth and security columns are never eligible.

Envelope

Messages carry object_type so one queue can serve many event kinds, plus schema_version. The core dispatcher additionally stamps event_type. Marketplace events carry object_type, offering_uuid and schema_version, but not event_type (only the core dispatcher stamps it) — see Known gaps.

Credentials and queue lifecycle

The RMQ password is the credential the consumer presents: if it registered with a Personal Access Token, that PAT is the password, so the queue survives a browser logout or session-token rotation (the raw secret is read from the request header — only its hash is stored). A caller still using a plain session token keeps the DRF token as the password.

Which of the two it was is recorded on the consumer (auth_kind, plus the PAT's prefix and name) together with the permission branch that authorised the registration (authorized_via), and refreshed on every re-registration. Without it a site agent on a staff session is indistinguishable from one on a scoped PAT, and a revoked PAT cannot be traced to the queues it backed. A non-PAT or staff-authorised registration is audited, on change only.

Queues are provisioned with the hardened arguments the legacy path uses — x-message-ttl (1h), x-max-length (10 000), reject-publish-dlx overflow and a DLQ — and registration is idempotent, reconciling stale RMQ state.

Housekeeping tasks operate on EventConsumer regardless of binding:

  • cleanup_stale_agent_queues (24h) — tears down consumers whose owner holds neither a live DRF token nor a live PAT.
  • cleanup_dangling_agent_queues (1h) — clears queue_created when the backing RMQ user vanished out of band.
  • cleanup_orphan_subscription_queues (6h) — deletes consumer_* queues with no matching row.
  • pre_delete handlers tear down RMQ state for both agent-linked and standalone consumers.

Registration

  • Site agents — POST /api/marketplace-site-agent-identities/{uuid}/register_queue/ creates the consumer bound to the agent's offering. Permissions are unchanged (_can_manage_offering_agent), and ownership is one-user-at-a-time (409 on takeover).
  • Everything else — POST /api/event-consumers/register/ takes object_types and scopes. Any authenticated user may register a bound consumer; an empty scopes list requires staff/support. Idempotent per binding set.

Webhook delivery

Some receivers — ITSM, CRM and IdM tools — can only accept HTTP calls: they expose a listener URL, often with the credential in its query string, and expect flat JSON. For them a consumer is created with delivery=webhook and a webhook_url instead of a queue. Matching, delivery authorization, filters and payloads are exactly those of a queue consumer; only the last hop differs.

  • Creation — POST /api/event-consumers/register_webhook/ takes webhook_url, object_types, payload_filters and scopes (empty = global). Staff only. The URL must be https, must not carry user:password@ (put a credential in the query string instead), and may not resolve to a private or loopback address. It is never returned: the API shows webhook_destination (scheme, host and port) instead.
  • Changes — POST /api/event-consumers/{uuid}/update_webhook/ changes the URL, object types or filters. Staff only. The editor becomes the owner, because delivery is authorized by the owner's privileges, and every change emits an event_consumer_webhook_updated audit event naming the host, never the URL. Deregistering a webhook consumer is staff only too; support can list them.
  • Routing — build_messages emits {"webhook_consumer": uuid, "payload": …} for a webhook consumer. Every producer already hands its messages to publish_messages, which sends vhost/topic messages to STOMP and gives each webhook consumer its own deliver_webhook_messages task on the background queue. A webhook does not count as delivery to its owner, so the owner's legacy subscription queues keep receiving their events.
  • Delivery — each payload is POSTed as the JSON body. The URL is re-validated at send time (DNS rebinding), redirects are refused, the connect timeout is 3 s, and the response body is never read. The owner is re-checked first: once they are no longer active staff or support, queued messages are dropped. A non-2xx answer or connection error records last_error and retries the undelivered remainder with exponential backoff (10 s doubling, capped at 10 min, 8 retries), after which the remainder is dropped and logged. Error text never contains the URL, since requests' exception messages embed it.
  • Guarantees — delivery is at-least-once with no ordering across events: every event is its own task and retries independently, so a revoke can arrive before the grant it follows. Every message carries emitted_at, and every request an X-Waldur-Message-Id header that stays the same across retries, so a receiver can drop duplicates and ignore an update older than one it already applied.
  • Filters — payload_filters narrow delivery by envelope field, for both transports (only the webhook endpoints set them so far): {"role_name": ["CUSTOMER.OWNER"], "scope_type": ["customer"]} delivers organization-owner changes only. Comparison is type-strict, so 1 does not match true. A filter on a field a message lacks drops that message: a role_name filter also drops profile and SSH-key events.

Resync

Events describe changes, so a receiver that starts late has no way to learn the existing state. POST /api/event-consumers/{uuid}/resync/ (staff/support, global consumers only) replays it asynchronously through the consumer's own transport and filters: every active role grant of an active user, as a user_role message with event_type: "role_snapshot" and granted: true, in batches of 200. It is throttled to once per 15 minutes per consumer and emits an event_consumer_resync_requested audit event. Re-run it whenever a receiver loses its state. Snapshot batches can interleave with live events, so a receiver should compare emitted_at. Bound consumers are refused because a snapshot of every grant would bypass the per-binding authorization applied to each live event, and so are consumers whose filters would exclude every snapshot.

user_role messages carry the organization the scope belongs to — customer_uuid, customer_name and customer_registration_code — the scope itself for an organization role, its parent for a project or offering role. That is the key an external system matches its own records on.

The legacy path

EventSubscription plus EventSubscriptionQueue — one queue per (subscription × offering × object_type) — still runs, unchanged, so existing consumers keep working. Its endpoints are marked deprecated in the OpenAPI schema, and therefore in the generated SDKs.

The two paths never double-deliver: a user who owns a unified queue is suppressed from the legacy loop, so every consumer is on exactly one path.

Retirement — the migration command, drain telemetry, and the removal itself — is tracked in WAL-10111, gated on drain-to-zero rather than a date, because removal is a breaking change for anyone still on the legacy endpoints. Site-agent adoption of the unified queue is tracked in WAL-10011.

Why it is built this way

App-side fan-out, not broker-native routing. A topic exchange with routing keys was considered and rejected: routing keys are static, but Waldur's authorization is dynamic and revocable, so the broker cannot answer "may this consumer see this event now". Recipient resolution therefore happens server-side at publish time. The cost of that choice is CPU and queries per event, which is why matching is a set intersection and the re-auth is batched.

Bindings normalized, not JSON. PersonalAccessToken stores allowed_scopes as JSON because enforcement checks one token per request. Pub/sub is the inverse problem — one event matched against many consumers — so bindings live in an indexed table the dispatcher can intersect against.

No new permissions. "What you may subscribe to" is already defined by "what you have access to", so binding validation reuses existing roles rather than introducing SUBSCRIBE_* permissions that would have to be kept in sync.

Known gaps

  • Envelope — marketplace events do not carry event_type (only the core dispatcher stamps it); they do carry object_type, offering_uuid and schema_version.
  • Round open/close — open/closed is derived from start_time/cutoff_time, so no round event is emitted when a round opens or ends. The stored lifecycle after the cut-off (evaluating, deciding, results_published, closed) is logged as round_* events but not yet published either. The proposal cancellations a round's end causes are published, as proposal events, on two paths:
  • at cutoff, the periodic proposals_for_ended_rounds_should_be_cancelled task cancels the round's draft proposals — submitted and in_review ones stay with the review workflow — saving them one at a time, so each goes out through post_save as its own publish task;
  • the manual close_round endpoint cancels only the drafts, in one bulk update announced as a single batch. A busy round therefore reaches consumers as a burst of cancellations at cutoff. On both paths a call with carry_over_drafts first moves its drafts to its next round; a move changes no state, so it publishes nothing (it is logged as a proposal_draft_carried_over event instead), and only drafts without a next round are cancelled.
  • Webhooks — delivery relies on the credential the receiver embeds in its URL; there is no HMAC signature header yet. Ordering is per event, not per consumer. Core role events are published from the signal, not on commit, so a grant rolled back after the signal still reaches consumers. Resync covers user_role only.
  • AgentIdentity retirement — the site-agent model still exists alongside its consumer; collapsing it is only worthwhile once the legacy path is gone.

References