From 999b122093c8d267af27bb71f7f52924b2932284 Mon Sep 17 00:00:00 2001 From: extolkom Date: Mon, 27 Jul 2026 01:17:42 -1200 Subject: [PATCH 1/4] docs: add indexer/realtime operational runbook (#849) --- backend/SSE_README.md | 21 ++++ backend/docs/SSE_ARCHITECTURE.md | 170 +++++++++++++++++++++++++++++++ 2 files changed, 191 insertions(+) create mode 100644 backend/SSE_README.md diff --git a/backend/SSE_README.md b/backend/SSE_README.md new file mode 100644 index 00000000..e705d2e1 --- /dev/null +++ b/backend/SSE_README.md @@ -0,0 +1,21 @@ +# FlowFi Real-Time Event Streaming & Indexer Overview + +This document provides a quick index for real-time event streaming via Server-Sent Events (SSE) and on-chain Soroban event indexing in the FlowFi backend. + +## Documentation Map + +- **[SSE Architecture Overview](docs/SSE_ARCHITECTURE.md)**: Details the architecture, system flow, connection handling, subscription filtering, horizontal scaling with Redis, event broadcasting logic, memory capacity, and security layers. +- **[Operational Runbook](docs/SSE_ARCHITECTURE.md#operational-runbook)**: **Full operational runbook** covering health check endpoints (`/health` and `/v1/admin/metrics`), lag monitoring thresholds, indexer environment variables, admin reset vs. replay procedures, deduplication guarantees, and the RPC outage recovery walkthrough. +- **[SSE Client Implementation Guide](docs/SSE_IMPLEMENTATION.md)**: Describes client integration (`GET /events/subscribe`), supported query parameters (`streams`, `users`, `all`), event types (`stream.created`, `stream.topped_up`, etc.), reconnection strategies, and code examples. + +## Quick Endpoint Reference + +| Endpoint | Method | Auth | Description | +|----------|--------|------|-------------| +| `/health` | `GET` | Public | Liveness and readiness check. Reports `indexerLag` and returns `503` if lag exceeds 60s while enabled. | +| `/v1/admin/metrics` | `GET` | Admin JWT | Detailed health metrics including `indexer.lastLedger`, `indexer.lagSeconds`, and `sse.activeConnections`. | +| `/v1/admin/indexer/reset` | `POST` | Admin JWT | Reset indexer `lastLedger` pointer for the next scheduled poll cycle. | +| `/v1/admin/indexer/replay` | `POST` | Admin JWT | Reset `lastLedger` pointer and immediately trigger event polling batch. | +| `/events/subscribe` | `GET` | Public / Auth | Connect to real-time Server-Sent Events stream. | + +For detailed operational guidance and troubleshooting, refer to the [Operational Runbook](docs/SSE_ARCHITECTURE.md#operational-runbook). diff --git a/backend/docs/SSE_ARCHITECTURE.md b/backend/docs/SSE_ARCHITECTURE.md index e6f5ab21..4edc1158 100644 --- a/backend/docs/SSE_ARCHITECTURE.md +++ b/backend/docs/SSE_ARCHITECTURE.md @@ -199,3 +199,173 @@ Optimization: │ - Subscription filtering │ └─────────────────────────────────────────────────┘ ``` + +## Operational Runbook + +This section details operational procedures, health monitoring metrics, environment configuration, and recovery workflows for the indexer and real-time event streaming components. + +### Monitoring & Health Check + +The backend exposes two primary endpoints for observing indexer health and operational metrics. + +#### 1. Liveness & Readiness Health Endpoint (`GET /health`) + +Unauthenticated health endpoint suitable for load balancer probes, Kubernetes liveness/readiness checks, and monitoring alerts. + +* **URL**: `GET /health` +* **Response Schema**: + ```json + { + "status": "ok", + "db": "connected", + "indexerEnabled": true, + "indexerLag": 5, + "uptime": 3600 + } + ``` +* **Key Fields**: + * `status`: `"ok"` (healthy) or `"degraded"` (unhealthy / degraded). + * `db`: `"connected"` or `"disconnected"`. + * `indexerEnabled`: `true` if `STREAM_CONTRACT_ID` is set, `false` if unconfigured/disabled. + * `indexerLag`: Integer seconds since last indexer DB state update (`updatedAt`), or `null` on cold start (when no `IndexerState` row exists yet). + * `uptime`: Node process uptime in seconds. +* **Threshold & Status Logic**: + * **HTTP 200 OK** (`status: "ok"`): DB status is `"connected"` and `indexerDegraded` is `false`. + * **HTTP 503 Service Unavailable** (`status: "degraded"`): Triggered if DB status is `"disconnected"` OR `indexerDegraded` is `true`. + * **Degraded Rule**: `indexerDegraded = indexerEnabled && indexerLag > 60` (indexer enabled and lag strictly exceeds 60 seconds). + * **Cold Start Handling**: If no `IndexerState` record exists in the database (`indexerLag === -1` internally), `indexerLag` is returned as `null` and does **not** trigger degradation (returns HTTP 200 OK as long as DB is connected). + * **Disabled Handling**: If `STREAM_CONTRACT_ID` is unset (`indexerEnabled: false`), lag degradation checks are bypassed and `/health` returns HTTP 200 OK. + +##### On-Call Operational Actions by Health Status + +| `indexerLag` Value | HTTP Status | Operational Status | On-Call Action | +|-------------------|-------------|--------------------|----------------| +| `indexerLag <= 60s` | `200 OK` | **Healthy** | Normal operation. | +| `indexerLag > 60s` | `503 Service Unavailable` | **Degraded** | Indexer is stale or stalled. Check Soroban RPC connectivity (`SOROBAN_RPC_URL`), worker logs (`[SorobanWorker]`), and DB load. If RPC went down, execute the RPC Outage Recovery workflow below. | +| `indexerLag: null` | `200 OK` | **Cold Start** | Indexer state record has not yet been created. Wait for initial poll cycle to complete. | + +#### 2. Admin Health Metrics Endpoint (`GET /v1/admin/metrics`) + +Authenticated endpoint providing detailed protocol counters, cache status, active SSE connections, and indexer state. + +* **URL**: `GET /v1/admin/metrics` +* **Authentication**: Requires Admin JWT (`requireAdmin` middleware). +* **Caching**: Cached in Redis with key `admin:metrics` for 60 seconds (`X-Cache: HIT|MISS` header returned). +* **Indexer & Realtime Payload Fields**: + ```json + { + "indexer": { + "lastLedger": 1234567, + "lagSeconds": 5, + "lastUpdated": "2026-07-27T01:15:00.000Z" + }, + "sse": { + "activeConnections": 42 + }, + "events": { + "last24h": 1250 + } + } + ``` + * `indexer.lastLedger`: Last processed Stellar ledger sequence (`number`, defaults to 0). + * `indexer.lagSeconds`: Seconds elapsed since last indexer state update (`number | null`). + * `indexer.lastUpdated`: ISO timestamp of last indexer update (`string | null`). + * `sse.activeConnections`: Current number of active client SSE connections. + * `events.last24h`: Count of indexed stream events in the past 24 hours. + +--- + +### Indexer Environment Variables + +The indexer worker behavior is controlled by environment variables configured in `backend/src/workers/soroban-event-worker.ts`. + +| Environment Variable | Description | Default Value | Behavior when Disabled / Unset | +|----------------------|-------------|---------------|--------------------------------| +| `STREAM_CONTRACT_ID` | Contract address of the streaming smart contract on Stellar. | `""` (empty string) | **Disables indexer worker**. On startup, `sorobanEventWorker.start()` logs `[SorobanWorker] STREAM_CONTRACT_ID is not set — event indexing disabled.` and exits gracefully without starting the polling loop. `/health` reports `indexerEnabled: false` and skips 503 lag checks. | +| `SOROBAN_RPC_URL` | Endpoint URL for the Soroban RPC node. | `"https://soroban-testnet.stellar.org"` | Uses default public testnet RPC URL. | +| `INDEXER_POLL_INTERVAL_MS` | Polling interval in milliseconds between event fetch cycles. | `"5000"` (5 seconds) | Uses default 5000 ms interval. | +| `INDEXER_START_LEDGER` | Starting Stellar ledger sequence number for cold starts when no `IndexerState` record exists in the database. | `"0"` | Starts indexing from ledger 0 on initial setup. | + +--- + +### Admin Operations: Reset vs. Replay + +The backend provides two admin endpoints for managing indexer positioning and re-processing events. Both require Admin JWT authentication (`requireAdmin`). + +#### 1. Reset Indexer (`POST /v1/admin/indexer/reset`) + +* **Route**: `POST /v1/admin/indexer/reset` +* **Request Body**: + ```json + { + "ledger": 1234500 + } + ``` +* **Handler Behavior**: + * Calls `resetIndexer(toLedger)` in `backend/src/services/indexerService.ts:30`. + * Upserts the singleton row in `prisma.indexerState` (`id: "singleton"`), setting `lastLedger` to `toLedger` and `lastCursor` to `null`. + * **Does NOT immediately trigger a poll**. The worker will pick up from `toLedger` on its next scheduled interval (`INDEXER_POLL_INTERVAL_MS`). + +#### 2. Replay Indexer (`POST /v1/admin/indexer/replay`) + +* **Route**: `POST /v1/admin/indexer/replay?from_ledger=1234500` +* **Query Parameter**: `from_ledger` (non-negative integer). +* **Response**: `202 Accepted` (`{ "ok": true, "replayingFrom": 1234500 }`). +* **Handler Behavior**: + * Calls `replayFromLedger(fromLedger)` in `backend/src/services/indexerService.ts:49`. + * Calls `resetIndexer(fromLedger)` to update `lastLedger` and clear `lastCursor`. + * **Immediately triggers an out-of-band poll cycle** via `sorobanEventWorker.triggerPoll()` without waiting for the next polling interval timer. + +#### Operational Comparison + +| Feature / Scenario | Reset (`POST /v1/admin/indexer/reset`) | Replay (`POST /v1/admin/indexer/replay`) | +|--------------------|------------------------------------------|-------------------------------------------| +| **Execution** | Passive: Updates DB pointer only. | Active: Updates DB pointer **and** triggers immediate poll batch. | +| **Use Case** | Reconfiguring start ledger prior to scheduled maintenance or service restart. | Recovering from RPC outages or backfilling missed ledgers immediately. | +| **Parameters** | JSON body: `{ "ledger": }` | Query param: `?from_ledger=` | + +#### Event Deduplication & Idempotency Guarantee + +Replaying events is safe against duplicate event creation in the database: + +* **Database Constraint**: Grounded in `backend/prisma/schema.prisma:89`, the `StreamEvent` model enforces a unique compound index: + ```prisma + @@unique([transactionHash, eventType]) + ``` +* **Worker Execution**: In `backend/src/workers/soroban-event-worker.ts`, event handlers execute `prisma.streamEvent.upsert(...)` matching on `{ transactionHash_eventType: { transactionHash, eventType } }`. If an event for that transaction and type has already been recorded, the insertion is skipped. +* **Operational Note / Caveat**: As documented in `backend/src/services/indexerService.ts:44-47`, deduplication applies to `StreamEvent` log rows. Stream entity state mutations (such as `Stream.withdrawnAmount` in `handleTokensWithdrawn`) are currently updated unconditionally upon reprocessing events; full state mutation idempotency is tracked under issue #808. + +--- + +### RPC Outage Recovery Walkthrough + +When the Soroban RPC node experiences downtime or network disruption, the indexer will pause event ingestion, causing `indexerLag` on `/health` to increase beyond 60 seconds and report HTTP 503. Follow these steps to restore service: + +1. **Identify the Last Processed Ledger**: + Query the metrics endpoint to obtain the last processed ledger before the outage: + ```bash + curl -H "Authorization: Bearer " http://localhost:3001/v1/admin/metrics + ``` + Note `indexer.lastLedger` from the JSON response. + +2. **Verify RPC Connectivity**: + Ensure `SOROBAN_RPC_URL` is reachable and serving valid Stellar ledger data. + +3. **Trigger Event Replay**: + Call the replay endpoint passing the last known valid ledger sequence: + ```bash + curl -X POST -H "Authorization: Bearer " \ + "http://localhost:3001/v1/admin/indexer/replay?from_ledger=" + ``` + +4. **Monitor Catch-Up Progress**: + Continuously monitor progress via `GET /v1/admin/metrics`: + * Observe `indexer.lastLedger` incrementing toward the current network ledger. + * Observe `indexer.lagSeconds` decreasing. + +5. **Verify Health Restoration**: + Confirm `/health` returns HTTP `200 OK` with `status: "ok"` and `indexerLag <= 60`: + ```bash + curl http://localhost:3001/health + ``` + From e5acff661161991bee4db447fc9d8fa41871570b Mon Sep 17 00:00:00 2001 From: extolkom Date: Mon, 27 Jul 2026 02:47:05 -1200 Subject: [PATCH 2/4] fix(backend): add pretest script to run prisma generate before test suite --- backend/package.json | 1 + 1 file changed, 1 insertion(+) diff --git a/backend/package.json b/backend/package.json index 42c27d07..c94c733a 100644 --- a/backend/package.json +++ b/backend/package.json @@ -6,6 +6,7 @@ "main": "index.js", "scripts": { "prebuild": "prisma generate", + "pretest": "prisma generate", "test": "vitest run", "dev": "nodemon", "build": "tsc", From 2361885956807aa72a53ca4c55706829b83c57c4 Mon Sep 17 00:00:00 2001 From: extolkom Date: Fri, 31 Jul 2026 22:55:58 -1200 Subject: [PATCH 3/4] fix(ci): merge origin/main into PR branch and fix frontend lint and test errors --- ...ms.test.ts => useIncomingStreams.test.tsx} | 29 +++++++++++-------- frontend/src/lib/dashboard.ts | 24 +++++++-------- 2 files changed, 28 insertions(+), 25 deletions(-) rename frontend/src/hooks/{useIncomingStreams.test.ts => useIncomingStreams.test.tsx} (79%) diff --git a/frontend/src/hooks/useIncomingStreams.test.ts b/frontend/src/hooks/useIncomingStreams.test.tsx similarity index 79% rename from frontend/src/hooks/useIncomingStreams.test.ts rename to frontend/src/hooks/useIncomingStreams.test.tsx index 39a9b9d8..b7fcebb7 100644 --- a/frontend/src/hooks/useIncomingStreams.test.ts +++ b/frontend/src/hooks/useIncomingStreams.test.tsx @@ -7,8 +7,9 @@ import { useWithdrawIncomingStream, incomingStreamsQueryKey, } from "./useIncomingStreams"; -import { fetchIncomingStreams } from "@/lib/api/streams"; +import { fetchIncomingStreams, type IncomingStreamRecord } from "@/lib/api/streams"; import { withdrawFromStream } from "@/lib/soroban"; +import type { WalletSession } from "@/lib/wallet"; vi.mock("@/lib/api/streams", () => ({ fetchIncomingStreams: vi.fn(), @@ -74,17 +75,18 @@ describe("useIncomingStreams hooks", () => { ); await expect( - result.current.mutateAsync({} as any) + result.current.mutateAsync({} as unknown as IncomingStreamRecord) ).rejects.toThrow("Please connect your wallet first"); expect(withdrawFromStream).not.toHaveBeenCalled(); }); it("invalidates incomingStreamsQueryKey(publicKey) on success", async () => { - (withdrawFromStream as any).mockResolvedValue({ status: "success" }); - (fetchIncomingStreams as any).mockResolvedValue([]); + vi.useFakeTimers(); + vi.mocked(withdrawFromStream).mockResolvedValue({ status: "success" } as never); + vi.mocked(fetchIncomingStreams).mockResolvedValue([]); const { result } = renderHook( - () => useWithdrawIncomingStream({} as any, "pubkey"), + () => useWithdrawIncomingStream({} as WalletSession, "pubkey"), { wrapper } ); @@ -99,15 +101,18 @@ describe("useIncomingStreams hooks", () => { ratePerSecond: 1, isPaused: false, lastUpdateTime: Date.now() / 1000, - } as any); + } as unknown as IncomingStreamRecord); }); - // Wait for pollIndexerForWithdraw to complete and call invalidateQueries - await waitFor(() => { - expect(invalidateSpy).toHaveBeenCalledWith({ - queryKey: incomingStreamsQueryKey("pubkey"), - }); - }, { timeout: 10000 }); + await act(async () => { + await vi.advanceTimersByTimeAsync(65000); + }); + + expect(invalidateSpy).toHaveBeenCalledWith({ + queryKey: incomingStreamsQueryKey("pubkey"), + }); + + vi.useRealTimers(); }); }); }); diff --git a/frontend/src/lib/dashboard.ts b/frontend/src/lib/dashboard.ts index 39a8814f..303f1289 100644 --- a/frontend/src/lib/dashboard.ts +++ b/frontend/src/lib/dashboard.ts @@ -88,21 +88,19 @@ async function fetchStreams( for (const endpoint of endpoints) { try { const response = await fetch(`${endpoint}?${params.toString()}`, { signal }); - if (response.ok) { - const payload = (await response.json()) as - | BackendStream[] - | { data?: BackendStream[] }; - return Array.isArray(payload) ? payload : payload.data ?? []; - } - - if (response.status === 404) { - lastError = new Error(`Endpoint not found: ${endpoint}`); - continue; - } + if (response.ok) { + const payload = (await response.json()) as + | BackendStream[] + | { data?: BackendStream[] }; + return Array.isArray(payload) ? payload : payload.data ?? []; + } - lastError = new Error(`Failed to fetch streams (${response.status}) from ${endpoint}`); - } + if (response.status === 404) { + lastError = new Error(`Endpoint not found: ${endpoint}`); + continue; + } + lastError = new Error(`Failed to fetch streams (${response.status}) from ${endpoint}`); } catch (err) { if (err instanceof Error && err.name === "AbortError") { throw err; From 114e5a108b890cb850a18d11554f615eb5dd1176 Mon Sep 17 00:00:00 2001 From: extolkom Date: Fri, 31 Jul 2026 23:01:41 -1200 Subject: [PATCH 4/4] fix(frontend): use fake timers for polling test in useIncomingStreams.test.tsx --- frontend/src/hooks/useIncomingStreams.test.tsx | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/frontend/src/hooks/useIncomingStreams.test.tsx b/frontend/src/hooks/useIncomingStreams.test.tsx index 73226205..60068461 100644 --- a/frontend/src/hooks/useIncomingStreams.test.tsx +++ b/frontend/src/hooks/useIncomingStreams.test.tsx @@ -112,6 +112,7 @@ describe("useIncomingStreams hooks", () => { }); it("invalidates incomingStreamsQueryKey(publicKey) on success", async () => { + vi.useFakeTimers(); vi.mocked(withdrawFromStream).mockResolvedValue({ success: true, txHash: "tx-hash" }); vi.mocked(fetchIncomingStreams).mockResolvedValue([]); @@ -134,12 +135,15 @@ describe("useIncomingStreams hooks", () => { } as unknown as IncomingStreamRecord); }); - // Wait for pollIndexerForWithdraw to complete and call invalidateQueries - await waitFor(() => { - expect(invalidateSpy).toHaveBeenCalledWith({ - queryKey: incomingStreamsQueryKey("pubkey"), - }); - }, { timeout: 10000 }); + await act(async () => { + await vi.advanceTimersByTimeAsync(65000); + }); + + expect(invalidateSpy).toHaveBeenCalledWith({ + queryKey: incomingStreamsQueryKey("pubkey"), + }); + + vi.useRealTimers(); }); }); });