openvibers/openvibe.events Commits

Add the cloud queue carrier (cloud-v1, SQS-compatible, QUEUE only, off unless CLOUD_QUEUE_URL is set) (#15)

OP OpenVibers committed 5 Oct 2026 · a89573e54c · 9 files, +342 −12

* Add the cloud queue carrier (cloud-v1, SQS-compatible, QUEUE only, off unless CLOUD_QUEUE_URL is set)

**Changes**
- `server/fabric/carriers/cloud.js:1` — new QUEUE-only `cloud-v1` adapter (classes `Set(['QUEUE'])`; `signal(rows,'QUEUE')` POSTs `{event_id, subscription_id}` as an SQS `SendMessage` form body to `CLOUD_QUEUE_URL`, SigV4 via node:crypto/node fetch; credentials optional; `unhealthyReason()` = `breaker open (cloud-v1)`; EWMA+breaker from `../signals.js`; `start/stop` no-ops — the poll is the recovery path; no module-load side effects, secret never logged).
- `server/fabric/carriers/index.js:25` — require `./cloud`; `:27` — `KNOWN` gains `cloud-v1`; `:49-52` — instantiate when `config.cloud.url` (beside the nats block).
- `server/config.js:99` — default `EVENTS_CARRIERS` gains `cloud-v1` (so setting the URL alone enables it, as with nats); `:113-122` — new `cloud` block: `CLOUD_QUEUE_URL`, `_REGION`, `_PREFIX`, `_ACCESS_KEY`, `_SECRET_KEY`.
- `README.md:145` — default list; `:151` — new `cloud-v1` carrier bullet.
- `.env.example:41-50` — documented `cloud-v1` vars; `:51` — default `EVENTS_CARRIERS`.
- `STATUS.json:23` — `updated: 2026-10-05`; `:36` — carrier-fabric feature notes `cloud-v1`.
- `test/fabric-cloud.test.js:1` — fake HTTP SQS endpoint: registry/config gate + unknown id refused, SigV4 POST of `{event_id, subscription_id}` (secret never logged, non-QUEUE kinds ignored), a due QUEUE delivery pinned to `cloud-v1` posted after commit, broker down (5 real failed POSTs) → breaker open → planner excludes with reason and `pg-v1` delivers, `EVENTS_CARRIERS_DISABLED=cloud-v1` → QUEUE on `pg-v1`.
- `server/fabric/rate-cards.json` — **unchanged on purpose**: every existing card carries a real owner price and `verified_at`/`source`, so there is no "not-yet-priced" form to copy. `cloud.js:112` references `rc-cloud-queue-v1` while it is absent from the file; `openvibe-sdk/placement` prices a paid provider without a card at `Infinity` (`placement.js:118-119`), so `cloud-v1` is eligible but never preferred over `pg-v1`/`valkey-v1` until the owner publishes the card. No price invented.

**Line-number drift from the prompt:** `index.js` had shifted — the nats block is 41-45, so the cloud instantiation landed at 49-52 and `KNOWN` at 27; `.env.example`'s nats/`EVENTS_CARRIERS` block ends at 42, not 39, so the cloud block is 41-50. `config.js` nats block was exactly 104-112, so the `cloud` block is at 113-122 as described. `CLOUD_QUEUE_PREFIX` is interpreted as an optional path prefix under `CLOUD_QUEUE_URL` (documented; unused when empty).

**Tests** — `nice -n 15 ov test --no-cache test/fabric-cloud.test.js test/fabric.test.js` → `2 passed in 8 s` (`fabric-cloud` 5/5; `fabric.test.js` passed, its Valkey parts reported skipped/`○` without a broker).

* Queue signals are bounded by a 3 s timeout and 8 in flight; the unused segment encoder is gone (review)

* The secrets crawl plants the cloud queue keys; the SSRF inventory reviews the cloud-v1 call site

Changes

ddiff --git a/.env.example b/.env.example
iindex 79feb7f..58d5cbe 100644
---- a/.env.example
++++ b/.env.example
@@@ -38,7 +38,17 @@ VALKEY_PREFIX=
  #NATS_SUBJECT_PREFIX=ov.events.
  #NATS_JETSTREAM=
  #NATS_STREAM=OV_EVENTS_STREAM
--#EVENTS_CARRIERS=pg-v1,valkey-v1,nats-v1,nats-js-v1
++# Optional cloud queue (ADR-042 decision 5): cloud-v1, a QUEUE-only push carrier that POSTs { event_id, subscription_id }
++# as an SQS SendMessage to a CLOUD_QUEUE_URL endpoint, SigV4-signed (node:crypto, no dependency). Credentials may be
++# omitted for a local endpoint; CLOUD_QUEUE_PREFIX is an optional path prefix under the URL. Off unless the URL is set;
++# EVENTS_CARRIERS_DISABLED=cloud-v1 rolls it back and pg-v1/valkey-v1 carries QUEUE. Its rate card (rc-cloud-queue-v1) is
++# not written until the owner sets its price: unpriced, the planner never prefers it.
++#CLOUD_QUEUE_URL=
++#CLOUD_QUEUE_REGION=us-east-1
++#CLOUD_QUEUE_PREFIX=
++#CLOUD_QUEUE_ACCESS_KEY=
++#CLOUD_QUEUE_SECRET_KEY=
++#EVENTS_CARRIERS=pg-v1,valkey-v1,nats-v1,nats-js-v1,cloud-v1
  #EVENTS_CARRIERS_DISABLED=
  # Retention in tiers (ADR-042 decision 8; README "Retention: three tiers"). Hot events (and their deliveries) older
  # than this move to the replay tier (events_archive) at the hourly prune.
ddiff --git a/README.md b/README.md
iindex 83342f9..02c8632 100644
---- a/README.md
++++ b/README.md
@@@ -142,12 +142,13 @@ node scripts/events-archive.js restore --month 2026-01 [--hold-days 30]
  
  How an event type is carried is its delivery policy (`events.delivery-policy@1`, [ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 1). The carrier class — TOPIC (fan-out, no ack), QUEUE (competing consumers, ack) or STREAM (retained, ordered per key) — is a derived, internal label, never in the envelope. PostgreSQL stays the record: every event and every delivery is committed before any carrier sees it, and the worker's lease is the only thing that makes a send happen. A publish commits, then the chosen carrier is signalled; a lost or duplicated signal changes latency, never the outcome.
  
--Carriers (`server/fabric/carriers`; `EVENTS_CARRIERS`, default `pg-v1,valkey-v1,nats-v1,nats-js-v1`):
++Carriers (`server/fabric/carriers`; `EVENTS_CARRIERS`, default `pg-v1,valkey-v1,nats-v1,nats-js-v1,cloud-v1`):
  
  - `pg-v1` — the poll. Always present, every class; the worker's 500 ms tick and its in-process `kick()` carry delivery. It is the floor every delivery falls back to.
  - `valkey-v1` — push, only when `VALKEY_URL` is set. **QUEUE:** after the commit, each new delivery is XADDed to a Valkey stream (`openvibe-sdk/queue`, consumer group `workers`); each Events process claims just that one row (`store.claimDelivery`, the same key-head and inflight-cap rules as the poll) and hands it to the worker's normal send — the item is acked either way, so a lost item only falls to the poll. **TOPIC:** the stored envelopes are PUBLISHed on a Valkey channel tagged with this process's id, and every process re-emits envelopes from other processes to its own SSE clients (deduped by event id), so an event published on one process reaches a browser connected to another. Health is an EWMA plus a breaker; while a breaker is open the planner excludes the carrier with a reason and `pg-v1` delivers.
  - `nats-v1` — optional NATS Core, TOPIC only, only when `NATS_URL` is set (`nats://[user:pass@]host:4222`, loopback or the private network; no TLS; subjects under `NATS_SUBJECT_PREFIX`, default `ov.events.`). After the commit the event ids (never the envelopes) are published on `<prefix>fabric.topic`; every process loads those from other processes from PostgreSQL and re-emits the committed rows to its own SSE clients (an id that is not committed is ignored; deduped by event id; visibility applied by the receiving gateway). Core NATS keeps nothing: a lost message reaches a remote browser on its next resume (`Last-Event-ID` replays from PostgreSQL). While it is not connected, the broker refuses its publish or subscribe (a permissions violation; the handshake probes the subject, a refused connection is re-opened every 30 s) or its breaker is open, the planner excludes it with that reason and `valkey-v1` or `pg-v1` carries TOPIC; `EVENTS_CARRIERS_DISABLED=nats-v1` stops it connecting at all. During a rolling deploy, old processes drop `{ids}` messages and new ones drop old `{rows}` messages, so cross-process TOPIC fan-out falls back to resume-from-PostgreSQL until every process runs the new version. Tests: `eval "$(scripts/test-nats.sh up)"` (a broker with JetStream) sets `OV_TEST_NATS_URL` for `test/fabric-nats.test.js` and `test/fabric-jetstream.test.js`.
  - `nats-js-v1` — optional JetStream, STREAM only, when `NATS_URL` is set and `NATS_JETSTREAM` is not `off` (nats-server 2.10+ with `-js`; `deploy/nats/nats-server.conf`). Spoken as JSON over the same dependency-free client (`server/fabric/nats-core.js`, request/reply), no client library. After the commit each new STREAM delivery's `{ event_id, subscription_id }` (never content) is published, with the stream's ack awaited, on `<prefix>fabric.stream.<key>` of the stream `NATS_STREAM` (default `OV_EVENTS_STREAM`, file storage, kept 24 h; `<key>` is a subject-safe hash of the ordering key). Every process running the delivery worker has its own durable push consumer (new messages only, explicit ack, one unacked message at a time) and, in stream order, claims exactly that row (`store.claimDelivery`: the key-head lease, `max_inflight` and due rules of the poll), sends it through the worker, then acks — whatever the claim answered. A message whose claim yields (another process or the poll has it, not its key's head yet, done, or never committed) leaves the row to the poll; a process that dies before its ack gets the message again after `ack_wait` and the claim, finding it delivered, sends nothing. Order per key is the store's, never the broker's. While it is not connected, the broker refuses it, the stream cannot be ensured (re-checked after every reconnect) or its breaker is open, the planner excludes it with that reason and `pg-v1` carries STREAM. The planner asks `events:durable` + `events:ordered` for STREAM, so only `nats-js-v1` and `pg-v1` can carry it; `rc-nats-js-v1` is cheaper than `rc-pg-v1`, so JetStream is preferred while healthy. Rollback: `EVENTS_CARRIERS_DISABLED=nats-js-v1` (or `NATS_JETSTREAM=off`) — it never connects and new STREAM deliveries go to `pg-v1`; rows already placed on it are delivered by the poll either way.
++- `cloud-v1` — optional SQS-compatible cloud queue, QUEUE only, only when `CLOUD_QUEUE_URL` is set (an `http(s)` endpoint; SigV4 with `CLOUD_QUEUE_REGION`, `CLOUD_QUEUE_ACCESS_KEY`, `CLOUD_QUEUE_SECRET_KEY` — credentials may be omitted for a local endpoint — and an optional path prefix `CLOUD_QUEUE_PREFIX`; signed with node:crypto, sent with node's fetch, no dependency). After the commit each new delivery's `{ event_id, subscription_id }` (never content) is POSTed as an SQS `SendMessage`; nothing is consumed here, so the endpoint's consumer claims the row and this process's poll recovers what the signal lost. While the breaker is open on consecutive signal errors the planner excludes it with reason `breaker open (cloud-v1)` and `pg-v1`/`valkey-v1` carries QUEUE; `EVENTS_CARRIERS_DISABLED=cloud-v1` (or no `CLOUD_QUEUE_URL`) leaves it out. Its offer names `rc-cloud-queue-v1`, which is not in `server/fabric/rate-cards.json` until the owner sets its price: `openvibe-sdk/placement` prices an offer without a card at Infinity and never assumes a paid provider free, so `pg-v1`/`valkey-v1` keep carrying QUEUE until that card is published.
  
  The planner (`server/fabric/planner.js`) calls `openvibe-sdk/placement` `plan()` per (carrier class, ordering key) with one `resource-offer@1` per adapter and the rate cards in `server/fabric/rate-cards.json`, caching decisions with hysteresis and recording a change in `delivery_placements`. A key's backlog stays on one carrier until it drains (a placement is pinned per (subscription, key), not per time window). An adapter named in `EVENTS_CARRIERS_DISABLED` is ineligible at once with reason `disabled by configuration` — the ADR's rollback; an unknown id in `EVENTS_CARRIERS` refuses to start.
  
ddiff --git a/STATUS.json b/STATUS.json
iindex 3700ce1..d655583 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -20,7 +20,7 @@
    "capabilitiesRegistered": true,
    "deployed": true,
    "deployment": "Deployed with `sudo ovhost deploy events`: /opt/openvibe.events, unit openvibe-events, 127.0.0.1:4300; public at https://events.openvibe.network (realtime SSE, health, token-guarded /api/v1/events, /api/v1/subscriptions, /api/v1/checkpoints; /api/v1/deliveries host-local); real traffic since 2026-09-23 (Live, Media, Chat, Games, Network and OpenRe publish; on 2026-09-23 8 subscriptions, held by Search, Tips, VIP, News and Reviews); restore drill passed 2026-09-23",
--  "updated": "2026-10-04",
++  "updated": "2026-10-05",
    "features": [
      "durable event store with the canonical envelope",
      "subscriptions with signed webhook delivery (v1 and v2), retry, dead letters and replay",
@@@ -33,7 +33,7 @@
      "queue-operation-64kb billing readings (platform.usage-sample@1) per production project and closed hour, posted to Billing through billing.usage.record (EVENTS_BILLING_INTERVAL_MS, off by default; docs/cutover-events-billing-readings.md)",
      "redaction tombstones",
      "per-actor limits (openvibe-sdk/limits; shared on Valkey)",
--    "carrier fabric (ADR-042): the pg-v1 poll beside a valkey-v1 push carrier for QUEUE (streams) and TOPIC (SSE fan-out) an optional nats-v1 NATS Core TOPIC carrier (NATS_URL) and an optional nats-js-v1 JetStream STREAM carrier (NATS_URL, NATS_JETSTREAM; deploy/nats), with a placement planner, per-key pinning, delivery_placements history and a GET /api/v1/placement explain API; EVENTS_CARRIERS[_DISABLED] switches it",
++    "carrier fabric (ADR-042): the pg-v1 poll beside a valkey-v1 push carrier for QUEUE (streams) and TOPIC (SSE fan-out) an optional nats-v1 NATS Core TOPIC carrier (NATS_URL) and an optional nats-js-v1 JetStream STREAM carrier (NATS_URL, NATS_JETSTREAM; deploy/nats) and an optional cloud-v1 SQS-compatible QUEUE carrier (CLOUD_QUEUE_URL; SigV4, no dependency), with a placement planner, per-key pinning, delivery_placements history and a GET /api/v1/placement explain API; EVENTS_CARRIERS[_DISABLED] switches it",
      "at_most_once delivery semantics (a failed attempt is final)"
    ]
  }
ddiff --git a/server/config.js b/server/config.js
iindex e743dec..b1225d2 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -94,11 +94,11 @@ function load(env = process.env) {
          // PostgreSQL (ADR-035): DATABASE_URL serves (PgBouncer), DATABASE_DIRECT_URL migrates (owner role).
          db: { url: env.DATABASE_URL || '', directUrl: env.DATABASE_DIRECT_URL || '' },
          // Carriers (ADR-042 decisions 1-2, 5; server/fabric/carriers): the adapters to instantiate and the ones the
--        // operator switch removes. pg-v1 is always present; valkey-v1 needs VALKEY_URL, nats-v1 NATS_URL and nats-js-v1
--        // NATS_URL with JetStream on, each silently absent without it. A disabled adapter is ineligible at once (reason
--        // `disabled by configuration`): the ADR's rollback.
++        // operator switch removes. pg-v1 is always present; valkey-v1 needs VALKEY_URL, nats-v1 NATS_URL, nats-js-v1
++        // NATS_URL with JetStream on and cloud-v1 CLOUD_QUEUE_URL, each silently absent without it. A disabled adapter
++        // is ineligible at once (reason `disabled by configuration`): the ADR's rollback.
          carriers: {
--            enabled: list(env.EVENTS_CARRIERS, ['pg-v1', 'valkey-v1', 'nats-v1', 'nats-js-v1']),
++            enabled: list(env.EVENTS_CARRIERS, ['pg-v1', 'valkey-v1', 'nats-v1', 'nats-js-v1', 'cloud-v1']),
              disabled: list(env.EVENTS_CARRIERS_DISABLED, []),
          },
          // Valkey (ADR-035): per-actor limit counters shared across processes; without it they count in this process.
@@@ -112,6 +112,16 @@ function load(env = process.env) {
              jetstream: env.NATS_JETSTREAM !== 'off',
              stream: env.NATS_STREAM || 'OV_EVENTS_STREAM',
          },
++        // Cloud queue (ADR-042 decision 5, optional): the cloud-v1 QUEUE carrier over an SQS-compatible endpoint
++        // (<CLOUD_QUEUE_URL>, signed with SigV4: CLOUD_QUEUE_REGION, CLOUD_QUEUE_ACCESS_KEY, CLOUD_QUEUE_SECRET_KEY).
++        // CLOUD_QUEUE_PREFIX is an optional path prefix under the URL. Off unless CLOUD_QUEUE_URL is set.
++        cloud: {
++            url: env.CLOUD_QUEUE_URL || '',
++            region: env.CLOUD_QUEUE_REGION || 'us-east-1',
++            prefix: env.CLOUD_QUEUE_PREFIX || '',
++            accessKey: env.CLOUD_QUEUE_ACCESS_KEY || '',
++            secretKey: env.CLOUD_QUEUE_SECRET_KEY || '',
++        },
          retentionDays: int(env.EVENTS_RETENTION_DAYS, 30),
          // retention.replay (ADR-042 decision 8): days an event stays readable in events_archive after it leaves the
          // hot store, counted from receipt like retentionDays. 0: no replay tier (pruned events are deleted, as before).
ddiff --git a/server/fabric/carriers/cloud.js b/server/fabric/carriers/cloud.js
nnew file mode 100644
iindex 0000000..f795b9a
---- /dev/null
++++ b/server/fabric/carriers/cloud.js
@@@ -0,0 +1,119 @@
++'use strict';
++/**
++ * cloud-v1 — a QUEUE-only push carrier over an SQS-compatible endpoint (ADR-042 decision 5), only when
++ * CLOUD_QUEUE_URL is set.
++ *
++ * QUEUE: after a publish commits, signal(rows) POSTs { event_id, subscription_id } — one message per new delivery —
++ * to the queue endpoint, SigV4-signed with node:crypto and sent with node's fetch. Nothing is consumed here: the
++ * endpoint's own consumer claims those rows, and this process's poll is the recovery path. A lost or duplicated
++ * message changes latency, never the outcome, because PostgreSQL stays the record and every delivery is committed
++ * before the signal.
++ *
++ * Health: an EWMA of signal latency and a breaker on consecutive errors (../signals.js). healthy() is false while the
++ * breaker is open, which drops cloud-v1 from the plan with a reason; pg-v1 (the poll) then carries delivery.
++ *
++ * Pricing: offer() names rc-cloud-queue-v1. No rate card with that id is written until the owner sets its price, so
++ * openvibe-sdk/placement prices this paid provider at Infinity and never assumes it free — pg-v1/valkey-v1 keep
++ * carrying QUEUE until the owner publishes the card.
++ */
++const crypto = require('crypto');
++const { createSignals } = require('../signals');
++
++const QUEUE_ACTION = 'SendMessage';
++const QUEUE_VERSION = '2012-11-05';
++const SERVICE = 'sqs';
++
++const sha256 = (data) => crypto.createHash('sha256').update(data).digest('hex');
++const hmac = (key, data) => crypto.createHmac('sha256', key).update(data).digest();
++
++/** The SigningKey/Authorization for one request; `headers` are the ones to sign (lower-case names, host included). */
++function signV4({ method, path: p, headers, payloadHash, region, service, accessKeyId, secretAccessKey, amzDate }) {
++    const names = Object.keys(headers).map((h) => h.toLowerCase()).sort();
++    const lower = Object.fromEntries(Object.entries(headers).map(([k, v]) => [k.toLowerCase(), String(v).trim().replace(/\s+/g, ' ')]));
++    const signedHeaders = names.join(';');
++    const canonical = [method, p, '', names.map((n) => `${n}:${lower[n]}\n`).join(''), signedHeaders, payloadHash].join('\n');
++    const day = amzDate.slice(0, 8);
++    const scope = `${day}/${region}/${service}/aws4_request`;
++    const toSign = ['AWS4-HMAC-SHA256', amzDate, scope, sha256(canonical)].join('\n');
++    const key = hmac(hmac(hmac(hmac(`AWS4${secretAccessKey}`, day), region), service), 'aws4_request');
++    const signature = crypto.createHmac('sha256', key).update(toSign).digest('hex');
++    return `AWS4-HMAC-SHA256 Credential=${accessKeyId}/${scope}, SignedHeaders=${signedHeaders}, Signature=${signature}`;
++}
++
++/** The queue URL: CLOUD_QUEUE_URL, with CLOUD_QUEUE_PREFIX (if any) added to its path. */
++function queueUrl(url, prefix) {
++    if (!prefix) return url;
++    const u = new URL(url);
++    u.pathname = `${u.pathname.replace(/\/+$/, '')}/${String(prefix).replace(/^\/+|\/+$/g, '')}`;
++    return u.toString();
++}
++
++function createCloudCarrier({ url, region = 'us-east-1', prefix = '', accessKey = '', secretKey = '', timeoutMs = 3000, concurrency = 8, clock = { now: () => Date.now() }, log = console, fetchImpl = fetch } = {}) {
++    const signals = createSignals({ now: () => clock.now() });
++    const endpoint = queueUrl(url, prefix);
++
++    async function send(row) {
++        const body = new URLSearchParams({
++            Action: QUEUE_ACTION,
++            Version: QUEUE_VERSION,
++            MessageBody: JSON.stringify({ event_id: row.event_id, subscription_id: row.subscription_id }),
++        }).toString();
++        const u = new URL(endpoint);
++        const amzDate = new Date(clock.now()).toISOString().replace(/[-:]/g, '').replace(/\.\d{3}/, '');
++        const payloadHash = sha256(body);
++        // AWS signs content-type (form data), the payload hash, the date and host. Credentials are optional so a local
++        // SQS-compatible endpoint without auth can be used; when present the request is SigV4-signed.
++        const headers = { host: u.host, 'content-type': 'application/x-www-form-urlencoded; charset=utf-8', 'x-amz-content-sha256': payloadHash, 'x-amz-date': amzDate };
++        const sent = { 'content-type': headers['content-type'], 'x-amz-content-sha256': payloadHash, 'x-amz-date': amzDate };
++        if (accessKey && secretKey) sent.authorization = signV4({ method: 'POST', path: u.pathname, headers, payloadHash, region, service: SERVICE, accessKeyId: accessKey, secretAccessKey: secretKey, amzDate });
++        // Bounded: a slow or hung queue must not hold the publish that signalled it (the poll still carries the row).
++        const res = await fetchImpl(endpoint, { method: 'POST', headers: sent, body, redirect: 'manual', signal: AbortSignal.timeout(timeoutMs) });
++        if (!res.ok) throw new Error(`queue answered ${res.status}`);
++    }
++
++    async function signalQueue(rows) {
++        const t0 = clock.now();
++        try {
++            // At most `concurrency` requests in flight, so a batch is bounded by about rows/concurrency timeouts.
++            for (let i = 0; i < rows.length; i += concurrency) await Promise.all(rows.slice(i, i + concurrency).map(send));
++            signals.record('queue', true, clock.now() - t0);
++        } catch (err) {
++            signals.record('queue', false);
++            log.warn(`[cloud-v1] queue signal: ${err.message}`);   // the credentials are never logged
++        }
++    }
++
++    return {
++        id: 'cloud-v1',
++        classes: new Set(['QUEUE']),
++
++        // Nothing to consume: the endpoint's consumer claims the rows and this process's poll recovers what the signal
++        // lost. start/stop exist to match the registry's adapter shape.
++        async start() {},
++        async stop() {},
++
++        signal(rows, kind) { return kind === 'QUEUE' ? signalQueue(rows) : undefined; },
++        healthy: () => signals.healthy('queue'),
++        unhealthyReason: () => (signals.healthy('queue') ? null : 'breaker open (cloud-v1)'),
++        signals,
++
++        offer() {
++            const ms = signals.ewma('queue') || 1;
++            return {
++                offer_id: 'cloud-v1',
++                kind: 'provider',
++                provider: 'cloud',
++                adapter: 'cloud-v1',
++                region: 'cloud',
++                trust: 'first-party',
++                capabilities: ['events:gateway'],
++                latency_ms: { publish_p95: ms, ack_p95: ms },
++                health: { status: 'up', checked_at: new Date(clock.now()).toISOString() },
++                pricing: { model: 'per-operation', unit: 'queue-operation', rate_card: 'rc-cloud-queue-v1' },
++                updated_at: new Date(clock.now()).toISOString(),
++            };
++        },
++    };
++}
++
++module.exports = { createCloudCarrier, signV4, queueUrl };
ddiff --git a/server/fabric/carriers/index.js b/server/fabric/carriers/index.js
iindex 911581f..e658f87 100644
---- a/server/fabric/carriers/index.js
++++ b/server/fabric/carriers/index.js
@@@ -4,12 +4,15 @@
   * carry, an offer, health, a signal() and a start/stop. pg-v1 is always present and carries every class; valkey-v1
   * is registered only when VALKEY_URL is set and carries QUEUE and TOPIC; nats-v1 is registered only when NATS_URL is
   * set and carries TOPIC over NATS Core; nats-js-v1 is registered when NATS_URL is set and NATS_JETSTREAM is not `off`
-- * and carries STREAM over JetStream. Without them STREAM is pg-v1's.
++ * and carries STREAM over JetStream; cloud-v1 is registered only when CLOUD_QUEUE_URL is set and carries QUEUE.
++ * Without them STREAM is pg-v1's.
   *
   * Config (server/config.js):
-- *   EVENTS_CARRIERS           ids to instantiate (default pg-v1,valkey-v1,nats-v1,nats-js-v1). An unknown id refuses to start.
++ *   EVENTS_CARRIERS           ids to instantiate (default pg-v1,valkey-v1,nats-v1,nats-js-v1,cloud-v1). An unknown id refuses to start.
   *   NATS_URL                  nats://[user:pass@]host:port of the cell's NATS (nats-v1, nats-js-v1); NATS_SUBJECT_PREFIX
   *                             its subjects; NATS_JETSTREAM=off leaves nats-js-v1 out; NATS_STREAM its stream name.
++ *   CLOUD_QUEUE_URL           the SQS-compatible endpoint of cloud-v1 (CLOUD_QUEUE_REGION, _PREFIX, _ACCESS_KEY,
++ *                             _SECRET_KEY); unset leaves cloud-v1 out.
   *   EVENTS_CARRIERS_DISABLED  ids excluded at once with reason `disabled by configuration` (the ADR's rollback).
   *
   * The planner asks offers(class) for one resource-offer@1 per eligible adapter and forClass(class) for the
@@@ -19,8 +22,9 @@ const { createPgCarrier } = require('./pg');
  const { createValkeyCarrier, newInstanceId } = require('./valkey');
  const { createNatsCarrier } = require('./nats');
  const { createJetStreamCarrier } = require('./jetstream');
++const { createCloudCarrier } = require('./cloud');
  
--const KNOWN = ['pg-v1', 'valkey-v1', 'nats-v1', 'nats-js-v1'];
++const KNOWN = ['pg-v1', 'valkey-v1', 'nats-v1', 'nats-js-v1', 'cloud-v1'];
  
  function createCarriers({ config, clock = { now: () => Date.now() }, log = console, valkey = null } = {}) {
      const enabled = new Set(config.carriers.enabled);
@@@ -43,6 +47,10 @@ function createCarriers({ config, clock = { now: () => Date.now() }, log = conso
              adapters.push(createJetStreamCarrier({ url: config.nats.url, subjectPrefix: config.nats.subjectPrefix, stream: config.nats.stream, clock, log, instanceId: newInstanceId() }));
          }
      }
++    // cloud-v1 signals over fetch in signal(); a disabled one is never signalled (signal() skips it), so it is inert.
++    if (config.cloud && config.cloud.url) {
++        adapters.push(createCloudCarrier({ url: config.cloud.url, region: config.cloud.region, prefix: config.cloud.prefix, accessKey: config.cloud.accessKey, secretKey: config.cloud.secretKey, clock, log }));
++    }
      for (const a of adapters) if (!a.disabledReason && off(a.id)) a.disabledReason = 'disabled by configuration';
  
      const get = (id) => adapters.find((a) => a.id === id) || null;
ddiff --git a/test/fabric-cloud.test.js b/test/fabric-cloud.test.js
nnew file mode 100644
iindex 0000000..a0e5430
---- /dev/null
++++ b/test/fabric-cloud.test.js
@@@ -0,0 +1,176 @@
++'use strict';
++// cloud-v1 (ADR-042 decision 5, optional): a QUEUE-only push carrier over an SQS-compatible endpoint, SigV4-signed.
++// The registry builds it only with CLOUD_QUEUE_URL, refuses an unknown id, and honours the disable switch. Against a
++// fake HTTP SQS endpoint: a due QUEUE delivery pinned to cloud-v1 is POSTed as {event_id, subscription_id}; a broker
++// that errors opens the breaker, so the planner excludes cloud-v1 with a reason and pg-v1 carries QUEUE; disabled,
++// QUEUE stays on pg-v1. The rate card is not written (no owner price yet), so the offer names rc-cloud-queue-v1 and
++// placement prices the paid provider at Infinity — pg-v1/valkey-v1 keep carrying QUEUE.
++const assert = require('assert');
++const nodeHttp = require('http');
++const { validate } = require('openvibe-contracts');
++const { boot, request, serviceToken, envelope, subscriber, suite } = require('./helpers');
++const { load } = require('../server/config');
++const { createCarriers } = require('../server/fabric/carriers');
++const { createPlanner, RATE_CARDS } = require('../server/fabric/planner');
++const { createCloudCarrier } = require('../server/fabric/carriers/cloud');
++
++const t = suite('fabric-cloud');
++const silent = { log() {}, warn() {}, error() {} };
++const admin = serviceToken('ops', ['events.delivery.admin']);
++const live = serviceToken('live', ['events.event.publish']);
++const policy = { class: 'domain', durability: 'optional', delivery_semantics: 'at_least_once' };
++
++// A fake SQS endpoint: it records every form body and answers 200, or `fail()` to answer 500 (a broker outage).
++async function fakeSqs() {
++    const calls = [];
++    let status = 200;
++    const server = nodeHttp.createServer((req, res) => {
++        const chunks = [];
++        req.on('data', (c) => chunks.push(c));
++        req.on('end', () => {
++            const raw = Buffer.concat(chunks).toString('utf8');
++            calls.push({ url: req.url, headers: req.headers, raw, body: Object.fromEntries(new URLSearchParams(raw)) });
++            res.statusCode = status;
++            res.end(status === 200 ? '<SendMessageResponse/>' : 'error');
++        });
++    });
++    await new Promise((r) => server.listen(0, '127.0.0.1', r));
++    return {
++        calls,
++        url: `http://127.0.0.1:${server.address().port}/queue/ov-events`,
++        fail: () => { status = 500; },
++        close: () => new Promise((r) => { server.closeAllConnections?.(); server.close(() => r()); }),
++    };
++}
++async function subscribe(base, slug, topic, url) {
++    const r = await request(base, 'POST', '/api/v1/subscriptions', { token: serviceToken(slug, ['events.subscription.manage']), body: { topic_pattern: topic, endpoint: url } });
++    assert.strictEqual(r.status, 201, r.text);
++    return r.body;
++}
++async function publish(base, env) {
++    const r = await request(base, 'POST', '/api/v1/events', { token: live, body: env });
++    assert.ok(r.status === 201 || r.status === 200, r.text);
++    return r.body;
++}
++async function putPolicy(base, pattern, body) {
++    const r = await request(base, 'PUT', `/api/v1/delivery-policies/${pattern}`, { token: admin, body });
++    assert.strictEqual(r.status, 200, r.text);
++    return r.body;
++}
++
++// ── Registry, offer, unpriced rate card, unknown id ─────────────────────────
++
++t('registry: cloud-v1 only with CLOUD_QUEUE_URL; QUEUE only; unknown id refused; the offer names the unpriced card', async () => {
++    const without = createCarriers({ config: load({ NODE_ENV: 'test' }), log: silent });
++    assert.strictEqual(without.get('cloud-v1'), null, 'no CLOUD_QUEUE_URL: no cloud-v1');
++
++    const reg = createCarriers({ config: load({ NODE_ENV: 'test', CLOUD_QUEUE_URL: 'http://127.0.0.1:1/queue/ov-events' }), log: silent });
++    const c = reg.get('cloud-v1');
++    assert.deepStrictEqual([...c.classes], ['QUEUE'], 'QUEUE only');
++    assert.ok(!c.disabledReason, 'enabled by default when the URL is set');
++    assert.ok(reg.forClass('QUEUE').some((a) => a.id === 'cloud-v1'));
++    assert.ok(reg.forClass('TOPIC').every((a) => a.id !== 'cloud-v1') && reg.forClass('STREAM').every((a) => a.id !== 'cloud-v1'));
++    const offer = c.offer();
++    assert.ok(validate('platform.resource-offer@1', offer).valid, JSON.stringify(offer));
++    assert.strictEqual(offer.capabilities.includes('events:gateway'), true);
++    assert.strictEqual(offer.pricing.rate_card, 'rc-cloud-queue-v1');
++    // The card is absent on purpose: no owner price yet, so the paid provider is never assumed free by placement.
++    assert.strictEqual(RATE_CARDS.find((x) => x.id === 'rc-cloud-queue-v1'), undefined, 'no AI-written price');
++
++    assert.throws(() => createCarriers({ config: load({ NODE_ENV: 'test', EVENTS_CARRIERS: 'pg-v1,kafka-v1' }) }), /unknown carrier adapter "kafka-v1"/);
++});
++
++// ── Signalling (fake SQS endpoint) ──────────────────────────────────────────
++
++t('cloud-v1 QUEUE: a signal POSTs {event_id, subscription_id} with SigV4 and never logs the secret', async () => {
++    const sqs = await fakeSqs();
++    const warnings = [];
++    const log = { log() {}, warn: (m) => warnings.push(String(m)), error() {} };
++    const carrier = createCloudCarrier({ url: sqs.url, region: 'eu-west-3', accessKey: 'AKIA_TEST', secretKey: 'very-secret-key', log });
++    try {
++        await carrier.signal([{ event_id: 'evt_1', subscription_id: 'sub_1' }], 'QUEUE');
++        assert.strictEqual(sqs.calls.length, 1);
++        const c = sqs.calls[0];
++        assert.strictEqual(c.body.Action, 'SendMessage');
++        assert.strictEqual(c.body.Version, '2012-11-05');
++        assert.deepStrictEqual(JSON.parse(c.body.MessageBody), { event_id: 'evt_1', subscription_id: 'sub_1' });
++        assert.match(c.headers.authorization, /^AWS4-HMAC-SHA256 Credential=AKIA_TEST\/\d{8}\/eu-west-3\/sqs\/aws4_request,/);
++        assert.ok(c.headers['x-amz-date'] && c.headers['x-amz-content-sha256'].length === 64);
++        await carrier.signal([{ event_id: 'evt_2', subscription_id: 'sub_2' }], 'TOPIC');   // not this carrier's class
++        assert.strictEqual(sqs.calls.length, 1);
++        assert.ok(warnings.every((w) => !w.includes('very-secret-key')), 'the secret is never logged');
++    } finally { await sqs.close(); }
++});
++
++t('cloud-v1 QUEUE: a due delivery pinned to cloud-v1 is posted to the endpoint after the commit', async () => {
++    const sqs = await fakeSqs();
++    const h = await boot({ env: {
++        EVENTS_CARRIERS: 'pg-v1,cloud-v1',
++        CLOUD_QUEUE_URL: sqs.url, CLOUD_QUEUE_REGION: 'eu-west-3',
++        CLOUD_QUEUE_ACCESS_KEY: 'AKIA_TEST', CLOUD_QUEUE_SECRET_KEY: 'very-secret-key',
++    }, worker: 'off' });
++    const stub = await subscriber(() => 204);
++    try {
++        const sub = await subscribe(h.base, 'cloud', 'live.cl.*', stub.url);
++        await putPolicy(h.base, 'live.cl.*', { ...policy, ordering: { scope: 'key', key: 'subject' } });
++        // The first delivery is planned on pg-v1 (cloud-v1 is unpriced); rewriting its carrier models a backlog already
++        // placed on cloud-v1, so the next same-key event is pinned there and its signal goes to the endpoint.
++        const E0 = envelope('live', { event_type: 'live.cl.x', subject: { type: 'stream', id: 'K' } });
++        await publish(h.base, E0);
++        await h.store.db.prepare("UPDATE deliveries SET carrier = 'cloud-v1' WHERE event_id = ? AND subscription_id = ?").run(E0.event_id, sub.id);
++        sqs.calls.length = 0;
++        const E1 = envelope('live', { event_type: 'live.cl.x', subject: { type: 'stream', id: 'K' } });
++        await publish(h.base, E1);
++        assert.strictEqual(sqs.calls.length, 1, 'the pinned QUEUE delivery was posted');
++        assert.deepStrictEqual(JSON.parse(sqs.calls[0].body.MessageBody), { event_id: E1.event_id, subscription_id: sub.id });
++    } finally { await stub.close(); await h.stop(); await sqs.close(); }
++});
++
++// ── The planner ─────────────────────────────────────────────────────────────
++
++t('planner: cloud-v1 is unpriced while up; a down broker (breaker open) excludes it with a reason and pg-v1 delivers', async () => {
++    const sqs = await fakeSqs();
++    const reg = createCarriers({ config: load({ NODE_ENV: 'test', CLOUD_QUEUE_URL: sqs.url, CLOUD_QUEUE_REGION: 'eu-west-3' }), log: silent });
++    const planner = createPlanner({ registry: reg });
++    const c = reg.get('cloud-v1');
++    try {
++        // Up: eligible, but no rate card means Infinity cost, so pg-v1 is planned (the paid provider is not assumed free).
++        let ex = planner.explain('QUEUE', 'stream:K', policy);
++        assert.strictEqual(ex.carrier, 'pg-v1');
++        const up = ex.result.candidates.find((x) => x.id === 'cloud-v1');
++        assert.strictEqual(up.eligible, true);
++
++        // Broker down: five failed signals open the breaker; the planner excludes cloud-v1 with the reason and pg-v1 carries.
++        sqs.fail();
++        for (let i = 0; i < 5; i++) await c.signal([{ event_id: 'evt_x', subscription_id: 'sub_x' }], 'QUEUE');
++        assert.strictEqual(c.healthy(), false, 'the breaker is open after consecutive errors');
++        ex = planner.explain('QUEUE', 'stream:K', policy);
++        assert.strictEqual(ex.carrier, 'pg-v1', 'pg-v1 (the poll) delivers');
++        const down = ex.result.candidates.find((x) => x.id === 'cloud-v1');
++        assert.strictEqual(down.eligible, false);
++        assert.strictEqual(down.excluded_because, 'breaker open (cloud-v1)');
++    } finally { await sqs.close(); }
++});
++
++t('a hung queue cannot hold the publish: each request is bounded by timeoutMs and counts as a failure', async () => {
++    const silentLog = { warn() {}, info() {}, error() {} };
++    // A fetch that never answers unless its abort signal fires.
++    const hung = (_url, opts) => new Promise((_resolve, reject) => { opts.signal.addEventListener('abort', () => reject(opts.signal.reason)); });
++    const c = createCloudCarrier({ url: 'http://queue.test/q', timeoutMs: 50, concurrency: 4, log: silentLog, fetchImpl: hung });
++    const t0 = Date.now();
++    await c.signal(Array.from({ length: 8 }, (_, i) => ({ event_id: `evt_${i}`, subscription_id: 'sub' })), 'QUEUE');
++    const took = Date.now() - t0;
++    assert.ok(took < 1000, `eight hung requests finish in about two timeouts (took ${took} ms)`);
++});
++
++t('EVENTS_CARRIERS_DISABLED=cloud-v1: QUEUE stays on pg-v1 and the reason is configuration', async () => {
++    const reg = createCarriers({ config: load({ NODE_ENV: 'test', CLOUD_QUEUE_URL: 'http://127.0.0.1:1/queue/ov-events', EVENTS_CARRIERS_DISABLED: 'cloud-v1' }), log: silent });
++    assert.strictEqual(reg.get('cloud-v1').disabledReason, 'disabled by configuration');
++    const ex = createPlanner({ registry: reg }).explain('QUEUE', 'k', policy);
++    assert.strictEqual(ex.carrier, 'pg-v1');
++    const cand = ex.result.candidates.find((x) => x.id === 'cloud-v1');
++    assert.strictEqual(cand.eligible, false);
++    assert.strictEqual(cand.excluded_because, 'disabled by configuration');
++});
++
++t.run();
ddiff --git a/test/security-secrets.test.js b/test/security-secrets.test.js
iindex c5698a7..114010b 100644
---- a/test/security-secrets.test.js
++++ b/test/security-secrets.test.js
@@@ -26,7 +26,12 @@ const t = suite('security-secrets');
  let w;
  const needles = {};
  // Secrets Events does read from its environment, each planted in the world below as a sentinel the crawl looks for.
--const PLANTED = { OV_OAUTH_CLIENT_SECRET: `ovsecret-events-client-${crypto.randomBytes(12).toString('hex')}` };
++const PLANTED = {
++    OV_OAUTH_CLIENT_SECRET: `ovsecret-events-client-${crypto.randomBytes(12).toString('hex')}`,
++    // cloud-v1 (an SQS-compatible queue): signs requests, never echoed.
++    CLOUD_QUEUE_ACCESS_KEY: `ovsecret-events-cloud-access-${crypto.randomBytes(12).toString('hex')}`,
++    CLOUD_QUEUE_SECRET_KEY: `ovsecret-events-cloud-secret-${crypto.randomBytes(12).toString('hex')}`,
++};
  
  t('Events reads no secret from its environment but the planted ones', async () => {
      const names = new Set();
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex 7085034..e2fb982 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -221,6 +221,7 @@ const REVIEWED = {
      'server/egress.js http.request': [1, 'the guarded poster\'s https.request (default requestImpl): its own lookup pins the checked address'],
      'server/egress.js requestImpl': [1, 'the guarded poster\'s request, after parseAppEndpoint and with the guarded lookup'],
      'server/billing.js fetch': [1, 'Billing readings: OV_BILLING_INTERNAL_URL /api/v1/usage (operator config, a fixed path); the reading is built from counts, never a request input'],
++    'server/fabric/carriers/cloud.js fetch': [1, 'the cloud-v1 queue: CLOUD_QUEUE_URL (operator config, an SQS-compatible endpoint), redirect: manual, bounded by timeoutMs; never a request input'],
      'server/fabric/nats-core.js net.connect': [1, 'the nats-v1 broker: NATS_URL (operator config, nats:// on loopback or the private network); never a request input'],
  };
  /** JS source with comments blanked (strings, template literals and regex literals kept). */