openvibers/openvibe.media Commits

Media S3: per-zone usage — requests, egress and stored bytes counted, every closed hour sent as media.usage.recorded (contracts 0.175.0) (#111)

OP OpenVibers committed 29 days ago · 6211b40726 · 10 files, +460 −9

Media now counts each zone's use of the S3 surface and sends every closed UTC hour as **`media.usage.recorded`** (contracts 0.175.0, ADR-031 section 8). This was the last ADR-031 item marked "not emitted yet".

**Counting (`server/s3/usage.js`):**
- **When:** a request is counted against its zone when its answer finishes. The zone is set once `zoneFor` resolves it, so `NoSuchBucket`, bad signatures and keys scoped elsewhere count nowhere.
- **Series:**

| Series | Capability | Unit |
|---|---|---|
| Writes (PUT, multipart, POST) | `media.object.upload` | `requests` |
| Deletes (DELETE, DeleteObjects, abort) | `media.object.delete` | `requests` |
| Lists (ListObjects, ListParts, ListMultipartUploads, HeadBucket) | `media.object.list` | `requests` |
| Reads (GET, HEAD) | `media.object.read` | `requests`, plus `egress_bytes` for the GETs answered |
| Stored bytes | `media.object.upload` | `byte_hours`: each active zone's live bytes sampled every minute, a sixtieth each time |

- **Refusals:** counted in `errors` by S3 code (`media.s3.no_such_key` …), the 20 largest codes and the last 10 samples (time, code, status). `s3error` records the code on the response.
- **Storage:** counts stay in memory and are added to `media_s3_usage` every 30 s and at shutdown (migration 0011, expand). A failed flush puts them back.
- **Sending:** every 5 minutes the closed hours go to the outbox (`events.recordUsage`, subject the project, priority low), in the transaction that marks them sent. Counts that land after an hour was sent raise its `revision`, and it is sent again with its new total.
- **Validation:** each rollup is checked against `media.usage.recorded@1` and `checkUsageRecorded` (its resource is a zone of its project). One that fails stays unsent and is logged.
- **Retention:** sent rows go after 40 days.
- **Sandbox zones** roll up too: unlike outcome events, a rollup says its env, and only project usage views read it.
- **What a rollup never contains:** a subject, key, object key or address.
- **Drill mode:** the timers start inside the drill guard, like the others.

**Readers:** Network adds the rollups to the project's usage (Network#192). Zone sums them in its usage view (next PR).

**Also:**
- `openvibe-contracts` v0.173.0 → v0.175.0.
- The README's stale contracts versions are corrected (`pins.test.js`).

**Tests:** `test/s3-usage.test.js` drives every series through the AWS SDK. It then checks the stored counts, the byte-hour sample, that each closed hour is sent once and validates, the late-count revision 2, that an invalid rollup stays unsent, and the pruning. The full suite passes locally: `pins.test.js` was the one failure on the first run, and it passes since the README fix.

Changes

ddiff --git a/README.md b/README.md
iindex 4ecfc8e..c8a692b 100644
---- a/README.md
++++ b/README.md
@@@ -45,7 +45,7 @@ without `DATABASE_URL`, development uses an embedded PGlite database in `data/pg
    export and deletion), OpenVibe.Live (`live.lineage.resolve`), OpenVibe.Events (the outbox relay and
    the account and revocation subscriptions)
  - Backblaze B2 and Cloudflare R2 (S3 API) when configured; `ffmpeg`/`ffprobe` on the host
--- `openvibe-contracts` v0.143.0, `openvibe-sdk` v0.44.0 (tokens, events outbox, usage readings, per-actor limits),
++- `openvibe-contracts` v0.175.0, `openvibe-sdk` v0.44.0 (tokens, events outbox, usage readings, per-actor limits),
    `openvibe-shared` v4.1.3, pinned by release tarball
  
  ## Capabilities
@@@ -129,7 +129,7 @@ server/
                           storage.orphans.scan (monthly storage orphan report, service-wide, report only),
                           storage.move.cleanup (the alert for placement moves whose delete keeps failing, service-wide)
    client-ip.js           trust proxy = loopback; req.ip is the only client address
--(openvibe-shared v4.1.3, openvibe-contracts v0.173.0, openvibe-sdk v0.44.0: pinned release tarballs, installed by npm)
++(openvibe-shared v4.1.3, openvibe-contracts v0.175.0, openvibe-sdk v0.44.0: pinned release tarballs, installed by npm)
  scripts/smoke-test.sh    end-to-end smoke test (boots a temp instance)
  scripts/reconcile-objects.js / object-invariant.js / no-good-copy-report.js   object-model operator tools
  scripts/media-jobs.js     list jobs, run the size-invariant scan (dry run by default), approve/cancel proposals
@@@ -749,7 +749,7 @@ Rollback: ovhost puts the previous sha back by itself when `/api/ready` does not
  restart; afterwards `sudo ovhost rollback media --to <sha>`.
  
  <!-- versions:start -->
--- openvibe-contracts: v0.143.0
++- openvibe-contracts: v0.175.0
  - openvibe-sdk: v0.44.0
  - openvibe-shared: v4.1.3
  <!-- versions:end -->
ddiff --git a/docs/s3.md b/docs/s3.md
iindex b4bf56a..4a12667 100644
---- a/docs/s3.md
++++ b/docs/s3.md
@@@ -13,6 +13,7 @@ a named object zone of one project environment. Zones are made and deleted only
  | Access keys | `server/s3/keys.js` | `ovk_` keys for one project environment: read, write or both, and optionally limited to some zones. Secrets are encrypted with `MEDIA_S3_KEY_SECRET` and shown once. |
  | Zones | `server/s3/zones.js` | Media's copy of Zone's records. It resolves bucket names, gives each zone's namespace its quota, and runs zone deletion. |
  | S3 routes | `server/s3/routes.js` | The operations below, dispatched by `Host` (`MEDIA_S3_HOSTS`). |
++| Usage | `server/s3/usage.js` | Counts each zone's requests, egress and stored bytes, and sends every closed hour as `media.usage.recorded` (see "Usage" below). |
  | Control doors | `server/s3/admin-routes.js` | `PUT/GET /internal/zones/:id` (Zone only, `media.zone.provision`) and `/api/v1/projects/:project/access-keys` (`media.access_key.manage`, on behalf of a project admin). |
  
  ## Operations
@@@ -39,10 +40,35 @@ a named object zone of one project environment. Zones are made and deleted only
  - **HEAD through a proxy.** Cloudflare forwards a HEAD as a GET for paths it may cache (by extension: `.jpg`, `.bin`, `.zip` …), but SigV4 signs the method. A header-signed request with no body that arrives as GET and verifies only as HEAD is served as the GET, and the proxy returns just the headers. A presigned URL never gets this, because one may be shared for a HEAD alone; nor does any other method. Every S3 response is `Cache-Control: private, no-store`, so the edge never caches one (`cf-cache-status: BYPASS`). The fix at the edge itself is a DNS-only `s3.openvibe.media`.
  - **No browser execution.** Every response carries `Content-Security-Policy: sandbox; default-src 'none'` and `nosniff`, whatever type the object declares.
  
++## Usage
++
++Media is the only counter (ADR-031 section 8). Every request that reaches a zone counts against it when its answer is
++sent. Every closed UTC hour goes out as one `media.usage.recorded` event per series (`common.usage-recorded@1`):
++
++| Series | Capability | Unit |
++|---|---|---|
++| Stored bytes | `media.object.upload` | `byte_hours`: the zone's live bytes sampled each minute, a sixtieth each time |
++| Writes (PUT, multipart, POST) | `media.object.upload` | `requests` |
++| Deletes (DELETE, DeleteObjects, abort) | `media.object.delete` | `requests` |
++| Lists (ListObjects, ListParts, ListMultipartUploads, HeadBucket) | `media.object.list` | `requests` |
++| Reads (GET, HEAD) | `media.object.read` | `requests`, and `egress_bytes` for the GETs answered |
++
++Each rollup names its zone twice: `resource` is the zone's OVRN and `dimension` its id in lowercase. A refused request
++counts in `errors`, by its S3 code (`media.s3.no_such_key`, …), with the last ten failures as samples. A request that
++names no zone of its key (`NoSuchBucket`, a bad signature) counts nowhere.
++
++- **Storage:** counts are kept in memory and added to `media_s3_usage` every 30 s, and at shutdown.
++- **Sending:** every 5 minutes, the closed hours go to the outbox, in the transaction that marks them sent. Counts that land after an hour was sent raise its `revision`, and it is sent again with its new total.
++- **Validation:** a rollup that does not match the contract stays unsent and is logged.
++- **Retention:** sent rows go after 40 days.
++- **Sandbox zones** are rolled up too: the rollup says its `env`.
++- **Who reads them:** Network adds the rollups to the project's usage and Zone sums them in its usage view. They are counts, never money.
++
  ## Tests
  
  - `test/s3-zones.test.js` runs end to end with the AWS SDK for JavaScript v3: every operation, presigned URLs, isolation, quotas, holds, two doors and zone deletion.
  - `test/s3-multipart.test.js` uses the SDK's `Upload` helper (parallel parts) and the raw commands: S3's rules, ETags, abort, quota and expiry.
++- `test/s3-usage.test.js` counts every series through the SDK, then checks the rollups against the contract, their revisions and the pruning.
  - `test/s3-sigv4.test.js` checks the worked examples of the S3 API reference, including the chunk signature chain.
  
  ## Using it
ddiff --git a/migrations/0011_s3_usage.sql b/migrations/0011_s3_usage.sql
nnew file mode 100644
iindex 0000000..96b43d8
---- /dev/null
++++ b/migrations/0011_s3_usage.sql
@@@ -0,0 +1,22 @@
++-- phase: expand
++-- Per-zone usage on the S3 surface (ADR-031 section 8, server/s3/usage.js): the counts of one zone's series in one UTC
++-- hour, added to every 30 s from memory. A closed hour is sent as media.usage.recorded (common.usage-recorded@1) and
++-- marked with the revision it was sent at; counts that land after it raise the revision, and the hour is sent again.
++-- Counts only: no key, object key, address or subject is kept.
++CREATE TABLE media_s3_usage (
++    zone_id         text COLLATE "C" NOT NULL,
++    project_id      text COLLATE "C" NOT NULL,
++    env             text COLLATE "C" NOT NULL,
++    window_start    bigint NOT NULL,                             -- ms since the epoch, on the hour (UTC)
++    capability      text COLLATE "C" NOT NULL,                   -- media.object.upload | read | list | delete
++    unit            text COLLATE "C" NOT NULL,                   -- requests | egress_bytes | byte_hours
++    quantity        bigint NOT NULL DEFAULT 0,
++    errors          bigint NOT NULL DEFAULT 0,
++    error_codes     text NOT NULL DEFAULT '{}',                  -- { "media.s3.<code>": n }, the 20 largest
++    samples         text NOT NULL DEFAULT '[]',                  -- the last 10 failures: { at, code, status }
++    revision        integer NOT NULL DEFAULT 1,
++    sent_revision   integer,                                     -- the revision last sent; NULL: not sent yet
++    updated_at      bigint NOT NULL,
++    PRIMARY KEY (zone_id, window_start, capability, unit)
++);
++CREATE INDEX media_s3_usage_unsent ON media_s3_usage (window_start) WHERE sent_revision IS NULL OR sent_revision < revision;
ddiff --git a/package-lock.json b/package-lock.json
iindex fac3075..c84e2e4 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -16,7 +16,7 @@
          "express": "^4.19.2",
          "iovalkey": "^0.4.0",
          "multer": "^1.4.5-lts.1",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.173.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.175.0",
          "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
          "pg": "^8.23.0",
@@@ -1847,9 +1847,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.173.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.173.0",
--      "integrity": "sha512-6zHbU9zghIYDd3PPGtgMTsIgxrWwsoBHO/rvuLW1rghr6kRK6VwCmSMZRcjQbcPODPH1XrUAFuLg/T1K4nOCMQ==",
++      "version": "0.175.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.175.0",
++      "integrity": "sha512-AJNkb/g7mMQDi1oGPiKpVVEmk+oGvzJYLsgxCzd7bPkxN7aos3y/VKh7+tAlfnAOKTajpOTAz5cvLXcGJQHiiQ==",
        "license": "AGPL-3.0-only",
        "dependencies": {
          "ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex c661df0..7ac6872 100644
---- a/package.json
++++ b/package.json
@@@ -20,7 +20,7 @@
      "express": "^4.19.2",
      "iovalkey": "^0.4.0",
      "multer": "^1.4.5-lts.1",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.173.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.175.0",
      "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
      "pg": "^8.23.0",
ddiff --git a/server/events.js b/server/events.js
iindex d4348eb..e052820 100644
---- a/server/events.js
++++ b/server/events.js
@@@ -166,6 +166,34 @@ async function enqueue(eventType, appId, subject, data, priority) {
      return env;
  }
  
++/**
++ * Queue one closed hour of a zone's usage (media.usage.recorded, common.usage-recorded@1; server/s3/usage.js) inside the
++ * transaction that marks it sent. Unlike outcomes, sandbox zones count too: the rollup says its env, and only the
++ * project's usage views (Network, Zone) read it. Subject: the project.
++ */
++async function recordUsage(payload) {
++    if (!outbox) return null;
++    const env = await outbox.enqueue(inTx(), {
++        event_type: 'media.usage.recorded',
++        actor: { type: 'service', id: 'media' },
++        subject: { type: 'project', id: payload.project_id },
++        visibility: 'internal',
++        priority: 'low',
++        payload,
++    });
++    stats.queued++;
++    return env;
++}
++
++/** Does a rollup match media.usage.recorded@1 and name a resource of its own project? { valid, errors } */
++function checkUsage(payload) {
++    const contracts = require('openvibe-contracts');
++    const v = contracts.validate('media.usage.recorded@1', payload);
++    if (!v.valid) return { valid: false, errors: v.errors.map((e) => `${e.path || '/'} ${e.message}`).join('; ') };
++    const c = contracts.usage && contracts.usage.checkUsageRecorded ? contracts.usage.checkUsageRecorded(payload) : { valid: true, errors: [] };
++    return c.valid ? { valid: true, errors: '' } : { valid: false, errors: c.errors.map((e) => `${e.path || '/'} ${e.message}`).join('; ') };
++}
++
  /**
   * Queue a Search document or tombstone (media.index_document.*, ./public/search-documents.js) as is:
   * no tenant field, low priority. Same rule as record(): inside the transaction that records the push.
@@@ -285,4 +313,4 @@ function _reset() {
      stats.queued = 0; stats.lastError = null;
  }
  
--module.exports = { init, initWriter, record, recordPlacement, recordIndexDocument, recordModeration, recordJob, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, PLACEMENT_TYPES, JOB_TRANSITIONS, _reset };
++module.exports = { init, initWriter, record, recordPlacement, recordIndexDocument, recordModeration, recordJob, recordUsage, checkUsage, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, PLACEMENT_TYPES, JOB_TRANSITIONS, _reset };
ddiff --git a/server/index.js b/server/index.js
iindex 1bf03f9..1dfb5bf 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -296,6 +296,15 @@ const ready = (async () => {
          every(60 * 1000, async () => {
              try { const r = await require('./s3/zones').sweepDeleting(); if (r.tombstoned || r.purged) console.log(`[Zones] ${r.zones} deleting zone(s): ${r.tombstoned} tombstoned, ${r.purged} purged`); } catch (err) { console.warn('[Zones] deletion sweep:', err.message); }
          });
++        // Per-zone usage on the S3 surface (ADR-031 section 8): counts flushed every 30 s, stored bytes sampled every
++        // minute, and every closed hour sent as media.usage.recorded (server/s3/usage.js).
++        const s3usage = require('./s3/usage');
++        every(30 * 1000, async () => { try { await s3usage.flush(); } catch (err) { console.warn('[S3] usage flush:', err.message); } });
++        every(60 * 1000, async () => { try { await s3usage.sample(); } catch (err) { console.warn('[S3] usage sample:', err.message); } });
++        every(5 * 60 * 1000, async () => {
++            try { const n = await s3usage.emit(); if (n) console.log(`[S3] ${n} usage rollup(s) sent`); } catch (err) { console.warn('[S3] usage rollups:', err.message); }
++            try { await s3usage.prune(); } catch (err) { console.warn('[S3] usage prune:', err.message); }
++        });
          every(60 * 60 * 1000, () => thumbService.cleanupOldThumbnails());  // stale live thumbs
          // Object model: purge native objects whose soft-delete retention has passed (held ones are kept).
          every(60 * 60 * 1000, async () => {
@@@ -389,6 +398,7 @@ const ready = (async () => {
              () => require('./events')._reset(),
              () => { const v = require('./actor-limits').valkey(); if (v) v.close().catch(() => {}); },
              () => { for (const t of timers) clearInterval(t); },
++            () => require('./s3/usage').flush().catch((err) => console.warn('[S3] usage flush at stop:', err.message)),
          ],
          close: [() => db.close()],
          deadlineMs: 70000,
ddiff --git a/server/s3/routes.js b/server/s3/routes.js
iindex 90fe81b..fcc0477 100644
---- a/server/s3/routes.js
++++ b/server/s3/routes.js
@@@ -43,6 +43,7 @@ const model = require('../objects/model');
  const namespaces = require('../objects/namespaces');
  const sigv4 = require('./sigv4');
  const keys = require('./keys');
++const usage = require('./usage');
  const zones = require('./zones');
  const mp = require('./multipart');
  
@@@ -83,6 +84,7 @@ function xml(res, status, body) {
  
  function s3error(req, res, code, message, extra = '') {
      const status = STATUS[code] || 400;
++    res.locals.s3code = code; // the zone's usage counts refusals by code (server/s3/usage.js)
      baseHeaders(res);
      if (req.method === 'HEAD') return res.status(status).end();
      xml(res, status, `<Error><Code>${esc(code)}</Code><Message>${esc(message)}</Message><Resource>${esc(req.originalUrl.split('?')[0])}</Resource>`
@@@ -129,6 +131,7 @@ async function zoneFor(req, res, bucket) {
      const zone = await zones.byName(key.project_id, key.env, bucket);
      if (!zone || zone.state !== 'active') { s3error(req, res, 'NoSuchBucket', 'The specified bucket does not exist'); return null; }
      if (key.zones.length && !key.zones.includes(zone.id)) { s3error(req, res, 'AccessDenied', 'This access key is not scoped to this bucket'); return null; }
++    req.s3.zone = zone; // from here the request counts in the zone's usage
      return zone;
  }
  
@@@ -668,6 +671,8 @@ router.use(async (req, res) => {
          const slash = p.indexOf('/', 1);
          const bucket = decodeURIComponent(slash < 0 ? p.slice(1) : p.slice(1, slash));
          const rest = slash < 0 ? '' : p.slice(slash + 1);
++        req.s3.hasKey = !!rest;
++        usage.track(req, res);
          const q = req.query;
          if (!bucket) {
              if (req.method === 'GET') return await listBuckets(req, res);
ddiff --git a/server/s3/usage.js b/server/s3/usage.js
nnew file mode 100644
iindex 0000000..8dcef9b
---- /dev/null
++++ b/server/s3/usage.js
@@@ -0,0 +1,193 @@
++'use strict';
++/**
++ * Per-zone usage on Media's S3 surface (ADR-031 section 8; openvibe-contracts media.usage.recorded@1, which is
++ * common.usage-recorded@1). Media is the only counter: per zone and closed UTC hour it stores one rollup per series as
++ * its own event, and Network (the project's usage) and Zone (its usage view) add them up.
++ *
++ *   track(req, res)     a request that reached a zone (req.s3.zone): counted when its answer finishes
++ *   sample()            each active zone's live bytes for this minute, added as byte_hours (bytes / 60 a minute)
++ *   flush()             the counts kept in memory are added to media_s3_usage (every 30 s, and at shutdown)
++ *   emit()              every closed hour not sent yet, or corrected since (revision), goes to the outbox as
++ *                       media.usage.recorded, in the transaction that marks it sent
++ *
++ * The series (capability, unit): stored bytes (media.object.upload, byte_hours); writes (media.object.upload and
++ * media.object.delete, requests); lists (media.object.list, requests); reads (media.object.read, requests and
++ * egress_bytes). A refused or failed request counts in `errors`, by its S3 code as media.s3.<snake_case>. Each rollup
++ * names its zone twice: resource is the zone's OVRN and dimension its id in lowercase. Nothing here names who sent a
++ * request, or a key, an object key or an address: rollups are counts, never money (Billing never reads them).
++ */
++const db = require('../db/database');
++const events = require('../events');
++
++const HOUR_MS = 3600 * 1000;
++const MAX_CODES = 20;
++const MAX_SAMPLES = 10;
++const hourOf = (ms) => Math.floor(ms / HOUR_MS) * HOUR_MS;
++const parse = (v, d) => { if (v && typeof v === 'object') return v; try { return JSON.parse(v); } catch { return d; } };
++const snake = (code) => String(code).replace(/([a-z0-9])([A-Z])/g, '$1_$2').replace(/[^A-Za-z0-9]+/g, '_').toLowerCase();
++
++let pending = new Map(); // zone|window|capability|unit → { zone, window, capability, unit, quantity, errors, codes, samples }
++
++function bucket(zone, capability, unit, now) {
++    const window = hourOf(now);
++    const k = `${zone.id}|${window}|${capability}|${unit}`;
++    let b = pending.get(k);
++    if (!b) {
++        b = { zone: { id: zone.id, project_id: zone.project_id, env: zone.env }, window, capability, unit, quantity: 0, errors: 0, codes: {}, samples: [] };
++        pending.set(k, b);
++    }
++    return b;
++}
++
++/** Add to one series of a zone now: quantity units, and an error when code is set. */
++function count(zone, capability, unit, quantity, { code = null, status = null, now = Date.now() } = {}) {
++    if (!zone || !zone.id) return;
++    const b = bucket(zone, capability, unit, now);
++    b.quantity += Math.max(0, Math.round(quantity || 0));
++    if (code) {
++        b.errors++;
++        const c = `media.s3.${snake(code)}`;
++        b.codes[c] = (b.codes[c] || 0) + 1;
++        b.samples.unshift({ at: new Date(now).toISOString(), code: c, ...(status ? { status } : {}) });
++        b.samples.length = Math.min(b.samples.length, MAX_SAMPLES);
++    }
++}
++
++/** The capability a request to a zone counts against, from its method and query (hasKey: an object, not the bucket). */
++function capabilityOf(req, hasKey) {
++    const m = req.method;
++    const q = req.query || {};
++    if (!hasKey) {
++        if (m === 'POST' && q.delete !== undefined) return 'media.object.delete';
++        if ((m === 'GET' && q.location === undefined) || m === 'HEAD') return 'media.object.list';
++        return null;
++    }
++    if (m === 'PUT' || m === 'POST') return 'media.object.upload';
++    if (m === 'DELETE') return 'media.object.delete';
++    if (m === 'GET' && q.uploadId !== undefined) return 'media.object.list';
++    if (m === 'GET' || m === 'HEAD') return 'media.object.read';
++    return null;
++}
++
++/** Count a request to a zone when its answer is sent (req.s3.zone and req.s3.hasKey are set by the dispatcher). */
++function track(req, res) {
++    res.once('finish', () => {
++        const s3 = req.s3;
++        if (!s3 || !s3.zone) return;
++        const capability = capabilityOf(req, s3.hasKey);
++        if (!capability) return;
++        const failed = res.statusCode >= 400;
++        const code = failed ? (res.locals.s3code || `http_${res.statusCode}`) : null;
++        count(s3.zone, capability, 'requests', 1, { code, status: failed ? res.statusCode : null });
++        if (capability === 'media.object.read' && req.method === 'GET' && !failed) {
++            count(s3.zone, capability, 'egress_bytes', Number(res.getHeader('content-length')) || 0);
++        }
++    });
++}
++
++/** Each active zone's live bytes, as one minute of byte_hours (bytes / 60). */
++async function sample({ now = Date.now() } = {}) {
++    const zones = require('./zones');
++    const rows = await db.all("SELECT * FROM media_zones WHERE state = 'active'");
++    for (const z of rows) {
++        const st = await zones.status(z, { now });
++        if (st.stored_bytes > 0) count(z, 'media.object.upload', 'byte_hours', st.stored_bytes / 60, { now });
++    }
++    return rows.length;
++}
++
++/** Add what is kept in memory to media_s3_usage. A window already sent gets a new revision (it is re-sent). */
++async function flush() {
++    if (!pending.size) return 0;
++    const batch = [...pending.values()];
++    pending = new Map();
++    try {
++        await db.getDb().tx(async () => {
++            for (const b of batch) {
++                const cur = await db.get(`SELECT * FROM media_s3_usage WHERE zone_id = ? AND window_start = ? AND capability = ? AND unit = ?`,
++                    [b.zone.id, b.window, b.capability, b.unit]);
++                if (!cur) {
++                    await db.run(`INSERT INTO media_s3_usage (zone_id, project_id, env, window_start, capability, unit, quantity, errors, error_codes, samples, revision, updated_at)
++                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?)`,
++                    [b.zone.id, b.zone.project_id, b.zone.env, b.window, b.capability, b.unit, b.quantity, b.errors,
++                        JSON.stringify(topCodes(b.codes)), JSON.stringify(b.samples), Date.now()]);
++                    continue;
++                }
++                const codes = parse(cur.error_codes, {});
++                for (const [c, n] of Object.entries(b.codes)) codes[c] = (codes[c] || 0) + n;
++                const samples = [...b.samples, ...parse(cur.samples, [])].slice(0, MAX_SAMPLES);
++                const sent = cur.sent_revision != null && Number(cur.sent_revision) >= Number(cur.revision);
++                await db.run(`UPDATE media_s3_usage SET quantity = quantity + ?, errors = errors + ?, error_codes = ?, samples = ?,
++                    revision = revision + ?, updated_at = ? WHERE zone_id = ? AND window_start = ? AND capability = ? AND unit = ?`,
++                [b.quantity, b.errors, JSON.stringify(topCodes(codes)), JSON.stringify(samples), sent ? 1 : 0, Date.now(),
++                    b.zone.id, b.window, b.capability, b.unit]);
++            }
++        });
++    } catch (err) {
++        // Put the counts back: the next flush tries again.
++        for (const b of batch) {
++            const t = bucket(b.zone, b.capability, b.unit, b.window);
++            t.quantity += b.quantity;
++            t.errors += b.errors;
++            for (const [c, n] of Object.entries(b.codes)) t.codes[c] = (t.codes[c] || 0) + n;
++            t.samples = [...b.samples, ...t.samples].slice(0, MAX_SAMPLES);
++        }
++        throw err;
++    }
++    return batch.length;
++}
++
++function topCodes(codes) {
++    return Object.fromEntries(Object.entries(codes).sort((a, b) => b[1] - a[1]).slice(0, MAX_CODES));
++}
++
++/** The rollup a stored row becomes (common.usage-recorded@1). */
++function rollupOf(row) {
++    const codes = parse(row.error_codes, {});
++    const samples = parse(row.samples, []);
++    const start = Number(row.window_start);
++    const out = {
++        project_id: row.project_id, env: row.env, capability: row.capability,
++        resource: `ovrn:zone:${row.project_id}:object-zone/${row.zone_id}`, dimension: String(row.zone_id).toLowerCase(),
++        unit: row.unit, window: 'hour', window_start: new Date(start).toISOString(), window_end: new Date(start + HOUR_MS).toISOString(),
++        quantity: Number(row.quantity), errors: Number(row.errors),
++    };
++    if (Object.keys(codes).length) out.error_codes = codes;
++    if (samples.length) out.samples = samples;
++    if (Number(row.revision) > 1) out.revision = Number(row.revision);
++    return out;
++}
++
++/** Store every closed hour not sent yet (or corrected since) as media.usage.recorded. */
++async function emit({ now = Date.now(), limit = 500 } = {}) {
++    const rows = await db.all(`SELECT * FROM media_s3_usage WHERE window_start < ? AND (sent_revision IS NULL OR sent_revision < revision)
++        ORDER BY window_start, zone_id, capability, unit LIMIT ?`, [hourOf(now), limit]);
++    let sent = 0;
++    for (const row of rows) {
++        const rollup = rollupOf(row);
++        const v = events.checkUsage(rollup);
++        if (!v.valid) {
++            console.error(`[S3] usage rollup ${row.zone_id} ${rollup.window_start} ${row.capability}/${row.unit} does not match media.usage.recorded@1 and stays unsent: ${v.errors}`);
++            continue;
++        }
++        await db.getDb().tx(async () => {
++            const env = await events.recordUsage(rollup);
++            await db.run(`UPDATE media_s3_usage SET sent_revision = ? WHERE zone_id = ? AND window_start = ? AND capability = ? AND unit = ?`,
++                [Number(row.revision), row.zone_id, row.window_start, row.capability, row.unit]);
++            if (env) sent++;
++        });
++    }
++    return sent;
++}
++
++/** Rows older than keepDays (the events carry them on; Zone and Network keep their own sums). */
++async function prune({ now = Date.now(), keepDays = 40 } = {}) {
++    const r = await db.run('DELETE FROM media_s3_usage WHERE window_start < ? AND sent_revision IS NOT NULL AND sent_revision >= revision',
++        [now - keepDays * 24 * HOUR_MS]);
++    return r && r.changes ? r.changes : 0;
++}
++
++function _reset() { pending = new Map(); }
++function _pending() { return pending; }
++
++module.exports = { track, count, capabilityOf, sample, flush, emit, prune, rollupOf, snake, _reset, _pending };
ddiff --git a/test/s3-usage.test.js b/test/s3-usage.test.js
nnew file mode 100644
iindex 0000000..82b4dd3
---- /dev/null
++++ b/test/s3-usage.test.js
@@@ -0,0 +1,167 @@
++'use strict';
++// ADR-031 section 8: Media counts each zone's use of the S3 surface and sends every closed hour as media.usage.recorded
++// (common.usage-recorded@1): writes, lists, reads and egress as they are answered, stored bytes as byte_hours sampled
++// each minute, refusals by code. Late counts raise a sent hour's revision and it is sent again.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const crypto = require('crypto');
++
++(async () => {
++    const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-s3u-'));
++    const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
++    process.env.VOD_PATH = dir('vods');
++    process.env.FILES_PATH = dir('files');
++    process.env.OBJECTS_PATH = dir('objects');
++    process.env.THUMBNAILS_PATH = dir('thumbnails');
++    process.env.OV_NETWORK_URL = 'https://openvibe.network';
++    process.env.MEDIA_PUBLIC_URL = 'https://media.test';
++    process.env.MEDIA_SIGNING_SECRET = 'test-signing-secret';
++    process.env.MEDIA_S3_KEY_SECRET = 'k'.repeat(40);           // a test value: 32+ characters
++    process.env.MEDIA_S3_HOSTS = '127.0.0.1';
++
++    const db = require('../server/db/database');
++    const auth = require('../server/auth');
++    const events = require('../server/events');
++    const usage = require('../server/s3/usage');
++    const { serviceAuth, zones: zoneContracts, ids, validate } = require('openvibe-contracts');
++    const express = require('express');
++    const S3 = require('@aws-sdk/client-s3');
++
++    const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });
++    auth._setNetworkPublicKeyForTests(keys.publicKey);
++    const now = Math.floor(Date.now() / 1000);
++    const svc = (sub, cap) => serviceAuth.signServiceToken({ iss: 'https://openvibe.network', sub, actor_type: 'service', aud: ['openvibe.media'], cap, iat: now, exp: now + 600, jti: `tok_${crypto.randomBytes(6).toString('hex')}` }, keys.privateKey);
++    const ZONE_TOKEN = svc('svc:zone', ['media.zone.provision']);
++    const SERVICES_TOKEN = svc('svc:services', ['media.access_key.manage']);
++    const ADMIN = `usr_${ids.ulid()}`;
++    const PRJ = `prj_${ids.ulid()}`;
++
++    const s3router = require('../server/s3/routes');
++    const admin = require('../server/s3/admin-routes');
++    const app = express();
++    app.use((req, res, next) => (s3router.isS3Host(req) ? s3router(req, res, next) : next()));
++    app.use('/internal/zones', admin.internalRouter());
++    app.use('/api/v1/projects/:project/access-keys', admin.keysRouter({ projects: { isAdmin: async (p, s) => s === ADMIN && p === PRJ } }));
++    const server = http.createServer(app);
++    await new Promise((r) => server.listen(0, '127.0.0.1', r));
++    const port = server.address().port;
++    const internal = async (method, p, { token, body } = {}) => {
++        const res = await fetch(`http://localhost:${port}${p}`, { method, headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' }, body: body ? JSON.stringify(body) : undefined });
++        return { status: res.status, body: await res.json().catch(() => null) };
++    };
++    const zoneRecord = (env, name) => {
++        const z = { id: `zon_${ids.ulid()}`, project_id: PRJ, env, name, default: name === 'default', access: name === 'default' ? 'per-object' : 'private', state: 'active',
++            quota_bytes: null, revision: 1, created_at: new Date().toISOString() };
++        return Object.assign(z, zoneContracts.identity(z));
++    };
++    const rows = async (zoneId) => db.all('SELECT * FROM media_s3_usage WHERE zone_id = ? ORDER BY capability, unit', [zoneId]);
++    const row = (list, capability, unit) => list.find((r) => r.capability === capability && r.unit === unit);
++
++    // What the outbox would store: the rollups, as recordUsage receives them inside its transaction.
++    const sent = [];
++    events.recordUsage = async (payload) => { sent.push(payload); return { event_id: `evt_${ids.ulid()}` }; };
++
++    try {
++        const assets = zoneRecord('production', 'assets');
++        const scratch = zoneRecord('sandbox', 'scratch');
++        for (const z of [zoneRecord('production', 'default'), assets, zoneRecord('sandbox', 'default'), scratch]) {
++            assert.strictEqual((await internal('PUT', `/internal/zones/${z.id}`, { token: ZONE_TOKEN, body: z })).status, 200);
++        }
++        const issue = async (env) => (await internal('POST', `/api/v1/projects/${PRJ}/access-keys`, { token: SERVICES_TOKEN, body: { on_behalf_of: ADMIN, env, name: env } })).body;
++        const k = await issue('production');
++        const ks = await issue('sandbox');
++        const client = (key) => new S3.S3Client({ endpoint: `http://127.0.0.1:${port}`, region: 'us-east-1', forcePathStyle: true, credentials: { accessKeyId: key.key.id, secretAccessKey: key.secret } });
++        const s3 = client(k);
++
++        // ── Requests count as they are answered ──
++        const body = crypto.randomBytes(3000);
++        await s3.send(new S3.PutObjectCommand({ Bucket: 'assets', Key: 'a.bin', Body: body }));
++        await s3.send(new S3.PutObjectCommand({ Bucket: 'assets', Key: 'b.txt', Body: 'hello' }));
++        const got = await s3.send(new S3.GetObjectCommand({ Bucket: 'assets', Key: 'a.bin' }));
++        assert.strictEqual(Buffer.from(await got.Body.transformToByteArray()).length, 3000);
++        await s3.send(new S3.HeadObjectCommand({ Bucket: 'assets', Key: 'b.txt' }));
++        await assert.rejects(s3.send(new S3.GetObjectCommand({ Bucket: 'assets', Key: 'missing' })), /NoSuchKey/);
++        await s3.send(new S3.ListObjectsV2Command({ Bucket: 'assets' }));
++        await s3.send(new S3.DeleteObjectCommand({ Bucket: 'assets', Key: 'b.txt' }));
++        await assert.rejects(s3.send(new S3.GetObjectCommand({ Bucket: 'nope', Key: 'x' })), /NoSuchBucket/);
++        await client(ks).send(new S3.PutObjectCommand({ Bucket: 'scratch', Key: 's.txt', Body: 'sandbox' }));
++        await new Promise((r) => setImmediate(r));
++        assert.ok(usage._pending().size > 0, 'counted in memory first');
++        await usage.flush();
++        assert.strictEqual(usage._pending().size, 0, 'and flushed');
++        let list = await rows(assets.id);
++        assert.strictEqual(Number(row(list, 'media.object.upload', 'requests').quantity), 2, 'two writes');
++        const reads = row(list, 'media.object.read', 'requests');
++        assert.deepStrictEqual([Number(reads.quantity), Number(reads.errors)], [3, 1], 'three reads (GET, HEAD, a missing key), one refused');
++        assert.deepStrictEqual(JSON.parse(reads.error_codes), { 'media.s3.no_such_key': 1 }, 'refusals by code');
++        assert.strictEqual(JSON.parse(reads.samples)[0].status, 404);
++        assert.strictEqual(Number(row(list, 'media.object.read', 'egress_bytes').quantity), 3000, 'egress: the bytes of the GET that succeeded');
++        assert.strictEqual(Number(row(list, 'media.object.list', 'requests').quantity), 1);
++        assert.strictEqual(Number(row(list, 'media.object.delete', 'requests').quantity), 1);
++        assert.ok(!(await db.all("SELECT * FROM media_s3_usage WHERE capability LIKE '%' AND zone_id NOT IN (?, ?)", [assets.id, scratch.id])).length,
++            'a request that names no zone of the key (NoSuchBucket) counts nowhere');
++        assert.strictEqual(Number(row(await rows(scratch.id), 'media.object.upload', 'requests').quantity), 1, 'the sandbox zone counts on its own');
++        console.log('✅ writes, reads, lists, deletes and egress counted per zone as they are answered; refusals by code');
++
++        // ── Stored bytes: one minute of byte_hours per sample ──
++        await usage.sample();
++        await usage.flush();
++        list = await rows(assets.id);
++        assert.strictEqual(Number(row(list, 'media.object.upload', 'byte_hours').quantity), Math.round(3000 / 60), 'live bytes / 60 a minute');
++        console.log('✅ stored bytes sampled each minute as byte_hours');
++
++        // ── Closed hours go out once, as media.usage.recorded ──
++        assert.strictEqual(await usage.emit(), 0, 'the current hour is not closed');
++        const later = Date.now() + 3600 * 1000;
++        const n = await usage.emit({ now: later });
++        assert.ok(n >= 8, `every series of both zones (${n})`);
++        for (const p of sent) {
++            const v = validate('media.usage.recorded@1', p);
++            assert.ok(v.valid, JSON.stringify(v.errors));
++            assert.ok(events.checkUsage(p).valid, 'and its resource is a zone of its project');
++        }
++        const readRollup = sent.find((p) => p.dimension === assets.id.toLowerCase() && p.capability === 'media.object.read' && p.unit === 'requests');
++        assert.strictEqual(readRollup.resource, `ovrn:zone:${PRJ}:object-zone/${assets.id}`, 'the zone named twice: its OVRN');
++        assert.deepStrictEqual([readRollup.quantity, readRollup.errors, readRollup.window, readRollup.env], [3, 1, 'hour', 'production']);
++        assert.ok(sent.some((p) => p.env === 'sandbox' && p.dimension === scratch.id.toLowerCase()), 'sandbox zones are rolled up too');
++        assert.ok(!sent.some((p) => /usr_|ovk_|a\.bin|127\.0\.0\.1/.test(JSON.stringify(p))), 'no subject, key, object key or address in a rollup');
++        const before = sent.length;
++        assert.strictEqual(await usage.emit({ now: later }), 0, 'a sent hour is not sent again');
++        assert.strictEqual(sent.length, before);
++        console.log('✅ each closed hour sent once as media.usage.recorded (valid, OVRN and dimension, sandbox included, no names)');
++
++        // ── A late count corrects the hour: revision 2 ──
++        const window = Number(readRollup.window_start ? Date.parse(readRollup.window_start) : 0);
++        usage.count({ id: assets.id, project_id: PRJ, env: 'production' }, 'media.object.read', 'requests', 1, { now: window + 1000 });
++        await usage.flush();
++        assert.strictEqual(await usage.emit({ now: later }), 1, 'the corrected hour goes out again');
++        const fixed = sent[sent.length - 1];
++        assert.deepStrictEqual([fixed.quantity, fixed.revision, fixed.capability], [4, 2, 'media.object.read'], 'with its new total and revision');
++        console.log('✅ late counts raise the revision and the hour is re-sent with its total');
++
++        // ── An invalid rollup stays unsent ──
++        await db.run(`INSERT INTO media_s3_usage (zone_id, project_id, env, window_start, capability, unit, quantity, errors, revision, updated_at)
++            VALUES (?, ?, 'production', ?, 'not a capability', 'requests', 1, 0, 1, ?)`, [assets.id, PRJ, window, Date.now()]);
++        const quiet = console.error;
++        const logged = [];
++        console.error = (...a) => logged.push(a.join(' '));
++        try { await usage.emit({ now: later }); } finally { console.error = quiet; }
++        assert.ok(logged.some((l) => /stays unsent/.test(l)), 'it is logged');
++        assert.ok((await db.get("SELECT sent_revision FROM media_s3_usage WHERE capability = 'not a capability'")).sent_revision == null, 'and not marked sent');
++        await db.run("DELETE FROM media_s3_usage WHERE capability = 'not a capability'");
++        console.log('✅ a rollup that does not match the contract stays unsent');
++
++        // ── Prune: sent rows past the retention go ──
++        assert.strictEqual(await usage.prune({ now: later + 41 * 24 * 3600 * 1000 }) > 0, true);
++        assert.strictEqual((await rows(assets.id)).length, 0);
++        console.log('✅ sent rows are pruned after 40 days');
++    } finally {
++        server.close();
++        fs.rmSync(tmp, { recursive: true, force: true });
++    }
++    console.log('s3 usage: all checks passed');
++    process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });