From f4fce35a23a82bfbf02e1cfa4ee4c2346bb66fa9 Mon Sep 17 00:00:00 2001 From: gloskull Date: Fri, 17 Jul 2026 18:22:47 +0100 Subject: [PATCH] Add per-tenant token bucket rate limiting --- docs/API_RATE_LIMITING.md | 56 ++++++++ meter-simulator/src/config.js | 7 + meter-simulator/src/contract-interface.js | 4 + meter-simulator/src/mqtt-publisher.js | 8 ++ meter-simulator/src/rate-limiter.js | 155 +++++++++++++++++++++ meter-simulator/tests/rate-limiter.test.js | 81 +++++++++++ 6 files changed, 311 insertions(+) create mode 100644 docs/API_RATE_LIMITING.md create mode 100644 meter-simulator/src/rate-limiter.js create mode 100644 meter-simulator/tests/rate-limiter.test.js diff --git a/docs/API_RATE_LIMITING.md b/docs/API_RATE_LIMITING.md new file mode 100644 index 0000000..56b945c --- /dev/null +++ b/docs/API_RATE_LIMITING.md @@ -0,0 +1,56 @@ +# API Rate Limiting with Per-Tenant Token Buckets + +## Architecture + +The simulator enforces a local per-tenant token bucket before critical API paths call the contract simulator or MQTT broker. Tenants are keyed as `meter:` for usage submissions and as narrower operational tenants for heartbeats and status updates. Each tenant receives an independent bucket, so a burst from one meter cannot starve another meter. + +The implementation is dependency-free and synchronous in the hot path. A limiter check performs a map lookup, timestamp arithmetic, and a token decrement, keeping the expected P99 overhead well below the 100ms target for local simulator paths. + +## Defaults and Configuration + +Defaults are defined in `meter-simulator/src/config.js` and can be tuned with environment variables: + +| Environment variable | Default | Meaning | +| --- | ---: | --- | +| `RATE_LIMIT_CAPACITY` | `60` | Maximum burst tokens per tenant. | +| `RATE_LIMIT_REFILL_PER_SECOND` | `1` | Tokens restored per second per tenant. | +| `RATE_LIMIT_IDLE_TTL_MS` | `600000` | Idle bucket eviction window. | + +## Enforcement Points + +- Direct usage submissions call `assertAllowed("meter:")` before validation and contract simulation. +- ZK usage submissions use the same tenant key so privacy and non-privacy submissions share quota. +- MQTT usage publishing uses the same quota, while heartbeat and status topics use weighted operational buckets. + +## Monitoring and Alerting + +`PerTenantRateLimiter.snapshotMetrics()` exposes counters for allowed requests, blocked requests, bucket creation, idle eviction, active buckets, capacity, and refill rate. Production adapters should export these as Prometheus counters/gauges: + +- `rate_limiter_allowed_total` +- `rate_limiter_blocked_total` +- `rate_limiter_active_buckets` +- `rate_limiter_evicted_buckets_total` + +Recommended alerts: + +1. **Sustained throttling:** blocked / (allowed + blocked) > 5% for 10 minutes. +2. **Tenant abuse:** any single tenant blocked more than 100 times in 5 minutes. +3. **Cardinality spike:** active buckets > expected meter count by 20%. +4. **Limiter disabled/misconfigured:** capacity or refill rate resolves to zero or missing. + +## Deployment Plan + +Use blue-green deployment with a canary phase: + +1. Deploy the limiter disabled in shadow mode if the adapter supports it, exporting decisions only. +2. Enable the limiter for 5% of tenants and compare request success rate, P99 latency, and blocked-rate dashboards. +3. Increase canary to 25%, then 50%, then 100% after at least one refill window with no alert regressions. +4. Keep the previous deployment warm for rollback. Roll back if P99 latency exceeds 100ms or legitimate traffic is blocked. + +## Runbook + +1. Check `rate_limiter_blocked_total` and identify the top tenants by blocked decisions. +2. Confirm tenant traffic shape against expected meter reporting intervals. +3. If legitimate traffic is throttled, temporarily raise `RATE_LIMIT_CAPACITY` for bursts or `RATE_LIMIT_REFILL_PER_SECOND` for sustained reporting. +4. If traffic is abusive, keep quotas in place and inspect tenant credentials and MQTT topics. +5. After mitigation, verify P99 latency, blocked ratio, and active bucket cardinality return to baseline. diff --git a/meter-simulator/src/config.js b/meter-simulator/src/config.js index c638789..3ad9ca5 100644 --- a/meter-simulator/src/config.js +++ b/meter-simulator/src/config.js @@ -21,6 +21,13 @@ const config = { qos: parseInt(process.env.MQTT_QOS) || 1 }, + // Per-tenant API rate limiting defaults + rateLimit: { + capacity: parseInt(process.env.RATE_LIMIT_CAPACITY) || 60, + refillRatePerSecond: parseFloat(process.env.RATE_LIMIT_REFILL_PER_SECOND) || 1, + idleTtlMs: parseInt(process.env.RATE_LIMIT_IDLE_TTL_MS) || 10 * 60 * 1000 + }, + // Simulation defaults simulation: { defaultInterval: 30, // seconds diff --git a/meter-simulator/src/contract-interface.js b/meter-simulator/src/contract-interface.js index d0cba3b..ee49750 100644 --- a/meter-simulator/src/contract-interface.js +++ b/meter-simulator/src/contract-interface.js @@ -2,6 +2,7 @@ const { Server, Networks, TransactionBuilder, Operation, Asset, Keypair } = requ const axios = require('axios'); const crypto = require('crypto'); const config = require('./config'); +const { PerTenantRateLimiter } = require('./rate-limiter'); class ContractInterface { constructor(contractConfig) { @@ -10,6 +11,7 @@ class ContractInterface { this.horizon = new Server(contractConfig.horizonUrl); this.contractId = contractConfig.contractId; this.friendbotUrl = contractConfig.friendbotUrl; + this.rateLimiter = new PerTenantRateLimiter(config.rateLimit); } /** @@ -40,6 +42,7 @@ class ContractInterface { async submitUsageData(signedUsageData) { try { console.log('📤 Submitting usage data to contract...'); + this.rateLimiter.assertAllowed(`meter:${signedUsageData.meter_id}`); // Validate the data before submission this._validateUsageData(signedUsageData); @@ -104,6 +107,7 @@ class ContractInterface { async submitZkUsageData(zkUsageData) { try { console.log('📤 Submitting ZK privacy usage report to contract...'); + this.rateLimiter.assertAllowed(`meter:${zkUsageData.meter_id}`); // In a real implementation, this would: // 1. Create a Soroban transaction diff --git a/meter-simulator/src/mqtt-publisher.js b/meter-simulator/src/mqtt-publisher.js index 4dbfa83..905cb47 100644 --- a/meter-simulator/src/mqtt-publisher.js +++ b/meter-simulator/src/mqtt-publisher.js @@ -1,12 +1,14 @@ const mqtt = require('mqtt'); const chalk = require('chalk'); const config = require('./config'); +const { PerTenantRateLimiter } = require('./rate-limiter'); class MQTTPublisher { constructor(mqttConfig) { this.config = mqttConfig; this.client = null; this.connected = false; + this.rateLimiter = new PerTenantRateLimiter(config.rateLimit); } /** @@ -79,6 +81,8 @@ class MQTTPublisher { throw new Error('Not connected to MQTT broker'); } + this.rateLimiter.assertAllowed(`meter:${usageData.meter_id}`); + const topic = this.config.topic.replace('+', usageData.meter_id.toString()); const payload = JSON.stringify({ meter_id: usageData.meter_id, @@ -119,6 +123,8 @@ class MQTTPublisher { throw new Error('Not connected to MQTT broker'); } + this.rateLimiter.assertAllowed(`meter:${meterId}:heartbeat`, 0.2); + const topic = `meters/${meterId}/heartbeat`; const payload = JSON.stringify({ meter_id: meterId, @@ -174,6 +180,8 @@ class MQTTPublisher { throw new Error('Not connected to MQTT broker'); } + this.rateLimiter.assertAllowed(`meter:${meterId}:status`, 0.5); + const topic = `meters/${meterId}/status`; const payload = JSON.stringify({ meter_id: meterId, diff --git a/meter-simulator/src/rate-limiter.js b/meter-simulator/src/rate-limiter.js new file mode 100644 index 0000000..61271d5 --- /dev/null +++ b/meter-simulator/src/rate-limiter.js @@ -0,0 +1,155 @@ +/** + * Per-tenant token bucket rate limiter for simulator API paths. + * + * The limiter is intentionally dependency-free so it can be used by direct + * contract submissions, MQTT publishing, CLI commands, and tests without + * introducing a network call on the critical path. + */ +class TenantTokenBucket { + constructor({ capacity, refillRatePerSecond, now = () => Date.now() }) { + if (!Number.isFinite(capacity) || capacity <= 0) { + throw new Error('capacity must be a positive number'); + } + if (!Number.isFinite(refillRatePerSecond) || refillRatePerSecond <= 0) { + throw new Error('refillRatePerSecond must be a positive number'); + } + + this.capacity = capacity; + this.refillRatePerSecond = refillRatePerSecond; + this.now = now; + this.tokens = capacity; + this.updatedAt = now(); + } + + refill() { + const currentTime = this.now(); + const elapsedSeconds = Math.max(0, currentTime - this.updatedAt) / 1000; + + if (elapsedSeconds > 0) { + this.tokens = Math.min( + this.capacity, + this.tokens + elapsedSeconds * this.refillRatePerSecond + ); + this.updatedAt = currentTime; + } + } + + tryRemove(tokens = 1) { + if (!Number.isFinite(tokens) || tokens <= 0) { + throw new Error('tokens must be a positive number'); + } + + this.refill(); + + if (this.tokens >= tokens) { + this.tokens -= tokens; + return { + allowed: true, + remaining: Math.floor(this.tokens), + retryAfterMs: 0, + resetAt: this.updatedAt + }; + } + + const deficit = tokens - this.tokens; + const retryAfterMs = Math.ceil((deficit / this.refillRatePerSecond) * 1000); + + return { + allowed: false, + remaining: Math.floor(this.tokens), + retryAfterMs, + resetAt: this.updatedAt + retryAfterMs + }; + } +} + +class PerTenantRateLimiter { + constructor(options = {}) { + this.capacity = options.capacity || 60; + this.refillRatePerSecond = options.refillRatePerSecond || 1; + this.idleTtlMs = options.idleTtlMs || 10 * 60 * 1000; + this.now = options.now || (() => Date.now()); + this.buckets = new Map(); + this.metrics = { + allowed: 0, + blocked: 0, + createdBuckets: 0, + evictedBuckets: 0 + }; + } + + check(tenantId, tokens = 1) { + const bucket = this.getBucket(tenantId); + const decision = bucket.tryRemove(tokens); + + if (decision.allowed) { + this.metrics.allowed += 1; + } else { + this.metrics.blocked += 1; + } + + return { + ...decision, + tenantId, + limit: this.capacity + }; + } + + assertAllowed(tenantId, tokens = 1) { + const decision = this.check(tenantId, tokens); + if (!decision.allowed) { + const error = new Error(`Rate limit exceeded for tenant ${tenantId}`); + error.code = 'RATE_LIMIT_EXCEEDED'; + error.retryAfterMs = decision.retryAfterMs; + error.decision = decision; + throw error; + } + return decision; + } + + getBucket(tenantId) { + if (!tenantId) { + throw new Error('tenantId is required'); + } + + this.evictIdleBuckets(); + + if (!this.buckets.has(tenantId)) { + this.buckets.set( + tenantId, + new TenantTokenBucket({ + capacity: this.capacity, + refillRatePerSecond: this.refillRatePerSecond, + now: this.now + }) + ); + this.metrics.createdBuckets += 1; + } + + return this.buckets.get(tenantId); + } + + evictIdleBuckets() { + const cutoff = this.now() - this.idleTtlMs; + for (const [tenantId, bucket] of this.buckets.entries()) { + if (bucket.updatedAt < cutoff) { + this.buckets.delete(tenantId); + this.metrics.evictedBuckets += 1; + } + } + } + + snapshotMetrics() { + return { + ...this.metrics, + activeBuckets: this.buckets.size, + capacity: this.capacity, + refillRatePerSecond: this.refillRatePerSecond + }; + } +} + +module.exports = { + TenantTokenBucket, + PerTenantRateLimiter +}; diff --git a/meter-simulator/tests/rate-limiter.test.js b/meter-simulator/tests/rate-limiter.test.js new file mode 100644 index 0000000..758f443 --- /dev/null +++ b/meter-simulator/tests/rate-limiter.test.js @@ -0,0 +1,81 @@ +const { PerTenantRateLimiter, TenantTokenBucket } = require('../src/rate-limiter'); + +describe('TenantTokenBucket', () => { + test('allows requests while tokens are available', () => { + let now = 0; + const bucket = new TenantTokenBucket({ + capacity: 2, + refillRatePerSecond: 1, + now: () => now + }); + + expect(bucket.tryRemove()).toMatchObject({ allowed: true, remaining: 1 }); + expect(bucket.tryRemove()).toMatchObject({ allowed: true, remaining: 0 }); + expect(bucket.tryRemove()).toMatchObject({ allowed: false, retryAfterMs: 1000 }); + }); + + test('refills tokens over time without exceeding capacity', () => { + let now = 0; + const bucket = new TenantTokenBucket({ + capacity: 5, + refillRatePerSecond: 2, + now: () => now + }); + + expect(bucket.tryRemove(5).allowed).toBe(true); + now = 1500; + + expect(bucket.tryRemove(3)).toMatchObject({ allowed: true, remaining: 0 }); + now = 10000; + bucket.refill(); + + expect(bucket.tokens).toBe(5); + }); +}); + +describe('PerTenantRateLimiter', () => { + test('isolates token buckets per tenant', () => { + const limiter = new PerTenantRateLimiter({ capacity: 1, refillRatePerSecond: 1 }); + + expect(limiter.check('tenant-a').allowed).toBe(true); + expect(limiter.check('tenant-a').allowed).toBe(false); + expect(limiter.check('tenant-b').allowed).toBe(true); + }); + + test('throws structured errors for blocked tenants', () => { + const limiter = new PerTenantRateLimiter({ capacity: 1, refillRatePerSecond: 1 }); + limiter.assertAllowed('tenant-a'); + + expect(() => limiter.assertAllowed('tenant-a')).toThrow('Rate limit exceeded for tenant tenant-a'); + try { + limiter.assertAllowed('tenant-a'); + } catch (error) { + expect(error.code).toBe('RATE_LIMIT_EXCEEDED'); + expect(error.retryAfterMs).toBeGreaterThan(0); + expect(error.decision).toMatchObject({ tenantId: 'tenant-a', limit: 1 }); + } + }); + + test('reports metrics and evicts idle buckets', () => { + let now = 0; + const limiter = new PerTenantRateLimiter({ + capacity: 1, + refillRatePerSecond: 1, + idleTtlMs: 1000, + now: () => now + }); + + limiter.check('tenant-a'); + limiter.check('tenant-a'); + now = 2001; + limiter.check('tenant-b'); + + expect(limiter.snapshotMetrics()).toMatchObject({ + allowed: 2, + blocked: 1, + createdBuckets: 2, + evictedBuckets: 1, + activeBuckets: 1 + }); + }); +});