Change Queue Service (CQS)¶

Purpose¶

The Change Queue Service (dtoflow-changequeue) is the event bus of the DTOflow platform. It replaces point-to-point Pub/Sub fan-out with a structured, prioritized notification system. When the DTOflow Server records a DTO change, CQS delivers a lightweight change notification to every queue that has subscribed to that DTO type. Each consuming service has exactly one queue, named after the service. The queue holds notifications in priority order (by SLA timestamp) and supports long-polling, so consumers can block efficiently until work is available.

CQS uses per-DTO Pub/Sub topics as its internal persistence backbone — it drains these topics continuously and re-presents notifications to consumers sorted by SLA priority and optionally grouped by tenant/store. The consumer-facing abstraction is a named queue per service, not a raw Pub/Sub subscription.

CQS solves several problems that arise with raw Pub/Sub:

Priority. A user clicking "update ESL" in the Studio UI should be processed before a background nightly batch job, even if the batch job's notifications were enqueued first. CQS uses the sla_timestamp from the write request as its primary sort key, so urgent changes always surface first.
Ordered batching. Processing one notification at a time is inefficient when many ESLs in the same store change simultaneously. CQS supports batched dequeue with prefix-group semantics: the first item in a batch is the most urgent across the entire queue, and the remaining items are drawn from the same store (or tenant) prefix — so a consumer can process a cohesive batch of related changes together.
Explicit acknowledgement. Notifications are not removed until the consumer explicitly acknowledges them. If a consumer crashes mid-processing, unacknowledged notifications become visible again for redelivery.

Queue model¶

One queue per consuming service¶

Each service that consumes DTOs declares one queue, named after its Cloud Run service name (or, for on-premise consumers, its canonical name). The queue name is stable across deployments and is used in every BatchDequeue and BatchAcknowledge call. Examples: studio-renderer, link-registry, pricer-server.

Notification structure¶

Each notification in the queue is a DtoChangeNotification:

message DtoChangeNotification {
  string ack_token = 1;           // Required for acknowledgement
  string dto_type = 2;            // e.g. "Storeesl", "Link"
  oneof urn {
    string dto_id = 3;            // ID of the specific DTO that changed
    string parent_id = 6;         // Parent ID: signals that multiple children changed
  }
  string alias_name = 7;          // Set if this notification is alias-scoped
  google.protobuf.Timestamp sla_timestamp = 4;  // Business deadline / priority key
  string tracking_id = 5;         // Distributed tracing identifier
}

The notification is intentionally minimal: it carries an ID and a type, not the DTO payload. This keeps the queue lean and avoids stale data issues — by the time a consumer reads the notification, the DTOflow Server always has the latest version of the DTO.

parent_id notifications signal that multiple children of a parent have changed but should not be processed individually right away. The consumer should typically re-enqueue those children — via the ReEnqueue RPC — for batch processing rather than reading each child immediately.

Alias-qualified notifications indicate that an alias mapping has changed: the alias now points to a different DTO instance (or has been created or deleted). A notification with both dto_id and alias_name set means the alias alias_name at path dto_id has been updated. This notification is not sent when the DTO the alias points to changes content — for that, the consumer receives a plain dto_id notification.

Consume-process-acknowledge cycle¶

loop:
  notifications = BatchDequeue(queue_name, max=10, prefix_depth=4, wait_ms=30000)
  for each notification:
    if dto is deleted: handle deletion
    else: dto = Reader.Read(notification.dto_id)  // may return NOT_FOUND → deleted
          process(dto)
  BatchAcknowledge(queue_name, [notification.ack_token for each notification])

An unacknowledged notification is not delivered again immediately, but it does become eligible for redelivery if the consumer fails to acknowledge within the server's visibility timeout. Acknowledging a token that was already acknowledged or is unknown is silently ignored — this makes acknowledgement safe to retry.

Priority and ordering¶

CQS sorts notifications by sla_timestamp in ascending order (nearest deadline first). The first notification returned by BatchDequeue is always the globally highest-priority item in the queue.

The remaining notifications in a batch are selected from the same path prefix group as the first item (controlled by prefix_depth). This keeps related changes together:

prefix_depth=2 groups by tenant (t/{tenantId})
prefix_depth=4 groups by store (t/{tenantId}/s/{storeId}) — the default

Example: Suppose the queue contains these notifications in SLA order:

Priority	DTO ID	SLA
1 (most urgent)	`t/farmersmarket/s/oslo-main/esls/7318690123456`	10:00:01
2	`t/farmersmarket/s/bergen-west/esls/9876543210987`	10:00:02
3	`t/farmersmarket/s/oslo-main/esls/7318690999001`	10:00:05

A BatchDequeue with max_notifications=2 and prefix_depth=4 returns items 1 and 3. Both share the prefix t/farmersmarket/s/oslo-main (4 segments). Item 2 is skipped — it is from a different store — even though its SLA is earlier than item 3's.

This behaviour allows a consumer to make a single BatchRead call to the DTOflow Server with a parent hint, minimising round trips to the DTO server.

Queue configuration¶

Queues are created and configured via the CreateOrConfigureQueue RPC at service startup. The call is idempotent: if the queue already exists with the same subscriptions, it is a no-op.

message CreateOrConfigureQueueRequest {
  string queue_name = 1;
  repeated DtoTypeSubscription subscriptions = 2;  // Replaces existing subscriptions
  int32 depth = 3;  // Default prefix_depth for batching (create-only; defaults to 4)
}

message DtoTypeSubscription {
  string dto_type   = 1;
  string alias_name = 2;  // Optional: subscribe only to alias-scoped notifications
}

The subscriptions list in CreateOrConfigureQueue replaces the existing subscription list on every call. This means services can safely reconfigure their queue on every startup without accumulating stale subscriptions.

Current queue subscriptions¶

Queue name (Cloud Run service)	Subscribed DTO types
`studio-renderer`	`studiolink.v1`, `storeitemvalues.v1`, `design.v1`, `storeesl.v1`, `canvasdesign.v1`, `designerlink.v1`
`studio-link-evaluator`	`communicationpack.v1`, `link.v2`, `storeitemvalues.v1`
`pricer-server`	`eslimage.v1`, `storeesl.v1`
`link-registry`	`link.v1` (legacy bridge only)

Note that studio-renderer subscribes to storeesl.v1 directly: when an ESL's store assignment changes, the renderer needs to re-render the display. pricer-server also subscribes to storeesl.v1 so it can register/unregister the device and transmit updated display data. Both queues receive the same notification; CQS fans it out independently.

link-registry subscribes to link.v1 only as a legacy compatibility bridge during the migration from link.v1 to link.v2. Once migration is complete, this subscription will be removed.

Dequeue pattern¶

The recommended production dequeue loop:

1. At service startup — register the queue:

CQS.CreateOrConfigureQueue(
  queue_name = "studio-renderer",
  subscriptions = [
    { dto_type = "studiolink.v1" },
    { dto_type = "storeitemvalues.v1" },
    { dto_type = "design.v1" },
    { dto_type = "storeesl.v1" },
    { dto_type = "canvasdesign.v1" },
    { dto_type = "designerlink.v1" },
  ],
  depth = 4
)

2. Main processing loop:

while running:
  response = CQS.BatchDequeue(
    queue_name = "studio-renderer",
    max_notifications = 20,
    prefix_depth = 4,
    wait_time_milliseconds = 30_000
  )

  if response.notifications is empty:
    continue  // long-poll timed out, no work available

  ids = [n.dto_id for n in response.notifications]
  parent = common_prefix(ids, depth=4)  // e.g. "t/farmersmarket/s/oslo-main"

  dtos = DTOflowReader.BatchRead(ids=ids, parent=parent)

  for each notification in response.notifications:
    if notification.dto_id not in dtos:
      handle_deletion(notification.dto_id, notification.dto_type)
    else:
      process(dtos[notification.dto_id])

  CQS.BatchAcknowledge(
    queue_name = "studio-renderer",
    ack_tokens = [n.ack_token for n in response.notifications]
  )

Key points: - Acknowledge only after processing is complete. If the service crashes between dequeue and acknowledge, the notifications will be redelivered. - Pass the common parent prefix to BatchRead as a sharding hint to improve read performance. - Treat a NOT_FOUND response from the Reader as a deletion: the DTO was deleted between the notification being enqueued and the consumer reading it.

Error handling¶

Processing a batch of notifications can fail in two fundamentally different ways, and the consumer must handle them differently.

Permanent errors — bad data, an ID that should exist but doesn't, an INVALID_ARGUMENT rejection from a downstream service — will never be resolved by retrying. If such an error propagates out of the processing loop uncaught, BatchAcknowledge is never reached and the notification re-enters the queue indefinitely. The correct action is to log the error with full context and include the notification's ack token in the final acknowledgement.

Transient errors — a downstream service temporarily unavailable, a deadline exceeded, an internal server error — may resolve on their own. The correct action is to log a warning and exclude the token: the notification will become eligible for redelivery after the server's visibility lease expires.

The error classification must happen per notification, not at batch level. A try-catch around the whole batch loses the ability to ack the other notifications in the batch.

For gRPC consumers, UNAVAILABLE, DEADLINE_EXCEEDED, and INTERNAL are transient (the gRPC equivalents of HTTP 5xx). All other status codes are permanent. Walk the exception cause chain when classifying — gRPC errors are sometimes wrapped.

Pattern¶

while running:
  response = CQS.BatchDequeue(...)
  if response.notifications is empty:
    continue

  tokens_to_ack = []

  for each notification in response.notifications:
    try:
      dto = Reader.Read(notification.dto_id)  // NOT_FOUND → deleted
      process(dto)  // or handle_deletion if not found
      tokens_to_ack.append(notification.ack_token)
    except TransientError as e:
      // Omit this token — let the notification redeliver after lease expiry
      log.warn("Transient error on {}, will redeliver: {}", notification.dto_id, e)
    except Exception as e:
      // Permanent: bad data, INVALID_ARGUMENT, NOT_FOUND-that-should-exist, etc.
      // Retrying will not help. Log with full context and ack.
      log.error("Permanent error on {} ({}): {}", notification.dto_id, notification.dto_type, e)
      tokens_to_ack.append(notification.ack_token)

  CQS.BatchAcknowledge(queue_name, tokens_to_ack)  // may be a proper subset of the batch

BatchAcknowledge takes a repeated ack_tokens list, so passing a subset is valid. Tokens that are omitted from the call are not acknowledged and will redeliver; all others are removed from the queue.

Re-enqueue: splitting a heavy notification¶

Some notifications imply much more work than a single ack lease can comfortably cover. A change to a widely-used design.v1 in a multi-store tenant fans out to every linked ESL — worst case, ~5 million label re-renders for a 500-store / 10 000-ESLs-per-store tenant. Holding one ack lease across that entire job has two failure modes:

If anything fails halfway through, the whole batch must be redone after the lease expires.
While the consumer is "under water" on this low-SLA work, urgent high-SLA notifications pile up behind it.

The ReEnqueue RPC lets a consumer give the notification back while pushing a stream of smaller replacement notifications into the same queue. The original is acked atomically on stream completion. The consumer is then free to dequeue normally — urgent items interleave naturally, and the smaller pieces of the original job are picked up in priority order.

Wire shape¶

ReEnqueue is a client-streaming RPC. The client sends a sequence of ReEnqueueRequest messages and receives one ReEnqueueResponse at the end:

rpc ReEnqueue(stream ReEnqueueRequest) returns (ReEnqueueResponse);

message ReEnqueueRequest {
  oneof payload {
    Header header  = 1;  // exactly one, first
    SplitItem item = 2;  // zero or more
    Footer footer  = 3;  // exactly one, last; commit signal
  }

  message Header {
    string queue_name = 1;   // same queue the original was dequeued from
    string ack_token  = 2;   // ack_token of the notification being split
  }
  message Footer {}          // intentionally empty; presence is the signal
}

message SplitItem {
  string dto_type = 1;
  oneof urn {
    string dto_id    = 2;
    string parent_id = 3;
  }
  string alias_name = 4;     // optional; valid with either dto_id or parent_id
}

message ReEnqueueResponse {
  bool  acknowledged   = 1;  // true only after the footer was received
  int32 items_enqueued = 2;
}

The shape mirrors the existing DtoChangeNotification.urn oneof, so the four legal forms on a SplitItem are: dto_id, dto_id + alias_name, parent_id, parent_id + alias_name. The only field-level rejection is "neither dto_id nor parent_id set".

Atomicity contract¶

The ack of the original happens after the final batch of items is durably enqueued. This produces an asymmetric guarantee:

Items without ack is recoverable. Already-enqueued items remain in the queue; the original returns via the normal lease-expiry path and the consumer's retry pushes a superset. Dedup on (dto_type, dto_id, alias_name) (see SlaPrefixTree.bulkEnqueue) makes this idempotent.
Ack without items is forbidden by construction. The server only acks after the Footer arrives and all preceding items have been flushed.

If the client disconnects before sending Footer, the call ends in CANCELLED, the items already flushed stay in the queue, and the original is not acked.

Inheritance¶

A SplitItem deliberately carries no sla_timestamp, tracking_id, or ack_token:

Field on the eventual `DtoChangeNotification`	Source
`dto_type`, `dto_id`/`parent_id`, `alias_name`	the `SplitItem`
`sla_timestamp`	inherited from the original notification
`tracking_id`	inherited from the original notification
`ack_token`	minted at the eventual re-dequeue

A consumer that wants to nudge itself to "finish one store before splitting the next" can do so by emitting tiered split-items whose effective SLA is implicitly slightly higher than the parent's: because the original's sla_timestamp is inherited and the queue ranks longer ids ahead of shorter ones at equal SLA, store-level splits naturally precede the design-level remainder, and link-level splits within a store naturally precede sibling stores.

Lease extension¶

Long re-enqueue streams (millions of items) easily exceed a default ack lease. The server extends the lease on the original notification automatically while the stream is open — every 10 seconds of wall clock or every 10 000 items, whichever comes first. Both client and server hold bounded memory; nothing buffers the full split-list.

Tiered split: worked example¶

A design.v1 is republished in a tenant where 500 stores each have 10 000 linked ESLs. studio-renderer receives one notification for the design and could try to re-render 5 000 000 images. Instead it uses ReEnqueue to split the work in two tiers:

First call — header carries the original design.v1 ack token. Body is 500 SplitItems, each in parent_id form scoped to one store (e.g. parent_id = t/farmersmarket/s/oslo-main/esls, dto_type = "studioeslimage.v1"). Footer commits. The original is acked; 500 per-store notifications are now in the renderer's queue.
For each per-store dequeue — the renderer reads one of the 500 parent_id notifications, then opens a new ReEnqueue stream. Header carries that notification's ack token. Body is 10 000 SplitItems in dto_id form, one per ESL barcode in that store. Footer commits. The per-store notification is acked; 10 000 per-ESL notifications are queued.
Steady state — studio-renderer now dequeues per-ESL items as fast as it can render. An urgent single-ESL change for a different design slots in by SLA priority; it does not wait behind the 5 M-item job.

Both tiers ride the same queue: ReEnqueue only ever lands items back in the originating queue.

Zero-item form: nack-with-no-replacement¶

A stream of just Header + Footer (no items) is valid and observably equivalent to a single-token BatchAcknowledge. It exists as the natural form of "I'm giving this back but have nothing to replace it with". No separate Nack RPC is needed.

Error contract¶

Condition	Status
`queue_name` blank or unknown	`NOT_FOUND`
`ack_token` unknown / already acked / lease expired	`FAILED_PRECONDITION`
First message is not `Header`, or a second `Header` arrives	`INVALID_ARGUMENT`
Item with no `dto_type`, or with neither `dto_id` nor `parent_id`	`INVALID_ARGUMENT`
Client disconnects before `Footer`	`CANCELLED`; items stay, original not acked

On any abort, items already flushed stay in the queue; the original ack is never granted unless the Footer was observed.

Relation to DTOflow Server¶

CQS does not store DTO payloads. It stores only change notifications: (DTO type, DTO ID or parent ID, SLA timestamp, ack token, optional alias name). The DTO data lives exclusively in the DTOflow Server.

This separation of concerns means CQS can be scaled, replaced, or maintained independently of the DTO store. It also means consumers always fetch fresh data from the DTO server at read time — there is no risk of consuming a stale payload from the queue.

Notification types¶

A DtoChangeNotification can address the changed DTO in one of three ways, depending on how the subscriber needs to process the change:

DTO ID (dto_id): A specific DTO was created, updated, or deleted. The consumer fetches that exact DTO by ID. This is the most common type.
DTO ID + alias (dto_id + alias_name): The notification carries an alias name alongside the DTO ID. This signals that the target of a named alias path has changed — not just that a DTO was written, but that the identity reachable via that alias is now different. Consumers subscribed via an alias use this to detect when a "pointer" they follow (e.g., by_storeesl) now resolves to a different DTO (or no DTO).
Parent ID (parent_id): A DTO under a given parent path was written, but the consumer does not need to track individual IDs — it only needs to know that something in that parent subtree changed and should re-process the whole group. Used for bulk invalidation (e.g., "re-evaluate all links under store s/001") and as the tiering primitive produced by ReEnqueue. May also carry an alias name when the parent notification is alias-scoped.

Pattern: alias-watch cleanup¶

Alias-qualified notifications enable a recurring deletion-cascade pattern in this codebase. The "Notification types" section above defines the mechanism; this section explains when and how to use it.

The problem¶

A consumer owns child DTOs keyed under one entity (typically an ESL barcode), but the trigger for deleting those children lives on a sibling DTO that is keyed under something else. Concrete example:

ecclink.v1 is keyed under t/{t}/s/{s}/links/{linkid}/ecc (linkid-scoped).
ecceslimage.v1 is keyed under t/{t}/s/{s}/esls/{barcode}/pages/{n}/ecceslimage (barcode-scoped).
When the ecclink for a link is deleted, ecc-image-render-service must delete every ecceslimage for the affected barcode — but the deletion notification on ecclink carries the linkid, not the barcode.

A primary-keyed subscription is therefore insufficient. The consumer needs a barcode-keyed handle on the deletion event.

The mechanic¶

The owning DTO declares an alias projecting to a barcode-keyed path. For ecclink.v1 this is the by_storeesl alias (ecclink.v1.proto):

__alias_name__   by_storeesl
__aliaspattern__ t/{tenantid}/s/{storeid}/esls/{barcode}/link/ecc

The consumer subscribes via CreateOrConfigureQueue with alias_name = "by_storeesl". From then on:

When the canonical ecclink is deleted, the alias mapping at …/esls/{barcode}/link/ecc clears.
An alias-qualified DtoChangeNotification fires: dto_id = the alias path (barcode-scoped), alias_name = "by_storeesl".
The consumer now has the barcode.

The cleanup step¶

The subscriber reads via the alias path:

Reader.Read(id = "t/{t}/s/{s}/esls/{barcode}/link/ecc", alias_name = "by_storeesl")
→ NOT_FOUND

That NOT_FOUND is the canonical deletion trigger — there is no separate "deleted" event. The subscriber then issues deleteChildren (or analogous batch delete) on the now-known barcode parent t/{t}/s/{s}/esls/{barcode}, which cascades to every per-page child it owns.

Worked trace: ECC barcode unlink¶

link-registry deletes link.v2 for a barcode.
ecclink-projector (subscribed to link.v2) sees the deletion and deletes the corresponding ecclink.v1. The by_storeesl alias mapping clears as a side effect.
ecc-image-render-service (subscribed to ecclink.v1 with alias_name = by_storeesl) receives the alias-qualified notification — carrying the barcode in dto_id.
EIRS reads via the alias path → NOT_FOUND → issues deleteChildren on t/{t}/s/{s}/esls/{barcode}. All ecceslimage pages for the ESL are gone.
eslimage-merger (subscribed to ecceslimage.v1) sees the deletions, re-runs its merge, finds no inputs, deletes eslimage.v1.
pricer-server (subscribed to eslimage.v1) sees the deletion → stops delivery → eventually writes storeeslstatus = OFFBOARDED, kicking off the terminal-cleanup cascade.

See ECC Rendering Pipeline for the full forward flow and ESL Hardware Lifecycle for the downstream lifecycle effects.

When to use it¶

Use the alias-watch cleanup pattern when a child DTO's lifetime is tied to the existence of a sibling DTO that is not a parent in the ID hierarchy. Current users:

studio-renderer watches studiolink.v1 via by_storeesl to clean up studioeslimage records when a link disappears.
ecc-image-render-service watches ecclink.v1 via by_storeesl to clean up ecceslimage records when a link disappears.

Aliases also exist for other purposes (e.g., by_item, by_sic, by_design, by_externalid) — those follow the same notification mechanic for reads but most are used for lookup, not cleanup. The cleanup-trigger pattern above is specific to using an alias as a stable barcode-keyed identity for an otherwise primary-keyed DTO.