Skip to content

Limit application operations

Use Tenchi's rate-limit primitive when an authenticated actor or tenant may perform an application operation only a fixed number of times in a window. The use case chooses the policy and scope; a replaceable store performs one atomic consume.

An exhausted window raises the standard RATE_LIMITED application error with HTTP status 429 and a Retry-After header.

Declare the HTTP outcome

Declare RATE_LIMITED on every contract that may expose it:

from tenchi.contracts import contract
from tenchi.rate_limits import RATE_LIMITED


create_task_contract = contract(
    method="POST",
    path="/tasks",
    request=CreateTask,
    response=Task,
    status=201,
    errors=(RATE_LIMITED,),
)

Tenchi documents the response and Retry-After header in OpenAPI. If the contract does not declare the error, the honesty rule turns it into a framework-owned 500 instead of exposing an undocumented 429.

Put shared storage in the context

Add the store protocol to the application context:

from dataclasses import dataclass

from tenchi.rate_limits import RateLimitStore


@dataclass(frozen=True, slots=True)
class AppContext:
    tasks: TaskRepository
    rate_limits: RateLimitStore
    user: User | None = None

The store has one async method:

from typing import Protocol

from tenchi.rate_limits import RateLimitDecision


class RateLimitStore(Protocol):
    async def consume(
        self,
        *,
        namespace: str,
        scope: str,
        limit: int,
        window_seconds: float,
        cost: int,
    ) -> RateLimitDecision: ...

consume() must be atomic and returns either:

The window semantics an adapter must implement are listed under Constraints.

Scope from verified identity

Enforce the policy after authentication, using identity derived from the application context:

from tenchi.rate_limits import enforce_rate_limit


async def create_task(request: CreateTask, context: AppContext) -> Task:
    user = require_user(context.user)
    await enforce_rate_limit(
        context.rate_limits,
        namespace="tasks.create",
        scope=user.id,
        limit=5,
        window_seconds=60,
    )
    return await context.tasks.create(
        project_id=request.project_id,
        title=request.title,
    )

The namespace is a stable dotted snake_case operation name. The scope is the actor, tenant, credential, or other application-owned subject that shares the allowance.

Do not use an account, tenant, or owner ID copied directly from request input. An attacker could select another subject's scope and exhaust its capacity. Authenticate first and derive the scope from verified context.

Use separate namespaces when operations have separate allowances. Use a tenant scope when every user in that tenant should share one budget. Prefixing values such as user:alice and tenant:acme can keep mixed scope kinds explicit.

Decide what one cost represents

Place the consume according to what the policy limits.

To count each attempt that reaches the use case, consume before its business operation in storage that commits independently. Those attempts cost capacity even if later application work fails. Malformed HTTP input is rejected before the use case runs, so this placement does not count it. Apply a separate boundary or gateway policy when you need to count rejected requests too.

For a logical-operation quota, combine it with idempotency so transport retries do not consume the allowance repeatedly:

async def create_task(
    headers: CreateTaskHeaders,
    request: CreateTask,
    context: AppContext,
) -> Task:
    user = require_user(context.user)

    async def create() -> Task:
        await enforce_rate_limit(
            context.rate_limits,
            namespace="tasks.create",
            scope=user.id,
            limit=5,
            window_seconds=60,
        )
        return await context.tasks.create(
            project_id=request.project_id,
            title=request.title,
        )

    return await run_idempotently(
        context.idempotency,
        namespace="tasks.create",
        scope=user.id,
        key=headers.idempotency_key,
        fingerprint=fingerprint(request, annotation=CreateTask),
        result_type=Task,
        operation=create,
    )

Here, a completed idempotent replay bypasses create() and costs nothing. Taskboard stores the rate-limit window in the same transaction as task creation, so a rollback also restores the capacity.

Choose independent storage instead when failed operation attempts must count. Write that choice down with the policy; transaction placement changes observable behavior during failures and retries.

Apply weighted costs

Use cost= when operations consume different amounts from one shared budget:

await enforce_rate_limit(
    context.rate_limits,
    namespace="reports.export",
    scope=f"tenant:{tenant.id}",
    limit=100,
    window_seconds=60 * 60,
    cost=10,
)

Tenchi validates the policy values before calling the store; see Constraints.

Test with the memory store

MemoryRateLimitStore is concurrency-safe within one process and accepts a clock for deterministic boundary tests:

import pytest

from tenchi.errors import AppError
from tenchi.rate_limits import (
    RATE_LIMITED,
    MemoryRateLimitStore,
    enforce_rate_limit,
)


now = 1_000.0
store = MemoryRateLimitStore(clock=lambda: now)

await enforce_rate_limit(
    store,
    namespace="tasks.create",
    scope="alice",
    limit=1,
    window_seconds=60,
)

with pytest.raises(AppError) as excinfo:
    await enforce_rate_limit(
        store,
        namespace="tasks.create",
        scope="alice",
        limit=1,
        window_seconds=60,
    )

assert excinfo.value.definition == RATE_LIMITED
assert excinfo.value.headers == {"Retry-After": "60"}

now += 60
permit = await enforce_rate_limit(
    store,
    namespace="tasks.create",
    scope="alice",
    limit=1,
    window_seconds=60,
)
assert permit.remaining == 0

Share one memory store across every context created by the test application. Creating a new store per request creates a new empty allowance and does not test rate limiting.

Do not deploy the memory store

MemoryRateLimitStore cannot coordinate multiple processes or hosts and loses all windows on restart. Use it for use-case tests, in-process HTTP tests, and local development only.

Implement production storage

A production adapter must atomically read and update one (namespace, scope) window. Use a conditional database write, a Redis script, or another operation that cannot admit two concurrent callers beyond the limit.

Run verify_rate_limit_store() against the adapter with a shared test backend and deterministic clock. It exercises capacity, rejected-cost behavior, resets, policy changes, identity isolation, and concurrent admission. Keep separate integration tests for transaction rollback and coordination across the processes or hosts you deploy.

Store at least:

Delete expired windows during consumption or through operational cleanup. Use the storage system's clock when several hosts share the store, so host clock skew cannot lengthen or shorten a caller's window.

Fixed windows can admit traffic near both sides of a reset boundary. Use an API gateway with a rolling-window, token-bucket, or other algorithm when that burst shape is unacceptable.

Keep edge protection at the edge

Tenchi's primitive protects authenticated application operations. It is not a denial-of-service boundary: the application has already accepted the connection, matched a route, and usually authenticated the caller.

Configure your reverse proxy, load balancer, CDN, or API gateway for:

Keep application limits for rules that need verified users, tenants, plans, operation names, or domain-specific weighted costs. Many deployed services need both layers.

Constraints