0.14.3: flow control for directed and room records (queue past the budget, cmb-budget / cmb-refused, sender pacing)

SECURITY.md

@@ -240,7 +240,13 @@ Every store a peer can feed has a fixed bound. Reaching one never stops the node
   one sealed frame. This node mints nothing larger, and refuses a received record over any limit
   before anything validates, verifies or encodes it; a sealed frame longer than a 720 KiB record can
   produce (983,062 base64url characters) is refused before it is opened. At most 8 new records a
-  second per session are evaluated (burst 32).
+  second per session are evaluated (burst 32; and 128 KiB of record text a second, burst 1.5 MiB),
+  and at most 32 a second checked before verification (burst 128; 1 MiB/s, burst 4 MiB). Past either
+  a record waits in the session's queue, in order (0.14.3): at most 512 records or 8 Mi characters a
+  session, 4,096 records or 32 Mi characters in all. Past the queue it is refused unchecked and its
+  sender told (`cmb-refused`, at most one notice a second per peer, at most 256 records named in one);
+  a peer that, told, keeps sending more than 512 records past its own queue before that queue drains
+  is closed and not admitted again for 60 s.
 - **Wake channels**: at most 1,024, learned only from a confirmed session's own nodeId (a peer's
   word about another node is never stored), dropped after 30 days unseen; when the table is full a
   new first-hand channel displaces the least recently seen one of the weakest source, so many

lib/frame-handler.js

@@ -7,6 +7,10 @@ const { wireNodeId, wireName } = require('./wire-identity');
 const { canonicalRecordV2_0 } = require('./core/record-canonical');
 const { moodFrameReason } = require('./core/mood-frame');
 const { ATTEST_FRAME, fromWireAttestation, fromWireCheckpoint, fromWireWitness, fromWireNodeStats } = require('./core/attestation');
+const {
+  RECORDS_PER_SECOND, RECORDS_BURST, RECORD_CHARS_PER_SECOND, RECORD_CHARS_BURST,
+  BUDGET_FRAME, REFUSED_FRAME, recordChars, refillBucket, bucketWait, budgetFrame, RecordAdmission,
+} = require('./flow-control');
 
 /**
  * @module @sym-bot/sym/frame-handler
@@ -94,23 +98,25 @@ const DIRECTED_MARK = /#(asrt-|sd:|sig:)/;
 const { PEER_INFO_MAX } = require('./core/wake');
 
 /** A record refused for its audience is said at most once per this many ms per peer and reason. */
-/** Refill a session's token bucket `slot` (a burst, then a rate a second) and return it. */
+/**
+ * A session's token bucket `slot` (a burst, then a rate a second), refilled to `now`. A clock stepped
+ * back refills nothing and takes nothing (lib/flow-control.js refillBucket).
+ */
 function refill(session, slot, perSecond, burst, now) {
   const b = session[slot] || (session[slot] = { tokens: burst, at: now });
-  b.tokens = Math.min(burst, b.tokens + ((now - b.at) * perSecond) / 1000);
-  b.at = now;
-  return b;
+  return refillBucket(b, perSecond, burst, now);
 }
-/** Take one record from a count bucket and `chars` from a size bucket, both or neither. */
-function takeBoth(session, countSlot, countRate, countBurst, sizeSlot, sizeRate, sizeBurst, chars) {
-  const now = Date.now();
-  const c = refill(session, countSlot, countRate, countBurst, now);
-  const s = refill(session, sizeSlot, sizeRate, sizeBurst, now);
-  const need = Math.max(0, Math.min(Number(chars) || 0, sizeBurst));
-  if (c.tokens < 1 || s.tokens < need) return false;
-  c.tokens -= 1;
-  s.tokens -= need;
-  return true;
+/** Milliseconds until a session's count lane holds one record and its size lane `chars` (0: now). */
+function laneWait(session, countSlot, countRate, countBurst, sizeSlot, sizeRate, sizeBurst, chars, now) {
+  return Math.max(
+    bucketWait(refill(session, countSlot, countRate, countBurst, now), countRate, countBurst, 1, now),
+    bucketWait(refill(session, sizeSlot, sizeRate, sizeBurst, now), sizeRate, sizeBurst, Math.max(0, Number(chars) || 0), now),
+  );
+}
+/** Spend one record and `chars` (at most the size burst) of a session's two lanes. */
+function laneSpend(session, countSlot, countRate, countBurst, sizeSlot, sizeRate, sizeBurst, chars, now) {
+  refill(session, countSlot, countRate, countBurst, now).tokens -= 1;
+  refill(session, sizeSlot, sizeRate, sizeBurst, now).tokens -= Math.min(Math.max(0, Number(chars) || 0), sizeBurst);
 }
 
 /** Freeze an object graph (plain data: a structured clone of a record). */
@@ -124,16 +130,14 @@ function deepFreeze(o) {
 
 /** A directed record older than this by its signed timestamp is refused as a replay (finding G). */
 /**
- * New records one session has evaluated, at most: a burst, then a rate a second (security review D).
- * A busy room's peer sends a few a second; one record costs the SVAF encoder tens of milliseconds.
- */
-const RECORDS_PER_SECOND = 8;
-const RECORDS_BURST = 32;
-/**
+ * New records one session has evaluated, at most: a burst, then a rate a second (security review D;
+ * RECORDS_PER_SECOND and RECORDS_BURST, lib/flow-control.js). A busy room's peer sends a few a second;
+ * one record costs the SVAF encoder tens of milliseconds.
+ *
  * Records one session may have checked at all, before any work on them (final re-review, Finding 3):
  * the lane spent ahead of the schema check, key recomputation and signature verification, as the
- * gossip lane is ahead of a statement's. Past it a record is dropped unchecked and unmarked. The burst
- * covers a fetch answer (at most 64 records) and the anchor replay.
+ * gossip lane is ahead of a statement's. The burst covers a fetch answer (at most 64 records) and the
+ * anchor replay.
  */
 const VERIFY_PER_SECOND = 32;
 const VERIFY_BURST = 128;
@@ -142,30 +146,19 @@ const VERIFY_BURST = 128;
  * the SVAF encoder in proportion to its text, and a record may carry 720 KiB, so counting records let
  * one peer send 32 of the largest a second and buy CPU, memory and disk without bound. A record spends
  * its size in characters (category texts and application data; recordChars) from a second bucket on
- * each lane, and passes only if both buckets hold enough. Each burst holds at least two records of the
- * largest size, so any one record can pass. The usual peer sends records of a few KiB: far inside both.
+ * each lane (RECORD_CHARS_PER_SECOND and RECORD_CHARS_BURST for the record lane). Each burst holds at
+ * least two records of the largest size, so any one record can pass. The usual peer sends records of
+ * a few KiB: far inside both.
+ *
+ * PAST THE BUDGET A RECORD WAITS, IT IS NOT DROPPED (flow control, 0.14.3; lib/flow-control.js). Every
+ * record a session brings is admitted only when both lanes hold it and nothing of that session waits
+ * ahead of it; otherwise it is queued, in order, and admitted as the lanes refill, so the budget bounds
+ * the work exactly as before and loses nothing. Only past the queue's bound is a record refused, and its
+ * sender is told (`cmb-refused`) when to try again. Until 0.14.3 a record past either lane was dropped
+ * unchecked or unevaluated while its sender reported it dispatched.
  */
 const VERIFY_CHARS_PER_SECOND = 1024 * 1024;
 const VERIFY_CHARS_BURST = 4 * 1024 * 1024;
-const RECORD_CHARS_PER_SECOND = 128 * 1024;
-const RECORD_CHARS_BURST = 1536 * 1024;
-/** What a record weighs for the size budgets: its texts and application data, in characters. */
-function recordChars(cmb) {
-  let n = 512;
-  const cats = cmb && typeof cmb === 'object' ? cmb.categories : null;
-  if (cats && typeof cats === 'object') {
-    for (const v of Object.values(cats)) {
-      if (typeof v === 'string') n += v.length;
-      else if (v && typeof v === 'object' && typeof v.text === 'string') n += v.text.length;
-    }
-  }
-  const md = cmb && typeof cmb === 'object' ? cmb.metadata : null;
-  if (md && typeof md === 'object') {
-    if (md.application && typeof md.application === 'object' && typeof md.application.data === 'string') n += md.application.data.length;
-    if (typeof md.createdBy === 'string') n += md.createdBy.length;
-  }
-  return n;
-}
 /** cmb-fetch serving bounds per session (security review D, fetch-amp). */
 const FETCH_SERVE_PER_SECOND = 2;
 const FETCH_SERVE_BURST = 8;
@@ -283,6 +276,8 @@ class FrameHandler {
     this._ledgerLines = 0;             // lines the file holds
     this._ledgerEvicted = 0;
     this._loadLedger();
+    // FLOW CONTROL (0.14.3): records past a session's budget wait here, in order, for it to refill.
+    this._admission = this._makeAdmission();
   }
 
   /** @private Read the delivery ledger (`mark\texpiresAt` per line; a 0.14.2 line is a bare mark). */
@@ -450,16 +445,60 @@ class FrameHandler {
    * @private
    */
   /**
-   * @private Take one of `session`'s pre-verification tokens and `chars` of its pre-verification size
-   * budget (R2-3), or neither. @returns {boolean}
+   * @private Milliseconds until both of `session`'s lanes hold a record of `chars` (0: now): the
+   * pre-verification lane and the record lane, each by count and by size (R2-3). A record is admitted
+   * only when this is 0 (see _handleMemoryShare), so neither lane is ever overdrawn.
    */
-  _verifyBudget(session, chars = 0) {
-    return takeBoth(session, '_verifyBucket', VERIFY_PER_SECOND, VERIFY_BURST, '_verifyCharsBucket', VERIFY_CHARS_PER_SECOND, VERIFY_CHARS_BURST, chars);
+  _admitWait(session, chars = 0, now = Date.now()) {
+    return Math.max(
+      laneWait(session, '_verifyBucket', VERIFY_PER_SECOND, VERIFY_BURST, '_verifyCharsBucket', VERIFY_CHARS_PER_SECOND, VERIFY_CHARS_BURST, chars, now),
+      laneWait(session, '_recordBucket', RECORDS_PER_SECOND, RECORDS_BURST, '_recordCharsBucket', RECORD_CHARS_PER_SECOND, RECORD_CHARS_BURST, chars, now),
+    );
+  }
+
+  /** @private Spend `session`'s pre-verification lane for a record of `chars`: before any work on it. */
+  _spendVerify(session, chars = 0, now = Date.now()) {
+    laneSpend(session, '_verifyBucket', VERIFY_PER_SECOND, VERIFY_BURST, '_verifyCharsBucket', VERIFY_CHARS_PER_SECOND, VERIFY_CHARS_BURST, chars, now);
+  }
+
+  /** @private Spend `session`'s record lane for a new record of `chars`: before SVAF encodes anything. */
+  _spendRecord(session, chars = 0, now = Date.now()) {
+    laneSpend(session, '_recordBucket', RECORDS_PER_SECOND, RECORDS_BURST, '_recordCharsBucket', RECORD_CHARS_PER_SECOND, RECORD_CHARS_BURST, chars, now);
+  }
+
+  /** The record budget this node advertises to each admitted session (`cmb-budget`). */
+  flowBudget() {
+    const L = this._admission.limits;
+    return budgetFrame({ recordsPerSecond: RECORDS_PER_SECOND, recordsBurst: RECORDS_BURST, charsPerSecond: RECORD_CHARS_PER_SECOND, charsBurst: RECORD_CHARS_BURST, queueRecords: L.queueRecords, queueChars: L.queueChars });
   }
 
-  /** @private Take one of `session`'s record tokens and `chars` of its record size budget (R2-3), or neither. @returns {boolean} */
-  _recordBudget(session, chars = 0) {
-    return takeBoth(session, '_recordBucket', RECORDS_PER_SECOND, RECORDS_BURST, '_recordCharsBucket', RECORD_CHARS_PER_SECOND, RECORD_CHARS_BURST, chars);
+  /** @private The admission queue in front of the record budget (lib/flow-control.js). */
+  _makeAdmission() {
+    const node = this._node;
+    return new RecordAdmission({
+      wait: (session, chars, now) => this._admitWait(session, chars, now),
+      run: (e) => {
+        this._spendVerify(e.session, e.chars);
+        return this._processRecord(e.peerId, e.peerName, e.msg, e.session);
+      },
+      // The door, consulted per frame (MMP §5.8.1), and again for a record that waited: a peer no
+      // longer admitted (evicted, penalised) has what it queued dropped, counted.
+      admitted: (e) => {
+        if (typeof node._penalised === 'function' && node._penalised(e.peerId)) return false;
+        if (typeof node._floodPenalised === 'function' && node._floodPenalised(e.peerId)) return false;
+        return typeof node._roomDoor !== 'function' || node._roomDoor(e.peerId).pass === true;
+      },
+      sessionFor: (peerId) => {
+        const s = node._peers && node._peers.get(peerId) && node._peers.get(peerId).transport;
+        return s && s.confirmed && !s.closed && !s.legacy && typeof s.trySend === 'function' ? s : null;
+      },
+      cut: (session) => { if (typeof node._penaliseFlood === 'function') node._penaliseFlood(session); },
+      fail: (e, err) => {
+        if (typeof node._refuseFrame === 'function') node._refuseFrame(e.peerId, e.peerName, e.msg, e.session && e.session.kind === 'relay' ? 'relay' : 'lan', err);
+      },
+      metric: (m) => { try { node.emit('metric', m); } catch { /* a sink's failure is not the queue's */ } },
+      say: (key, line) => this._sayOnceAMinute(this._audienceSaid, key, (more) => `${line}${more}`),
+    });
   }
 
   _markCmbSurfaced(key, msg) {
@@ -621,6 +660,15 @@ class FrameHandler {
         if (typeof this._node._noteSessionRefusal === 'function') this._node._noteSessionRefusal(session, msg.type, 'legacy-attest-frame');
         return undefined;
 
+      // Flow control (0.14.3, lib/flow-control.js): the peer's record budget, and the records of this
+      // node's it refused past its queue. A 0.14.2 node ignores both (§7.3, below).
+      case BUDGET_FRAME:
+        if (typeof this._node._onPeerBudget === 'function') this._node._onPeerBudget(session, msg);
+        return undefined;
+      case REFUSED_FRAME:
+        if (typeof this._node._onPeerRefused === 'function') this._node._onPeerRefused(session, msg);
+        return undefined;
+
       case 'cmb-fetch':
         return this._handleCmbFetch(peerId, peerName, msg, session);
 
@@ -861,15 +909,10 @@ class FrameHandler {
     if (!md.sig) return refuse('unsigned');
     if (typeof md.createdByNodeId !== 'string' || !md.createdByNodeId) return refuse('no-author-node-id');
     // The work a record costs is charged before any of it is done (final re-review, Finding 3): the
-    // session's verification lane, then the (relayer, author) failure window. Dropped unchecked and
-    // unmarked either way, so a genuine copy from elsewhere is still taken.
+    // session's verification lane was spent when the record was admitted (_handleMemoryShare), then
+    // the (relayer, author) failure window, dropped unchecked and unmarked, so a genuine copy from
+    // elsewhere is still taken.
     const session = this._sessionOf.get(msg);
-    if (session && !this._verifyBudget(session, recordChars(cmb))) {
-      msg._cmbVerified = false;
-      this._node.emit('metric', { type: 'cmb-over-verify-budget', from: peerName, key: k });
-      this._sayOnceAMinute(this._audienceSaid, `${peerId}|verify-budget`, (more) => `CMBs from ${peerName} over its verification budget (${VERIFY_PER_SECOND}/s) — dropped unchecked${more}`);
-      return true;
-    }
     if (typeof this._node._relayMuted === 'function' && this._node._relayMuted(peerId, md.createdByNodeId)) {
       msg._cmbVerified = false;
       this._node.emit('metric', { type: 'relayed-signer-muted', kind: 'record', peer: peerId, author: md.createdByNodeId.slice(0, 128) });
@@ -1209,6 +1252,15 @@ class FrameHandler {
       // own node authored is marked one.
       if (msg.cmb && msg.cmb.metadata && msg.cmb.metadata.createdByNodeId === session.nodeId) msg._anchor = true;
     }
+    // FLOW CONTROL (0.14.3): a session's records are admitted in order, each once both of its lanes
+    // hold it; one that arrives past the budget, or behind one that waits, is queued until they do,
+    // and only past the queue's bound refused, its sender told (lib/flow-control.js).
+    if (session) return this._admission.offer(session, peerId, peerName, msg, recordChars(msg && msg.cmb));
+    return this._processRecord(peerId, peerName, msg, session);
+  }
+
+  /** @private One record, admitted: verified, evaluated and surfaced (or refused on its merits). */
+  _processRecord(peerId, peerName, msg, session) {
     // The session rides beside the frame, never on it: entries are built from the frame and kept.
     if (session) this._sessionOf.set(msg, session);
     let pending;
@@ -1442,13 +1494,9 @@ class FrameHandler {
 
     // A session's new records are evaluated at most RECORDS_PER_SECOND (burst RECORDS_BURST): charged
     // here, after the signature and the de-duplication and before SVAF encodes anything, which is
-    // most of what a record costs (security review D, record-flood). Past it the record is dropped
-    // unmarked, so a later copy is taken.
-    if (session && !this._recordBudget(session, recordChars(msg.cmb))) {
-      this._node.emit('metric', { type: 'cmb-over-budget', from: peerName, key: cognitionKey });
-      this._sayOnceAMinute(this._audienceSaid, `${peerId}|record-budget`, (more) => `CMBs from ${peerName} over its budget (${RECORDS_PER_SECOND}/s) — dropped unevaluated${more}`);
-      return;
-    }
+    // most of what a record costs (security review D, record-flood). The record was admitted only once
+    // the lane held it (_handleMemoryShare), and nothing since spent it, so this never overdraws.
+    if (session) this._spendRecord(session, recordChars(msg.cmb));
 
     // A record signed to this node by its author, verified, fresh, new and within the session's
     // budget, is accepted for delivery from here on whatever SVAF decides about memory (it surfaces,