Api rate limiting with per tenant

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:<meter_id>` 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:<meter_id>")` 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.

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

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,

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
+};

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
+    });
+  });
+});