feat(ingest): add schema verification, cycle deadlines, lag telemetry, and /status health breakdown

.changeset/ingestion-telemetry-resiliency.md

@@ -0,0 +1,10 @@
+---
+'lens': patch
+---
+
+Add database schema startup verification, ingester cycle deadlines, ingestion lag telemetry, and `/status` venue health breakdown (Closes #352, #351, #340, #333).
+
+- **#352**: Added `verifyDatabaseSchema` and `assertSchemaValid` in `src/db/schemaVerify.ts` checking required tables and columns at boot, exposed via new Kubernetes `/ready` probe. Documented migration strategy in `docs/database-schema.md`.
+- **#351**: Implemented `runCycleWithDeadline` in `src/ingest/venues/runner.ts` using configurable timeout `[VENUE]_CYCLE_TIMEOUT_MS` / `INGESTER_CYCLE_TIMEOUT_MS` (default 30s) to abandon hanging poll cycles and increment `ingestion_cycles_abandoned_total` without concurrency drift.
+- **#340**: Exported Prometheus metrics `ingestion_lag_seconds`, `ingestion_cycle_duration_seconds`, `ingestion_errors_total` with bounded categories (`network`, `upstream_4xx`, `upstream_5xx`, `db`, `parse`, `timeout`, `unknown`).
+- **#333**: Enhanced `GET /status` to report `venues` breakdown with per-venue health (`healthy`, `stale`, `offline`), `lastWrite`, `ageSeconds`, and overall `ingestionHealth` (`healthy`, `degraded`, `unhealthy`).

.env.example

@@ -225,3 +225,10 @@ MCP_PAID_CALL_ALLOWED_HOSTS=
 # against a table that has never been pruned cannot hold a long lock or blow
 # the WAL. Default 1000.
 SNAPSHOT_PRUNE_BATCH_SIZE=1000
+
+# ─── Ingestion Cycle Timeouts & Telemetry (#351, #340, #333) ───────────────────
+# Upper bound deadline for a single ingestion cycle in milliseconds (default: 30000).
+# If a cycle runs longer than this, it is safely abandoned, an abandoned metric
+# is incremented, and the ingester proceeds to its next scheduled cycle without overlapping.
+# Can also be tuned per-venue, e.g. SDEX_CYCLE_TIMEOUT_MS, AMM_CYCLE_TIMEOUT_MS, SOROSWAP_CYCLE_TIMEOUT_MS.
+INGESTER_CYCLE_TIMEOUT_MS=30000

docs/database-schema.md

@@ -0,0 +1,31 @@
+# Database Schema & Migration Strategy
+
+## Current State: `prisma db push`
+
+`prisma/` currently contains `schema.prisma` without a `migrations/` directory. Lens currently leverages `prisma db push` for schema updates.
+
+### Architectural Rationale
+
+1. **Rapid DeFi Primitive Prototyping:**
+   During the initial development of the multi-venue aggregator (covering SDEX orderbooks, AMM liquidity pools, Soroswap factory pairs, and Reflector oracles), data models evolved rapidly across iterations. `prisma db push` allowed agile schema prototyping without accumulating redundant migration step churn.
+
+2. **Zero In-Process Migrations on Container Boot:**
+   Running complex automated migrations inside container entrypoints poses locking hazards when running horizontally scaled nodes against shared PostgreSQL databases. Decoupling schema validation from schema migration prevents distributed boot race conditions.
+
+3. **Time-Series Partitioning Compatibility:**
+   Lens's high-frequency tables (such as `price_points` and `price_aggregates`) require composite primary keys (e.g. `[id, timestamp]`) and time-series optimizations.
+
+## Startup Verification Guard (#352)
+
+To ensure that running code never crashes at request time due to missing columns or unapplied schema changes:
+- `src/db/schemaVerify.ts` runs on boot via `assertSchemaValid()` / `verifyDatabaseSchema()`.
+- It performs a low-overhead inspection against PostgreSQL `information_schema.tables` and `information_schema.columns`.
+- If an expected table or column is absent, it fails fast and names the missing entities.
+- `/ready` and `/status` expose `schemaReadiness` so load balancers and orchestrators isolate instances before they receive live traffic.
+
+## Path Forward: Transitioning to `prisma migrate`
+
+When moving to formal multi-region enterprise production deployments:
+1. Generate baseline migration from the current production database: `npx prisma migrate diff --from-empty --to-schema-datamodel prisma/schema.prisma --script > prisma/migrations/0_init/migration.sql`.
+2. Mark baseline as applied: `npx prisma migrate resolve --applied 0_init`.
+3. Enforce CI/CD pipeline validation using Prisma Migrate with the schema startup verification guard serving as the runtime barrier.

openapi.json

@@ -156,6 +156,18 @@
                     "ingestLagSeconds": {
                       "type": "integer",
                       "nullable": true
+                    },
+                    "ingestionHealth": {
+                      "type": "string",
+                      "enum": [
+                        "healthy",
+                        "degraded",
+                        "unhealthy"
+                      ]
+                    },
+                    "venues": {
+                      "type": "object",
+                      "additionalProperties": true
                     }
                   }
                 }
@@ -168,6 +180,132 @@
         }
       }
     },
+    "/ready": {
+      "get": {
+        "operationId": "getReadiness",
+        "summary": "Service readiness probe verifying database schema matches models",
+        "security": [],
+        "responses": {
+          "200": {
+            "description": "Service and database schema are ready",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "required": [
+                    "ready",
+                    "status",
+                    "schema"
+                  ],
+                  "properties": {
+                    "ready": {
+                      "type": "boolean"
+                    },
+                    "status": {
+                      "type": "string",
+                      "enum": [
+                        "ready",
+                        "not_ready"
+                      ]
+                    },
+                    "schema": {
+                      "type": "object",
+                      "required": [
+                        "status"
+                      ],
+                      "properties": {
+                        "status": {
+                          "type": "string",
+                          "enum": [
+                            "valid",
+                            "mismatch"
+                          ]
+                        },
+                        "missingTables": {
+                          "type": "array",
+                          "items": {
+                            "type": "string"
+                          }
+                        },
+                        "missingColumns": {
+                          "type": "array",
+                          "items": {
+                            "type": "string"
+                          }
+                        },
+                        "checkedAt": {
+                          "type": "string"
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }
+          },
+          "429": {
+            "$ref": "#/components/responses/TooManyRequests"
+          },
+          "503": {
+            "description": "Database schema mismatch or service not ready",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "object",
+                  "required": [
+                    "ready",
+                    "status",
+                    "schema"
+                  ],
+                  "properties": {
+                    "ready": {
+                      "type": "boolean"
+                    },
+                    "status": {
+                      "type": "string",
+                      "enum": [
+                        "ready",
+                        "not_ready"
+                      ]
+                    },
+                    "schema": {
+                      "type": "object",
+                      "required": [
+                        "status"
+                      ],
+                      "properties": {
+                        "status": {
+                          "type": "string",
+                          "enum": [
+                            "valid",
+                            "mismatch"
+                          ]
+                        },
+                        "missingTables": {
+                          "type": "array",
+                          "items": {
+                            "type": "string"
+                          }
+                        },
+                        "missingColumns": {
+                          "type": "array",
+                          "items": {
+                            "type": "string"
+                          }
+                        },
+                        "checkedAt": {
+                          "type": "string"
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/price/{assetA}/{assetB}": {
       "get": {
         "operationId": "getPrice",

openapi.yaml

@@ -114,6 +114,81 @@ paths:
                   ingestLagSeconds:
                     type: integer
                     nullable: true
+                  ingestionHealth:
+                    type: string
+                    enum: [healthy, degraded, unhealthy]
+                  venues:
+                    type: object
+                    additionalProperties: true
+        '429':
+          $ref: '#/components/responses/TooManyRequests'
+
+  /ready:
+    get:
+      operationId: getReadiness
+      summary: Service readiness probe verifying database schema matches models
+      security: []
+      responses:
+        '200':
+          description: Service and database schema are ready
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [ready, status, schema]
+                properties:
+                  ready:
+                    type: boolean
+                  status:
+                    type: string
+                    enum: [ready, not_ready]
+                  schema:
+                    type: object
+                    required: [status]
+                    properties:
+                      status:
+                        type: string
+                        enum: [valid, mismatch]
+                      missingTables:
+                        type: array
+                        items:
+                          type: string
+                      missingColumns:
+                        type: array
+                        items:
+                          type: string
+                      checkedAt:
+                        type: string
+        '503':
+          description: Database schema mismatch or service not ready
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [ready, status, schema]
+                properties:
+                  ready:
+                    type: boolean
+                  status:
+                    type: string
+                    enum: [ready, not_ready]
+                  schema:
+                    type: object
+                    required: [status]
+                    properties:
+                      status:
+                        type: string
+                        enum: [valid, mismatch]
+                      missingTables:
+                        type: array
+                        items:
+                          type: string
+                      missingColumns:
+                        type: array
+                        items:
+                          type: string
+                      checkedAt:
+                        type: string
         '429':
           $ref: '#/components/responses/TooManyRequests'