API 速率限制与配额政策

API 速率限制和配额策略

概述

Soroban Registry API 实现了速率限制，以确保公平使用、防止滥用，并为所有用户维持服务质量。速率限制使用基于身份的配额和滑动窗口算法。

速率限制层级

默认速率限制（每分钟）

适用于 GET /api/contracts 的分页大小感知限制

GET /api/contracts 除了标准读层级外，还使用分页大小感知速率限制。基础读限制假设 page size 为 50。请求更大的页面会按比例降低允许的请求速率。

使用默认 100 requests/min 读层级的示例：

该端点的分页校验也会强制执行：

• limit 必须在 1 和 1000 之间（默认 50）
• offset 必须为非负数
• 无效值返回 400 Bad Request

端点特定限制

算法

API 使用滑动窗口请求日志：

1. 为调用方身份记录每个请求时间戳。
2. 清除早于配置窗口的时间戳。
3. 如果当前滑动窗口内的请求已达到配额，API 返回 429。
4. Retry-After 和 X-RateLimit-Reset 根据窗口中仍然存在的最早请求计算。

这避免了固定窗口实现中常见的边界突增。

速率限制 Header

每个 API 响应都包含以下 header 中的速率限制信息：

响应 Header 示例

HTTP/1.1 200 OK
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 73
X-RateLimit-Reset: 42
Content-Type: application/json

超出速率限制（429 响应）

HTTP/1.1 429 Too Many Requests
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 42
Retry-After: 42
Content-Type: application/json

{
  "error_code": "RATE_LIMITED",
  "message": "请求过多。请在指定时间后重试。",
  "details": {
    "retry_after_seconds": 42,
    "correlation_id": "550e8400-e29b-41d4-a716-446655440000"
  },
  "timestamp": "2026-02-24T12:34:56Z"
}

重试策略

带抖动的指数退避

当你收到 429 Too Many Requests 响应时，实施带抖动的指数退避以避免惊群问题：

公式：wait_time = min(max_wait, base_delay * (2 ^ attempt)) + random_jitter

Python 示例

import time
import random
import requests
from typing import Optional

def call_api_with_retry(
    url: str,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> Optional[requests.Response]:
    """
    使用指数退避重试策略调用 API。

    Args:
        url: API 端点 URL
        max_retries: 最大重试次数
        base_delay: 初始延迟（秒）
        max_delay: 重试之间的最大延迟

    Returns:
        Response 对象；如果所有重试都耗尽则返回 None
    """
    for attempt in range(max_retries + 1):
        try:
            response = requests.get(url)

            # 主动检查速率限制 header
            remaining = int(response.headers.get('X-RateLimit-Remaining', 1))
            if remaining <= 5:
                print(f"Warning: 当前窗口仅剩 {remaining} 次请求")

            if response.status_code == 429:
                # 如果可用，使用 Retry-After header
                retry_after = int(response.headers.get('Retry-After', 0))
                if retry_after > 0:
                    wait_time = retry_after
                else:
                    # 计算带抖动的指数退避
                    wait_time = min(max_delay, base_delay * (2 ** attempt))
                    jitter = random.uniform(0, wait_time * 0.1)
                    wait_time += jitter

                print(f"Rate limited. 在重试 {attempt + 1}/{max_retries} 前等待 {wait_time:.2f}s")
                time.sleep(wait_time)
                continue

            # 成功或不可重试错误
            response.raise_for_status()
            return response

        except requests.RequestException as e:
            if attempt == max_retries:
                print(f"Max retries exhausted: {e}")
                return None

            # 网络错误的指数退避
            wait_time = min(max_delay, base_delay * (2 ** attempt))
            jitter = random.uniform(0, wait_time * 0.1)
            wait_time += jitter

            print(f"Request failed: {e}. 正在 {wait_time:.2f}s 后重试...")
            time.sleep(wait_time)

    return None

## 使用
response = call_api_with_retry('https://registry.soroban.example/api/contracts/search?query=token')
if response:
    data = response.json()
    print(f"Found {len(data['contracts'])} contracts")

JavaScript/TypeScript 示例

interface RetryConfig {
  maxRetries?: number;
  baseDelay?: number;
  maxDelay?: number;
}

interface RateLimitHeaders {
  limit: number;
  remaining: number;
  reset: number;
}

async function callApiWithRetry(
  url: string,
  options: RequestInit = {},
  config: RetryConfig = {}
): Promise<Response> {
  const {
    maxRetries = 5,
    baseDelay = 1000,
    maxDelay = 60000
  } = config;

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      const response = await fetch(url, options);

      // 解析速率限制 header
      const rateLimitHeaders: RateLimitHeaders = {
        limit: parseInt(response.headers.get('X-RateLimit-Limit') || '0'),
        remaining: parseInt(response.headers.get('X-RateLimit-Remaining') || '0'),
        reset: parseInt(response.headers.get('X-RateLimit-Reset') || '0')
      };

      // 主动速率限制警告
      if (rateLimitHeaders.remaining <= 5) {
        console.warn(
          `Rate limit warning: 剩余 ${rateLimitHeaders.remaining} 次请求。` +
          `将在 ${rateLimitHeaders.reset}s 后重置`
        );
      }

      if (response.status === 429) {
        const retryAfter = parseInt(response.headers.get('Retry-After') || '0');
        let waitTime: number;

        if (retryAfter > 0) {
          waitTime = retryAfter * 1000; // 转换为毫秒
        } else {
          // 带抖动的指数退避
          waitTime = Math.min(maxDelay, baseDelay * Math.pow(2, attempt));
          const jitter = Math.random() * waitTime * 0.1;
          waitTime += jitter;
        }

        console.log(
          `Rate limited. 在重试 ${attempt + 1}/${maxRetries} 前等待 ${(waitTime / 1000).toFixed(2)}s`
        );

        await new Promise(resolve => setTimeout(resolve, waitTime));
        continue;
      }

      // 成功或不可重试错误
      return response;

    } catch (error) {
      if (attempt === maxRetries) {
        throw new Error(`Max retries exhausted: ${error}`);
      }

      // 网络错误的指数退避
      const waitTime = Math.min(maxDelay, baseDelay * Math.pow(2, attempt));
      const jitter = Math.random() * waitTime * 0.1;
      const totalWait = waitTime + jitter;

      console.log(`Request failed: ${error}. 正在 ${(totalWait / 1000).toFixed(2)}s 后重试...`);
      await new Promise(resolve => setTimeout(resolve, totalWait));
    }
  }

  throw new Error('Max retries exhausted');
}

// 使用
try {
  const response = await callApiWithRetry(
    'https://registry.soroban.example/api/contracts/search?query=token',
    {
      method: 'GET',
      headers: {
        'Content-Type': 'application/json'
      }
    }
  );

  const data = await response.json();
  console.log(`Found ${data.contracts.length} contracts`);
} catch (error) {
  console.error('API call failed:', error);
}

Rust 示例

use reqwest::{Client, Response, StatusCode};
use std::time::Duration;
use tokio::time::sleep;

pub struct RetryConfig {
    pub max_retries: u32,
    pub base_delay_ms: u64,
    pub max_delay_ms: u64,
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_retries: 5,
            base_delay_ms: 1000,
            max_delay_ms: 60000,
        }
    }
}

pub async fn call_api_with_retry(
    client: &Client,
    url: &str,
    config: RetryConfig,
) -> Result<Response, Box<dyn std::error::Error>> {
    for attempt in 0..=config.max_retries {
        let response = client.get(url).send().await?;

        // 解析速率限制 header
        let remaining: u32 = response
            .headers()
            .get("x-ratelimit-remaining")
            .and_then(|h| h.to_str().ok())
            .and_then(|s| s.parse().ok())
            .unwrap_or(1);

        if remaining <= 5 {
            tracing::warn!(
                remaining,
                "Rate limit warning: 剩余请求较少"
            );
        }

        if response.status() == StatusCode::TOO_MANY_REQUESTS {
            let retry_after: u64 = response
                .headers()
                .get("retry-after")
                .and_then(|h| h.to_str().ok())
                .and_then(|s| s.parse().ok())
                .unwrap_or(0);

            let wait_ms = if retry_after > 0 {
                retry_after * 1000
            } else {
                // 带抖动的指数退避
                let base_wait = config.base_delay_ms * 2u64.pow(attempt);
                let wait = std::cmp::min(config.max_delay_ms, base_wait);
                let jitter = (rand::random::<f64>() * wait as f64 * 0.1) as u64;
                wait + jitter
            };

            tracing::info!(
                attempt,
                wait_ms,
                max_retries = config.max_retries,
                "Rate limited, waiting before retry"
            );

            sleep(Duration::from_millis(wait_ms)).await;
            continue;
        }

        // 成功或不可重试错误
        response.error_for_status_ref()?;
        return Ok(response);
    }

    Err("Max retries exhausted".into())
}

// 使用
##[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::new();
    let config = RetryConfig::default();

    let response = call_api_with_retry(
        &client,
        "https://registry.soroban.example/api/contracts/search?query=token",
        config,
    ).await?;

    let data: serde_json::Value = response.json().await?;
    println!("Found contracts: {}", data["contracts"].as_array().unwrap().len());

    Ok(())
}

最佳实践

1. 主动监控速率限制 Header

不要等到 429 响应才处理。对每个成功响应检查 X-RateLimit-Remaining，并在接近限制时降低速度：

remaining = int(response.headers.get('X-RateLimit-Remaining', 100))
reset_seconds = int(response.headers.get('X-RateLimit-Reset', 60))

if remaining <= 10:
    # 降低速度：在请求之间增加延迟
    delay = reset_seconds / remaining if remaining > 0 else reset_seconds
    time.sleep(delay)

2. 使用连接池

复用 HTTP 连接以减少开销：

## Python
session = requests.Session()
session.headers.update({'User-Agent': 'MyApp/1.0'})

## JavaScript/Node.js
const agent = new https.Agent({ keepAlive: true });

3. 实现请求批处理

在可能的情况下，使用批量端点减少请求次数：

## 而不是多个请求：
GET /api/contracts/{id1}
GET /api/contracts/{id2}
GET /api/contracts/{id3}

## 使用批量端点：
POST /api/contracts/batch
{
  "contract_ids": ["id1", "id2", "id3"]
}

4. 缓存响应

在本地缓存 API 响应以避免重复请求：

from functools import lru_cache
import time

@lru_cache(maxsize=1000)
def get_contract(contract_id: str, cache_key: int) -> dict:
    """使用基于时间的失效机制缓存合约数据"""
    response = requests.get(f'/api/contracts/{contract_id}')
    return response.json()

## 使用每 5 分钟变化一次的 cache key
cache_key = int(time.time() / 300)
contract = get_contract('contract_abc123', cache_key)

5. 在时间上分散负载

对于批量操作，将请求分散到一段时间内：

import asyncio

async def process_contracts_gradually(contract_ids: list[str]):
    """以受控速率处理合约"""
    for batch in chunks(contract_ids, size=10):
        tasks = [fetch_contract(cid) for cid in batch]
        await asyncio.gather(*tasks)
        # 批次之间等待以遵守速率限制
        await asyncio.sleep(6)  # 10 requests per minute = 每 6 秒 1 个批次

通过环境变量进行配置

可以使用环境变量自定义速率限制：

## 全局限制（每分钟）
RATE_LIMIT_ANON_PER_MINUTE=100          # 默认：100
## 向后兼容的回退：RATE_LIMIT_READ_PER_MINUTE
RATE_LIMIT_AUTH_PER_MINUTE=1000         # 默认：1000

## 时间窗口（秒）
RATE_LIMIT_WINDOW_SECONDS=60            # 默认：60

FAQ

Q: 速率限制是按用户还是按 IP 地址？

A: 匿名请求的速率限制是按 IP 地址。认证请求的速率限制是按 token，来自 Authorization header。

Q: 如果我超出速率限制会怎样？

A: 你会收到一个带有 Retry-After header 的 429 Too Many Requests 响应，指示你何时可以重试。你的请求不会被处理。

Q: 我可以请求更高的速率限制吗？

A: 对于生产部署或大流量集成，请联系 registry 运营方，讨论带有自定义限制的 enterprise tier 访问。

Q: 失败的请求会计入速率限制吗？

A: 会，所有请求（成功或失败）都会计入你的速率限制，以防止通过故意构造畸形请求进行滥用。

Q: 我如何进行认证以获得更高限制？

A: 在认证已实现时，包含一个带有有效 API key 的 Authorization header。认证请求会自动获得更高的认证层级限制（1,000 req/min）。

Q: WebSocket 连接也受速率限制吗？

A: WebSocket 连接目前不受支持。速率限制仅适用于 HTTP/REST API 请求。

Q: 我可以临时突破限制吗？

A: 速率限制器使用滑动窗口算法，因此突发流量会在最近的 RATE_LIMIT_WINDOW_SECONDS 区间内持续平滑处理。

故障排查

问题：意外收到 429 错误

可能原因：

1. 共享 IP 地址（NAT 后面的多个用户）
2. 没有延迟的激进轮询
3. 没有实现指数退避
4. 在紧密循环中发送请求

解决方案：

• 监控 X-RateLimit-Remaining header
• 实现带抖动的指数退避
• 在请求之间增加延迟
• 在可用时使用 webhooks/subscriptions 替代轮询
• 考虑请求批处理

问题：速率限制 header 缺失

可能原因：

• 使用了剥离 header 的反向代理
• 缓存层拦截了响应

解决方案：

• 检查代理配置
• 确保速率限制 header 被保留
• 直接发起 API 请求进行验证

问题：速率限制对使用场景来说过于严格

解决方案：

• 优化请求模式（批处理、缓存、过滤）
• 请求 enterprise tier 访问
• 使用认证端点以获得更高限制

相关文档

• 错误代码参考
• API 高级功能 - 批量操作
• 安全最佳实践 - API key 管理

支持

如需提高速率限制或遇到问题：

• GitHub Issues: https://github.com/ALIPHATICHYD/Soroban-Registry/issues
• 将问题标签设为：api, rate-limiting

• 原文链接： github.com/ALIPHATICHYD/...
• 登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～