Infrastructure/caching/ — In-Memory State & Rate Limiting

Contents

Architecture

The caching sub-module is organized around three pillars: presence & state maps (system-presence), event buses (heartbeat-events, system-events), and rate control (fixed-window-rate-limit, retry-policy, dedupe). External callers interact through clean functional interfaces while internal state is managed via global singletons and scoped maps.

flowchart TD
  subgraph Callers
    SVC["Service Layer"]
    API["API Gateway"]
    WS["WebSocket Handler"]
  end

  subgraph caching["caching/"]
    SP["system-presence.ts\nIn-memory map\n5-min TTL, max 200"]
    HB["heartbeat-events.ts\nGlobal singleton bus\nSymbol.for keyed"]
    SE["system-events.ts\nSession-scoped queues\nMax 20 per queue"]
    RL["fixed-window-rate-limit.ts\nFixedWindowRateLimiter\nconsume / reset"]
    RP["retry-policy.ts\nExponential backoff\n3 attempts, jitter"]
    DD["dedupe.ts\nPayload deduplication"]
  end

  SVC --> SP
  SVC --> HB
  SVC --> SE
  API --> RL
  API --> RP
  WS --> DD
  RL --> RP
  SE --> DD
  SP --> HB

Key Classes & Functions

Data Structures

Interaction Flow

This sequence diagram shows a typical flow when a service checks presence, rate-limits an outbound API call, and falls back through the retry policy on a 429 response.

sequenceDiagram
  participant S as Service Layer
  participant SP as system-presence
  participant RL as FixedWindowRateLimiter
  participant RP as retry-policy
  participant API as External API

  S->>SP: getPresence(deviceId)
  SP-->>S: SystemPresence or undefined

  S->>RL: consume()
  alt allowed
    RL-->>S: { allowed: true, remaining: N }
    S->>API: POST /send
    API-->>S: 200 OK
  else rate limited
    RL-->>S: { allowed: false, retryAfterMs: 1200 }
    S->>RP: createRateLimitRetryRunner()
    RP->>API: POST /send (attempt 1)
    API-->>RP: 429 Too Many Requests
    Note over RP: wait 400ms + jitter
    RP->>API: POST /send (attempt 2)
    API-->>RP: 200 OK
    RP-->>S: result
  end

Interactions

Configuration

Directory Map

caching/
├── system-presence.ts           ← Global in-memory presence map (5-min TTL, max 200)
├── heartbeat-events.ts          ← Singleton heartbeat event bus via Symbol.for
├── system-events.ts             ← Session-scoped ephemeral event queues (max 20)
├── fixed-window-rate-limit.ts   ← FixedWindowRateLimiter: consume() / reset()
├── retry-policy.ts              ← Exponential backoff runners (rate-limit & Telegram)
├── dedupe.ts                    ← Payload deduplication utilities
├── cache_strategies/            ← Deep-dive: TTL, LRU, and eviction strategies
├── invalidation/                ← Deep-dive: Cache invalidation patterns and hooks
└── memory_management/           ← Deep-dive: Bounded maps and memory pressure handling

Deep-Dive Highlights

Core mechanisms from the caching sub-module have been extracted into dedicated focus pages for in-depth exploration. Each page covers one algorithm, protocol, or subsystem with diagrams, code walks, and configuration details.

Deep-Dive Pages

Code Examples

Example: Checking and Updating System Presence

import { getPresence, updatePresence } from "./system-presence";
import type { SystemPresence, SystemPresenceUpdate } from "./system-presence";

// Look up cached presence for a device
const current: SystemPresence | undefined = getPresence("device-abc-123");

if (!current) {
  // Self-presence initialization: hostname, IP, macOS model, version
  const entry: SystemPresence = {
    host: os.hostname(),
    ip: getLocalIP(),
    version: "2.4.1",
    platform: process.platform,
    deviceFamily: getMacModel(),
    lastInputSeconds: 0,
    mode: "active",
    deviceId: "device-abc-123",
    roles: ["editor"],
    scopes: ["read", "write"],
    instanceId: randomUUID(),
    text: "Online",
    ts: Date.now(),
  };

  // Stored with 5-min TTL, auto-evicted when map exceeds 200 entries
  const update: SystemPresenceUpdate = updatePresence("device-abc-123", entry);
  console.log("Changed fields:", update.changes);
}

Example: Subscribing to Heartbeat Events

import {
  emitHeartbeatEvent,
  onHeartbeatEvent,
  getLastHeartbeatEvent,
} from "./heartbeat-events";

// Subscribe — returns an unsubscribe function
const unsubscribe = onHeartbeatEvent((evt) => {
  console.log("Heartbeat received:", evt);
});

// Emit a heartbeat (broadcasts to all listeners via global singleton)
emitHeartbeatEvent({ ts: Date.now(), status: "alive" });

// Query the most recent event without subscribing
const last = getLastHeartbeatEvent();
console.log("Last heartbeat:", last?.ts);

// Cleanup
unsubscribe();

Example: Fixed-Window Rate Limiting with Retry

import { FixedWindowRateLimiter } from "./fixed-window-rate-limit";
import { createRateLimitRetryRunner } from "./retry-policy";

// Allow 100 requests per 60-second window
const limiter = new FixedWindowRateLimiter({
  maxRequests: 100,
  windowMs: 60_000,
});

const result = limiter.consume();
// result: { allowed: boolean, retryAfterMs: number, remaining: number }

if (result.allowed) {
  await fetch("https://api.example.com/data");
} else {
  console.warn(`Rate limited. Retry after ${result.retryAfterMs}ms`);
}

// Retry runner: 3 attempts, 400ms min, 30s max, 0.1 jitter
// Matches: /429|timeout|connect|reset|closed|unavailable|temporarily/i
const runner = createRateLimitRetryRunner({
  attempts: 3,
  minDelay: 400,
  maxDelay: 30_000,
  jitter: 0.1,
});

const response = await runner(() => fetch("https://api.example.com/send"));

Example: Pushing Session-Scoped System Events

import { pushSystemEvent, getSessionEvents } from "./system-events";
import type { SystemEvent } from "./system-events";

// Push an event into the session-scoped queue (max 20 per queue)
// Automatically deduplicates and emits to internal hooks
pushSystemEvent({
  text: "User connected",
  contextKey: "session-xyz",
  deliveryContext: "websocket",
  trusted: true,
});

// Retrieve events for a session
const events: SystemEvent[] = getSessionEvents("session-xyz");
events.forEach((e) => console.log(`[${e.ts}] ${e.text}`));