[Security] DistributedData pending-writes map can accumulate stale timeouts

[pathosDev]

Severity / Size

• Severity: MEDIUM (defensive / load-shape concern — see "Caveat")
• Size: S
• Threat model: closed-group cluster under high contention or partition. A caller (local actor or remote service) issues many concurrent updateAsync(..., { consistency: 'majority' }) calls; quorum acks don't arrive (peer slow, partitioned, or overloaded); timeouts haven't yet fired.

Affected files

• src/crdt/DistributedData.ts:543-546 — pendingWrites: Map<string, PendingWrite> and pendingReads: Map<string, PendingRead> — no cap.
• src/crdt/DistributedData.ts:645-667 — handleUpdate arms a timeout, sets pendingWrites[pendingId]; the timer fires pendingWrites.delete(pendingId).
• src/crdt/DistributedData.ts:598-612 — postStop cleans up all pending requests.
• src/crdt/DistributedData.ts:394-415 — updateAsync (public API) is the source of pending writes; no rate limit.

Caveat — audit framing vs reality

The audit calls this "pending-writes map can accumulate stale timeouts". Reading the code:

1. Each pending write has a per-entry timer, set via system.scheduler.scheduleOnceFn(timeoutMs, () => …).
2. When the timer fires, the entry is removed from pendingWrites (DistributedData.ts:648) and the caller's promise is rejected.
3. On actor stop, postStop walks all pending requests, cancels their timers, and rejects the promises (DistributedData.ts:601-611).

So timer-cleanup is correct. The map doesn't "accumulate stale entries" in the sense of "entries that never get cleaned up". They get cleaned up when their respective timeouts fire.

The real concern is transient unboundedness: between the moment of issuing N concurrent updateAsync calls and the moment their timers fire (typically 5 seconds with default timeout), the map can hold N entries. Under bursty load:

• 100K updateAsync calls in 100ms → 100K pending entries for the next 4.9 seconds.
• Each entry holds: pendingId, Set of acks (one entry per peer that acked), timer reference, resolve/reject closures (which retain caller-side state via captured variables).
• Memory peak: ~500 bytes × 100K = 50 MB held transiently.

That's not "leak" but "DoS via transient memory pressure". Combined with caller-side promise pressure (every pending promise has its own microtask state), the actual peak per node can be larger.

The fix is therefore a caller-side rate limit / pending-count cap, not a "fix the leak that doesn't exist". This issue re-scopes accordingly.

Background

DistributedData.updateAsync and getAsync are the public quorum-write/read APIs. Both go through the actor's mailbox; both register a pending entry; both arm a timer. There's no:

• Cap on concurrent pending requests.
• Backpressure on the caller (updateAsync returns a Promise<void> immediately; doesn't await available slot).
• Per-key concurrency limit (issuing 1000 concurrent updates to the same key is allowed, each gets its own pendingId).

Concrete failure modes:

1. Memory peak under burst load: 100K concurrent calls = 50 MB+ transient.
2. Mailbox saturation: every pending write turns into an outbound transport.send per peer; at 100K × 5 peers = 500K outbound envelopes queued for the cluster transport. Transport backs up.
3. Per-peer transport queue depth explodes; cluster transport's backpressure (frame-cap, queue depth) kicks in; legitimate cluster traffic (gossip, heartbeats) gets delayed.
4. Cascade: gossip delayed → failure detector marks peers unreachable → downing kicks in → cluster splits.

Exploit walkthrough

Step 1 — App uses DistributedData for high-frequency counter updates:

// Handler called per HTTP request:
post(async (req) => {
  await dd.updateAsync<GCounter>('global-hits', GCounter.empty, (c) => c.increment(replicaId, 1), {
    consistency: 'majority',
  });
  return completeJson(200, { ok: true });
});

Step 2 — Partition + load: cluster partition reaches half the peers; load spikes to 10K requests/sec. Each request → one updateAsync → one pending entry. Timeouts are 5s default → at peak, 50K pending entries.

Step 3 — Memory and queue pressure: each pending entry holds:

• acks: Set<string> (one per peer).
• timer: Cancellable (closure reference to scheduler).
• resolve / reject (closures from updateAsync's Promise).
• pendingId string.

Per-entry cost ~500 bytes; total 25 MB at peak. Plus 50K × 5 peers = 250K outbound envelopes queued for the transport.

Step 4 — Cascade: cluster transport's per-peer queue saturates; gossip delayed by 500ms+; failure detector marks the (already-partitioned) half of the cluster as unreachable faster than the partition itself would have; downing provider acts; the unaffected half marks the partitioned half down; on heal, re-incarnation race kicks in (covered by other tickets).

Realistic worst case: load spike + partition → transient cluster split → recovery requires manual intervention. Memory pressure alone isn't fatal (transient) but the secondary effects compound.

A subtler exploit: an attacker that can issue updateAsync (via an HTTP endpoint that uses it) issues 100K calls with extremely long timeoutMs (e.g. 30s). Pending entries persist for 30s. Same memory pressure, sustained.

How the 8 already-landed security fixes inform this

• Wire-frame DoS cap — bound a queue with a configurable default. Same shape: cap concurrent pending requests.
• Hello-handshake hijack defence — validate input at the boundary. Same shape: validate timeoutMs (cap at sane upper bound, e.g. 60s).
• Snapshot seq integrity — strict bound. Same shape.

Fix design

Track 1 — Per-mediator pending-count cap (primary). Default: 10000 (covers 1K-req/sec × 5s timeout × 2x headroom).

export interface DistributedDataSettings {
  // ... existing fields ...
  /** Cap on concurrent pending quorum requests.  New calls past this cap reject immediately.  Default: 10000. */
  readonly maxPendingRequests?: number;
  /** Upper bound on the user-supplied timeoutMs.  Default: 60_000 (60s). */
  readonly maxTimeoutMs?: number;
}

private handleUpdate(msg: UpdateMsg): void {
  // ... existing apply logic ...
  if (!msg.quorum) return;

  // NEW — validate timeout + cap pending count
  if (msg.quorum.timeoutMs > this.maxTimeoutMs) {
    msg.quorum.reject(new Error(
      `DistributedData: timeoutMs ${msg.quorum.timeoutMs} exceeds cap ${this.maxTimeoutMs}`
    ));
    return;
  }
  const totalPending = this.pendingWrites.size + this.pendingReads.size;
  if (totalPending >= this.maxPendingRequests) {
    this.metrics.counter('ddata_request_rejected_total', { reason: 'cap-reached' }).inc();
    msg.quorum.reject(new Error(
      `DistributedData: concurrent-request cap ${this.maxPendingRequests} reached; try again later`
    ));
    return;
  }
  // ... existing quorum-write logic ...
}

Same shape in handleRead.

Track 2 — Caller-side backpressure helper. updateAsync returns a Promise; callers can in principle bound concurrency themselves via p-limit or similar. Document the pattern + provide a built-in helper:

export class BoundedDistributedData {
  constructor(
    private readonly inner: DistributedDataHandle,
    private readonly maxConcurrent: number,
  ) {}

  private inFlight = 0;
  private readonly waiters: Array<() => void> = [];

  async updateAsync<C extends Crdt<C>>(...args: Parameters<DistributedDataHandle['updateAsync']>): Promise<void> {
    while (this.inFlight >= this.maxConcurrent) {
      await new Promise<void>((r) => this.waiters.push(r));
    }
    this.inFlight++;
    try { return await this.inner.updateAsync(...args); }
    finally {
      this.inFlight--;
      this.waiters.shift()?.();
    }
  }
}

Track 3 — Per-key concurrency cap (optional, defensive). A single key shouldn't have >100 concurrent pending writes — symptom of either runaway code or hot-key contention. Track Map<key, number> of in-flight per key; reject above cap.

Track 4 — Metric. ddata_pending_requests{kind=write|read} Gauge + ddata_request_rejected_total{reason} Counter.

Track 5 — Documentation. README "Known security caveats":

- DistributedData: concurrent pending quorum requests capped at
  10000 by default; user-supplied timeoutMs capped at 60s.
  Apps issuing high-frequency updates should batch or use a
  BoundedDistributedData wrapper for caller-side backpressure.

API surface

new DistributedDataSettings({
  maxPendingRequests: 50_000,    // override
  maxTimeoutMs: 120_000,         // 2 minutes
});

const bounded = new BoundedDistributedData(dd, 100);
await bounded.updateAsync(...);

// Internal: a `ddata_request_rejected_total{reason='cap-reached'}` counter fires when cap hit.

Backward compatibility

Behaviour change for apps that legitimately have >10K concurrent pending requests (rare). Opt-out via Infinity. Document.

updateAsync may now reject with a "cap reached" error — that's a new error class callers should handle (otherwise it surfaces as an unhandled-rejection).

Test plan

1. Cap-respect — issue 10K concurrent updateAsync calls (with peers slow to ack); 10001st rejects immediately with cap-reached error.
2. Cap-disabled — maxPendingRequests: Infinity allows arbitrary concurrent.
3. Timeout-cap — updateAsync(key, factory, fn, { timeoutMs: 120_000 }) with maxTimeoutMs: 60_000 → immediate rejection.
4. Per-key cap — issue 101 concurrent updates to same key → 101st rejected.
5. Metric correctness — Gauge tracks live pending count; Counter tracks total rejections.
6. BoundedDistributedData wrapper — verify that 1000 issued calls with limit 100 produce at most 100 concurrent in-flight.
7. Stop while pending — postStop still rejects all pending (regression check on existing cleanup).
8. Timer fires correctly — pending entry removed on timeout regardless of cap.

Acceptance criteria

• DistributedDataSettings.maxPendingRequests (default 10000) honored.
• DistributedDataSettings.maxTimeoutMs (default 60000) honored.
• Per-key cap implemented (default 100).
• Metrics emitted.
• BoundedDistributedData wrapper exported.
• CHANGELOG + README "Known security caveats" updated.
• Test suite covers all 8 cases above.