diff --git a/apps/web/app/api/registry/specs/route.ts b/apps/web/app/api/registry/specs/route.ts new file mode 100644 index 00000000..fddcfde5 --- /dev/null +++ b/apps/web/app/api/registry/specs/route.ts @@ -0,0 +1,62 @@ +import { getSpecStore, getVerdictStore } from "@/lib/registry"; +import type { RegisteredSpec } from "@orbital-stellar/abi-registry"; +import { validateSpec } from "@orbital-stellar/abi-registry"; + +export const dynamic = "force-dynamic"; + +export async function GET() { + const specs = await getSpecStore().getAll(); + const verdicts = await getVerdictStore().getAll(); + const verdictMap = new Map(verdicts.map((v) => [v.contractId, v])); + + const result = specs.map((spec) => ({ + ...spec, + latestVerdict: verdictMap.get(spec.contractId) ?? null, + })); + + return Response.json(result); +} + +export async function POST(req: Request) { + try { + const body = (await req.json()) as { + contractId: string; + spec: Record; + publisher?: string; + }; + + if (!body.contractId || !body.spec) { + return Response.json( + { error: "invalid_request", message: "contractId and spec are required" }, + { status: 400 }, + ); + } + + const validation = validateSpec(body.spec); + if (!validation.valid) { + return Response.json( + { error: "invalid_spec", message: validation.errors.join("; ") }, + { status: 400 }, + ); + } + + const registered: RegisteredSpec = { + contractId: body.contractId, + spec: body.spec as RegisteredSpec["spec"], + publisher: body.publisher, + submittedAt: new Date().toISOString(), + }; + + await getSpecStore().register(registered); + + return Response.json(registered, { status: 201 }); + } catch (error) { + return Response.json( + { + error: "registration_failed", + message: error instanceof Error ? error.message : "Unknown error", + }, + { status: 500 }, + ); + } +} diff --git a/apps/web/app/api/registry/verdicts/route.ts b/apps/web/app/api/registry/verdicts/route.ts new file mode 100644 index 00000000..f77a6daf --- /dev/null +++ b/apps/web/app/api/registry/verdicts/route.ts @@ -0,0 +1,21 @@ +import { getVerdictStore } from "@/lib/registry"; + +export const dynamic = "force-dynamic"; + +export async function GET(req: Request) { + const { searchParams } = new URL(req.url); + const contractId = searchParams.get("contractId"); + + const store = getVerdictStore(); + + if (contractId) { + const verdict = await store.getLatest(contractId); + if (!verdict) { + return Response.json({ error: "not_found", message: "No verdict for this contract" }, { status: 404 }); + } + return Response.json(verdict); + } + + const verdicts = await store.getAll(); + return Response.json(verdicts); +} diff --git a/apps/web/app/api/registry/verify/route.ts b/apps/web/app/api/registry/verify/route.ts new file mode 100644 index 00000000..53e9618b --- /dev/null +++ b/apps/web/app/api/registry/verify/route.ts @@ -0,0 +1,18 @@ +import { runVerification } from "@/lib/registry"; + +export const dynamic = "force-dynamic"; + +export async function POST() { + try { + const result = await runVerification(); + return Response.json(result); + } catch (error) { + return Response.json( + { + error: "verification_failed", + message: error instanceof Error ? error.message : "Unknown error", + }, + { status: 500 }, + ); + } +} diff --git a/apps/web/app/registry/[contractId]/page.tsx b/apps/web/app/registry/[contractId]/page.tsx new file mode 100644 index 00000000..364c5a07 --- /dev/null +++ b/apps/web/app/registry/[contractId]/page.tsx @@ -0,0 +1,486 @@ +import { getSpecStore, getVerdictStore } from "@/lib/registry"; + +export const dynamic = "force-dynamic"; + +function StatusBadge({ status }: { status: string }) { + const colors: Record = { + verified: { bg: "#0a2a0a", border: "#1a4a1a", text: "#4ade80" }, + mismatch: { bg: "#2a0a0a", border: "#4a1a1a", text: "#ff5370" }, + unverifiable: { bg: "#2a2a00", border: "#4a4a00", text: "#facc15" }, + }; + + const c = colors[status] ?? { bg: "var(--surface2)", border: "var(--border)", text: "var(--muted)" }; + + return ( + + {status.toUpperCase()} + + ); +} + +export default async function ContractDetailPage({ + params, +}: { + params: Promise<{ contractId: string }>; +}) { + const { contractId } = await params; + const spec = await getSpecStore().get(contractId); + const history = await getVerdictStore().getHistory(contractId); + const latestVerdict = history[history.length - 1] ?? null; + + if (!spec) { + return ( +
+
+

+ Contract Not Found +

+

+ No spec registered for {contractId}. +

+
+
+ ); + } + + const isMismatch = latestVerdict?.status === "mismatch"; + const isUnverifiable = latestVerdict?.status === "unverifiable"; + + return ( +
+
+ {isMismatch && ( +
+ ⚠ SCHEMA MISMATCH +
+ The submitted schema does not match the on-chain contract spec. + {latestVerdict?.diffs && latestVerdict.diffs.length > 0 && ( + Found {latestVerdict.diffs.length} difference(s). + )} +
+ )} + + {isUnverifiable && ( +
+ ⓘ UNVERIFIABLE +
+ {latestVerdict?.reason ?? "This contract has no embedded spec (pre-SEP-48 or non-WASM)."} +
+ Displayed as attested-only — not verified. +
+ )} + +
+

+ {spec.spec.name} +

+ {latestVerdict && } +
+ +
+
+

+ Contract Info +

+
+
Contract ID
+
{contractId}
+
Version
+
{spec.spec.version}
+
Network
+
{spec.spec.network ?? "unknown"}
+
+
+ +
+

+ Verification +

+
+
Status
+
+ {latestVerdict?.status ?? "never verified"} +
+ {latestVerdict?.verifiedAt && ( + <> +
+ Last Verified +
+
+ {new Date(latestVerdict.verifiedAt).toLocaleString()} +
+ + )} + {latestVerdict?.previousStatus && ( + <> +
+ Previous Status +
+
+ +
+ + )} +
+
+
+ + {latestVerdict?.diffs && latestVerdict.diffs.length > 0 && ( +
+
+ Differences ({latestVerdict.diffs.length}) +
+
+ {latestVerdict.diffs.map((diff, i) => ( +
+

+ {diff.path} +

+
+
+

+ Submitted +

+ 
+                        {JSON.stringify(diff.submitted, null, 2) ?? "null"}
+
+
+
+

+ On-Chain +

+ 
+                        {JSON.stringify(diff.onChain, null, 2) ?? "null"}
+
+
+
+
+ ))} +
+
+ )} + + {history.length > 1 && ( +
+
+ Verification History +
+
+ {[...history].reverse().map((record, i) => ( +
+ + + {new Date(record.verifiedAt).toLocaleString()} + + {record.previousStatus && ( + + (was ) + + )} +
+ ))} +
+
+ )} + + {spec.spec.functions.length > 0 && ( +
+
+ Functions ({spec.spec.functions.length}) +
+
+ {spec.spec.functions.map((fn, i) => ( +
+

+ {fn.name} +

+

+ Params: {fn.params.map((p) => `${p.name}: ${JSON.stringify(p.type)}`).join(", ") || "none"} +
+ Returns: {JSON.stringify(fn.returns)} +

+
+ ))} +
+
+ )} + + {spec.spec.events.length > 0 && ( +
+
+ Events ({spec.spec.events.length}) +
+
+ {spec.spec.events.map((ev, i) => ( +
+

+ {ev.name} +

+

+ Topics: {ev.topics.map((t) => `${t.name}: ${JSON.stringify(t.type)}`).join(", ") || "none"} +
+ Data: {ev.data.map((d) => `${d.name}: ${JSON.stringify(d.type)}`).join(", ") || "none"} +

+
+ ))} +
+
+ )} +
+
+ ); +} diff --git a/apps/web/app/registry/page.tsx b/apps/web/app/registry/page.tsx new file mode 100644 index 00000000..b2ed0108 --- /dev/null +++ b/apps/web/app/registry/page.tsx @@ -0,0 +1,178 @@ +import Link from "next/link"; +import { getSpecStore, getVerdictStore } from "@/lib/registry"; + +export const dynamic = "force-dynamic"; + +function StatusBadge({ status }: { status: string | undefined }) { + if (!status) { + return ( + + UNVERIFIED + + ); + } + + const colors: Record = { + verified: { bg: "#0a2a0a", border: "#1a4a1a", text: "#4ade80" }, + mismatch: { bg: "#2a0a0a", border: "#4a1a1a", text: "#ff5370" }, + unverifiable: { bg: "#2a2a00", border: "#4a4a00", text: "#facc15" }, + }; + + const c = colors[status] ?? { bg: "var(--surface2)", border: "var(--border)", text: "var(--muted)" }; + + return ( + + {status.toUpperCase()} + + ); +} + +export default async function RegistryPage() { + const specs = await getSpecStore().getAll(); + const verdicts = await getVerdictStore().getAll(); + const verdictMap = new Map(verdicts.map((v) => [v.contractId, v])); + + return ( +
+
+

+ ABI Registry Explorer +

+

+ Every registered Soroban spec with on-chain verification status. + Mismatched specs are flagged and automatically reported. +

+ +
+
+
Contract
+
Status
+
Age
+
+ + {specs.length === 0 && ( +
+ No registered specs yet. +
+ )} + + {specs.map((spec) => { + const verdict = verdictMap.get(spec.contractId); + const verifiedAt = verdict?.verifiedAt + ? `${Math.round((Date.now() - new Date(verdict.verifiedAt).getTime()) / 60000)}m ago` + : "—"; + + return ( + +
+ {spec.spec.name} + + {spec.contractId.slice(0, 12)}… + +
+
+ +
+
+ {verifiedAt} +
+ + ); + })} +
+
+
+ ); +} diff --git a/apps/web/lib/registry.ts b/apps/web/lib/registry.ts new file mode 100644 index 00000000..45b0d846 --- /dev/null +++ b/apps/web/lib/registry.ts @@ -0,0 +1,135 @@ +import { + InMemoryVerdictStore, + InMemorySpecStore, + GitHubIssueReporter, + ConsoleAlertManager, + NoopIssueReporter, + NoopAlertManager, + runVerificationJob, +} from "@orbital-stellar/abi-registry"; +import type { RegisteredSpec } from "@orbital-stellar/abi-registry"; + +const g = globalThis as unknown as { + __orbitalVerdictStore?: InMemoryVerdictStore; + __orbitalSpecStore?: InMemorySpecStore; + __orbitalIssueReporter?: GitHubIssueReporter | NoopIssueReporter; + __orbitalAlertManager?: ConsoleAlertManager | NoopAlertManager; +}; + +export function getVerdictStore(): InMemoryVerdictStore { + if (!g.__orbitalVerdictStore) { + g.__orbitalVerdictStore = new InMemoryVerdictStore(); + } + return g.__orbitalVerdictStore; +} + +export function getSpecStore(): InMemorySpecStore { + if (!g.__orbitalSpecStore) { + g.__orbitalSpecStore = new InMemorySpecStore(); + } + return g.__orbitalSpecStore; +} + +export function getIssueReporter(): GitHubIssueReporter | NoopIssueReporter { + if (!g.__orbitalIssueReporter) { + const token = process.env.GITHUB_TOKEN; + const repo = process.env.GITHUB_REPO; + if (token && repo) { + g.__orbitalIssueReporter = new GitHubIssueReporter(token, repo); + } else { + g.__orbitalIssueReporter = new NoopIssueReporter(); + } + } + return g.__orbitalIssueReporter; +} + +export function getAlertManager(): ConsoleAlertManager | NoopAlertManager { + if (!g.__orbitalAlertManager) { + g.__orbitalAlertManager = + process.env.NODE_ENV === "production" + ? new ConsoleAlertManager() + : new NoopAlertManager(); + } + return g.__orbitalAlertManager; +} + +export async function registerSeedSpecs(): Promise { + const store = getSpecStore(); + const existing = await store.getAll(); + if (existing.length > 0) return; + + const seedSpecs: RegisteredSpec[] = [ + { + contractId: "CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75", + spec: { + version: "1.0.0", + name: "USDC", + contractId: "CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75", + network: "mainnet", + functions: [], + events: [], + types: {}, + }, + submittedAt: new Date().toISOString(), + }, + { + contractId: "CDTKPWPLOURQA2SGTKTUQOWRCBZEORB4BWBOMJ3D3ZTQQSGE5F6JBQLV", + spec: { + version: "1.0.0", + name: "EURC", + contractId: "CDTKPWPLOURQA2SGTKTUQOWRCBZEORB4BWBOMJ3D3ZTQQSGE5F6JBQLV", + network: "mainnet", + functions: [], + events: [], + types: {}, + }, + submittedAt: new Date().toISOString(), + }, + { + contractId: "CAUIKL3IYGMERDRUN5QQVPKPLZTRNVXV27LFCWQIRNOHSNGB3ZXAEFBX", + spec: { + version: "1.0.0", + name: "AQUA", + contractId: "CAUIKL3IYGMERDRUN5QQVPKPLZTRNVXV27LFCWQIRNOHSNGB3ZXAEFBX", + network: "mainnet", + functions: [], + events: [], + types: {}, + }, + submittedAt: new Date().toISOString(), + }, + ]; + + for (const spec of seedSpecs) { + await store.register(spec); + } +} + +export async function runVerification(): Promise< + Awaited> +> { + await registerSeedSpecs(); + + const storedSpecs = await getSpecStore().getAll(); + if (storedSpecs.length === 0) { + return { + total: 0, + verified: 0, + mismatch: 0, + unverifiable: 0, + errors: [], + issuesCreated: [], + }; + } + + return runVerificationJob({ + specStore: getSpecStore(), + verdictStore: getVerdictStore(), + verifyOptions: { + rpcUrl: process.env.ORBITAL_RPC_URL ?? "https://soroban-testnet.stellar.org", + network: (process.env.ORBITAL_NETWORK as "mainnet" | "testnet" | undefined) ?? "testnet", + }, + issueReporter: getIssueReporter(), + alertManager: getAlertManager(), + }); +} diff --git a/apps/web/package.json b/apps/web/package.json index 717f16ef..3dcdfef1 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -3,14 +3,15 @@ "version": "0.1.0", "private": true, "scripts": { - "predev": "pnpm --filter \"@orbital-stellar/pulse-core...\" --filter \"@orbital-stellar/pulse-webhooks...\" run build && pnpm -w run docs:reference", + "predev": "pnpm --filter \"@orbital-stellar/pulse-core...\" --filter \"@orbital-stellar/pulse-webhooks...\" --filter \"@orbital-stellar/abi-registry...\" run build && pnpm -w run docs:reference", "dev": "next dev", - "prebuild": "pnpm --filter \"@orbital-stellar/pulse-core...\" --filter \"@orbital-stellar/pulse-webhooks...\" run build && pnpm -w run docs:reference", + "prebuild": "pnpm --filter \"@orbital-stellar/pulse-core...\" --filter \"@orbital-stellar/pulse-webhooks...\" --filter \"@orbital-stellar/abi-registry...\" run build && pnpm -w run docs:reference", "build": "next build", "start": "next start", "lint": "next lint" }, "dependencies": { + "@orbital-stellar/abi-registry": "workspace:*", "@orbital-stellar/pulse-core": "workspace:*", "@orbital-stellar/pulse-webhooks": "workspace:*", "@stellar/stellar-sdk": "^16.1.0", diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index cd3dcee9..76ba540d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -254,6 +254,25 @@ For `engine.cursor_expired` notifications, the payload includes: Consumers can subscribe to these via `watcher.on("engine.reconnecting", …)` to surface UI banners or write structured logs. +### In-process backpressure and bounded queues + +To protect consumers from unbounded memory growth when a watcher is slow or +there's a burst of ledger activity, `EventEngine` now maintains a bounded +internal queue. Configuration lives in `CoreConfig.queue` and exposes three +knobs: + +- **`highWaterMark`**: the maximum queued events before backpressure triggers (default: 10000). +- **`lowWaterMark`**: the level below which backpressure is considered cleared (default: 50% of highWaterMark). +- **`policy`**: one of `pause` (default), `drop-oldest`, or `drop-newest`. + +When the high-water mark is crossed the engine emits `engine.backpressure` +with `{ active: true, queued, policy }`. The default `pause` policy stops +the underlying sources (Horizon and Soroban) until the queue drains below the +low-water mark, at which point `engine.backpressure` with `{ active: false }` +is emitted and sources are resumed. `drop-oldest` and `drop-newest` shed +events deterministically instead of pausing the source; these are useful for +best-effort dashboards where availability is preferred over perfect delivery. + --- ## 6. Webhook delivery internals diff --git a/packages/abi-registry/src/alertManager.ts b/packages/abi-registry/src/alertManager.ts new file mode 100644 index 00000000..6d451ef5 --- /dev/null +++ b/packages/abi-registry/src/alertManager.ts @@ -0,0 +1,20 @@ +import type { VerdictRecord } from "./verdictStore.js"; + +export interface AlertManager { + alertTransition(previous: VerdictRecord, current: VerdictRecord): Promise; +} + +export class ConsoleAlertManager implements AlertManager { + async alertTransition(previous: VerdictRecord, current: VerdictRecord): Promise { + console.error( + `[ABI Registry Alert] Contract ${current.contractId}: status transitioned from "${previous.status}" to "${current.status}" at ${current.verifiedAt}`, + ); + if (current.diffs && current.diffs.length > 0) { + console.error(`[ABI Registry Alert] Diffs:`, JSON.stringify(current.diffs, null, 2)); + } + } +} + +export class NoopAlertManager implements AlertManager { + async alertTransition(_previous: VerdictRecord, _current: VerdictRecord): Promise {} +} diff --git a/packages/abi-registry/src/index.ts b/packages/abi-registry/src/index.ts index 5ce38df5..9afe8d50 100644 --- a/packages/abi-registry/src/index.ts +++ b/packages/abi-registry/src/index.ts @@ -81,10 +81,17 @@ export type { VerifySchemaOptions, } from "./verifySchema.js"; -export { LabelResolver } from "./LabelResolver.js"; -export type { - LabelRecord, - LabelResolverConfig, - ResolvedLabel, - EntityType, -} from "./LabelResolver.js"; +export { InMemoryVerdictStore } from "./verdictStore.js"; +export type { VerdictStore, VerdictRecord, VerdictStatus } from "./verdictStore.js"; + +export { InMemorySpecStore } from "./specStore.js"; +export type { SpecStore, RegisteredSpec } from "./specStore.js"; + +export { GitHubIssueReporter, NoopIssueReporter } from "./issueReporter.js"; +export type { IssueReporter, MismatchReportParams } from "./issueReporter.js"; + +export { ConsoleAlertManager, NoopAlertManager } from "./alertManager.js"; +export type { AlertManager } from "./alertManager.js"; + +export { runVerificationJob } from "./verificationJob.js"; +export type { VerificationJobConfig, JobResult } from "./verificationJob.js"; diff --git a/packages/abi-registry/src/issueReporter.ts b/packages/abi-registry/src/issueReporter.ts new file mode 100644 index 00000000..01884efa --- /dev/null +++ b/packages/abi-registry/src/issueReporter.ts @@ -0,0 +1,76 @@ +export interface IssueReporter { + reportMismatch(params: MismatchReportParams): Promise; +} + +export type MismatchReportParams = { + contractId: string; + contractName: string; + diffs: { path: string; submitted: unknown; onChain: unknown }[]; + submittedVersion?: string; + previousStatus?: string; +}; + +export class GitHubIssueReporter implements IssueReporter { + constructor( + private readonly token: string, + private readonly repo: string, + ) {} + + async reportMismatch(params: MismatchReportParams): Promise { + const title = `[ABI Registry] Schema mismatch detected for ${params.contractName} (${params.contractId.slice(0, 8)}…)`; + const body = [ + `## ABI Schema Mismatch`, + ``, + `Contract **${params.contractName}** (\`${params.contractId}\`) has a submitted schema that does not match its on-chain contract spec.`, + ``, + `| Field | Detail |`, + `|---|---|`, + `| **Contract** | \`${params.contractId}\` |`, + `| **Name** | ${params.contractName} |`, + params.submittedVersion ? `| **Submitted version** | ${params.submittedVersion} |` : null, + params.previousStatus ? `| **Previous verdict** | ${params.previousStatus} |` : null, + ``, + `### Differences`, + ``, + params.diffs.length === 0 + ? `No specific field diffs recorded.` + : `| # | Path | Submitted | On-chain |`, + `|---|---|---|---|`, + ].concat( + params.diffs.map( + (d, i) => + `| ${i + 1} | \`${d.path}\` | \`${JSON.stringify(d.submitted)}\` | \`${JSON.stringify(d.onChain)}\` |`, + ), + ); + + body.push(``, `---`, `_Automatically reported by ABI Registry Verification Pipeline_`); + + const response = await fetch(`https://api.github.com/repos/${this.repo}/issues`, { + method: "POST", + headers: { + Authorization: `Bearer ${this.token}`, + "Content-Type": "application/json", + Accept: "application/vnd.github.v3+json", + }, + body: JSON.stringify({ + title, + body: body.filter(Boolean).join("\n"), + labels: ["abi-registry", "schema-mismatch", "automated"], + }), + }); + + if (!response.ok) { + const text = await response.text().catch(() => "unknown"); + throw new Error(`GitHubIssueReporter: failed to create issue (${response.status}): ${text}`); + } + + const data = (await response.json()) as { html_url?: string }; + return data.html_url ?? `https://github.com/${this.repo}/issues`; + } +} + +export class NoopIssueReporter implements IssueReporter { + async reportMismatch(_params: MismatchReportParams): Promise { + return ""; + } +} diff --git a/packages/abi-registry/src/specStore.ts b/packages/abi-registry/src/specStore.ts new file mode 100644 index 00000000..f5ba9945 --- /dev/null +++ b/packages/abi-registry/src/specStore.ts @@ -0,0 +1,36 @@ +import type { ContractSpec } from "./spec.js"; + +export type RegisteredSpec = { + contractId: string; + spec: ContractSpec; + publisher?: string; + submittedAt: string; + attestedBy?: string[]; +}; + +export interface SpecStore { + register(spec: RegisteredSpec): Promise; + get(contractId: string): Promise; + getAll(): Promise; + remove(contractId: string): Promise; +} + +export class InMemorySpecStore implements SpecStore { + private readonly specs: Map = new Map(); + + async register(spec: RegisteredSpec): Promise { + this.specs.set(spec.contractId, spec); + } + + async get(contractId: string): Promise { + return this.specs.get(contractId) ?? null; + } + + async getAll(): Promise { + return Array.from(this.specs.values()); + } + + async remove(contractId: string): Promise { + this.specs.delete(contractId); + } +} diff --git a/packages/abi-registry/src/verdictStore.ts b/packages/abi-registry/src/verdictStore.ts new file mode 100644 index 00000000..d02ab58a --- /dev/null +++ b/packages/abi-registry/src/verdictStore.ts @@ -0,0 +1,50 @@ +import type { SchemaFieldDiff } from "./verifySchema.js"; + +export type VerdictStatus = "verified" | "mismatch" | "unverifiable"; + +export type VerdictRecord = { + contractId: string; + status: VerdictStatus; + verifiedAt: string; + previousStatus?: VerdictStatus; + diffs?: SchemaFieldDiff[]; + reason?: string; + attestedBy?: string[]; + specVersion?: string; +}; + +export interface VerdictStore { + record(verdict: VerdictRecord): Promise; + getLatest(contractId: string): Promise; + getAll(): Promise; + getHistory(contractId: string): Promise; +} + +export class InMemoryVerdictStore implements VerdictStore { + private readonly records: Map = new Map(); + + async record(verdict: VerdictRecord): Promise { + const existing = this.records.get(verdict.contractId) ?? []; + existing.push(verdict); + this.records.set(verdict.contractId, existing); + } + + async getLatest(contractId: string): Promise { + const existing = this.records.get(contractId); + if (!existing || existing.length === 0) return null; + return existing[existing.length - 1]!; + } + + async getAll(): Promise { + const all: VerdictRecord[] = []; + for (const records of this.records.values()) { + const latest = records[records.length - 1]; + if (latest) all.push(latest); + } + return all; + } + + async getHistory(contractId: string): Promise { + return this.records.get(contractId) ?? []; + } +} diff --git a/packages/abi-registry/src/verificationJob.ts b/packages/abi-registry/src/verificationJob.ts new file mode 100644 index 00000000..ca393f1b --- /dev/null +++ b/packages/abi-registry/src/verificationJob.ts @@ -0,0 +1,120 @@ +import type { ContractSpec } from "./spec.js"; +import { verifySchema, type SchemaVerdict, type VerifySchemaOptions } from "./verifySchema.js"; +import type { VerdictStore, VerdictRecord } from "./verdictStore.js"; +import type { SpecStore, RegisteredSpec } from "./specStore.js"; +import type { IssueReporter } from "./issueReporter.js"; +import type { AlertManager } from "./alertManager.js"; + +export type VerificationJobConfig = { + specStore: SpecStore; + verdictStore: VerdictStore; + verifyOptions: VerifySchemaOptions; + issueReporter?: IssueReporter; + alertManager?: AlertManager; + onVerdict?: (record: VerdictRecord) => void; +}; + +export type JobResult = { + total: number; + verified: number; + mismatch: number; + unverifiable: number; + errors: { contractId: string; error: string }[]; + issuesCreated: string[]; +}; + +function verdictToStatus(verdict: SchemaVerdict): VerdictRecord["status"] { + switch (verdict.status) { + case "match": + return "verified"; + case "mismatch": + return "mismatch"; + case "unverifiable": + return "unverifiable"; + } +} + +export async function runVerificationJob(config: VerificationJobConfig): Promise { + const result: JobResult = { + total: 0, + verified: 0, + mismatch: 0, + unverifiable: 0, + errors: [], + issuesCreated: [], + }; + + const specs = await config.specStore.getAll(); + result.total = specs.length; + + for (const registered of specs) { + try { + const record = await verifySingleContract(registered, config); + if (record.status === "verified") result.verified++; + else if (record.status === "mismatch") result.mismatch++; + else if (record.status === "unverifiable") result.unverifiable++; + + config.onVerdict?.(record); + + if (record.status === "mismatch" && config.issueReporter) { + const issueUrl = await config.issueReporter.reportMismatch({ + contractId: record.contractId, + contractName: registered.spec.name, + diffs: record.diffs ?? [], + submittedVersion: registered.spec.version, + previousStatus: record.previousStatus, + }); + if (issueUrl) result.issuesCreated.push(issueUrl); + } + + if (record.previousStatus && record.previousStatus !== record.status && config.alertManager) { + const prevRecord: VerdictRecord = { + contractId: record.contractId, + status: record.previousStatus, + verifiedAt: "", + }; + await config.alertManager.alertTransition(prevRecord, record); + } + } catch (error) { + result.errors.push({ + contractId: registered.contractId, + error: error instanceof Error ? error.message : String(error), + }); + } + } + + return result; +} + +async function verifySingleContract( + registered: RegisteredSpec, + config: VerificationJobConfig, +): Promise { + const previous = await config.verdictStore.getLatest(registered.contractId); + + const submittedSpec: ContractSpec = { + ...registered.spec, + contractId: registered.contractId, + }; + + const verdict = await verifySchema(registered.contractId, submittedSpec, config.verifyOptions); + + const record: VerdictRecord = { + contractId: registered.contractId, + status: verdictToStatus(verdict), + verifiedAt: new Date().toISOString(), + previousStatus: previous?.status, + specVersion: registered.spec.version, + }; + + if (verdict.status === "mismatch") { + record.diffs = verdict.diffs; + } + if (verdict.status === "unverifiable") { + record.reason = verdict.reason; + record.attestedBy = registered.attestedBy; + } + + await config.verdictStore.record(record); + return record; +} diff --git a/packages/pulse-core/src/EventEngine.ts b/packages/pulse-core/src/EventEngine.ts index ca74a324..81471706 100644 --- a/packages/pulse-core/src/EventEngine.ts +++ b/packages/pulse-core/src/EventEngine.ts @@ -206,6 +206,14 @@ export class EventEngine { private consecutiveCursorFailures = 0; private isCursorStoreUnhealthy = false; private pausedSources = new Set<"horizon" | "soroban">(); + /** Internal bounded queue for normalized events to protect slow consumers. */ + private eventQueue: Array> = []; + private queueHighWaterMark: number; + private queueLowWaterMark: number; + private queuePolicy: "pause" | "drop-oldest" | "drop-newest"; + private processingQueue = false; + private inBackpressure = false; + private wePausedSourcesForBackpressure = false; /** * Optional live Soroban subscriber. Wired only when the engine is configured * for live contract streaming; otherwise undefined, and the guarded calls @@ -287,6 +295,14 @@ export class EventEngine { this.abiRegistry = EventEngine.resolveAbiRegistry(config.abiRegistry); this.cursorStore = config.cursorStore; this.network = config.network; + // Queue configuration + const defaultHigh = 10000; + this.queueHighWaterMark = Math.max(1, Math.floor(config.queue?.highWaterMark ?? defaultHigh)); + this.queueLowWaterMark = Math.max( + 1, + Math.floor(config.queue?.lowWaterMark ?? Math.floor(this.queueHighWaterMark / 2)), + ); + this.queuePolicy = config.queue?.policy ?? "pause"; if (config.soroban) { const rpc = new SorobanRpcClient({ @@ -1152,7 +1168,7 @@ export class EventEngine { } this.lastEventAt = event.timestamp; - this.route(event); + this.enqueueEvent(event); }, onerror: (error) => { this.log.error("[pulse-core] SSE error.", { error }); @@ -1176,6 +1192,110 @@ export class EventEngine { }); } + private enqueueEvent(event: Timestamped): void { + // If over capacity, apply configured policy. + if (this.eventQueue.length >= this.queueHighWaterMark) { + if (this.queuePolicy === "pause") { + if (!this.inBackpressure) { + this.inBackpressure = true; + // Pause sources to prevent further incoming events. + try { + this.pauseSource("horizon"); + this.pauseSource("soroban"); + this.wePausedSourcesForBackpressure = true; + } catch (err) { + /* swallow - pauseSource logs */ + } + this.notifyWatchers("engine.backpressure", { + type: "engine.backpressure", + attempt: 0, + emittedAt: new Date().toISOString(), + active: true, + queued: this.eventQueue.length, + policy: this.queuePolicy, + } as any); + } + // Still accept the event so in-flight sources pause and we can drain. + this.eventQueue.push(event); + } else if (this.queuePolicy === "drop-oldest") { + // Drop one oldest then push + this.eventQueue.shift(); + this.eventQueue.push(event); + if (!this.inBackpressure) { + this.inBackpressure = true; + this.notifyWatchers("engine.backpressure", { + type: "engine.backpressure", + attempt: 0, + emittedAt: new Date().toISOString(), + active: true, + queued: this.eventQueue.length, + policy: this.queuePolicy, + } as any); + } + } else { + // drop-newest: ignore incoming + if (!this.inBackpressure) { + this.inBackpressure = true; + this.notifyWatchers("engine.backpressure", { + type: "engine.backpressure", + attempt: 0, + emittedAt: new Date().toISOString(), + active: true, + queued: this.eventQueue.length, + policy: this.queuePolicy, + } as any); + } + return; + } + } else { + this.eventQueue.push(event); + } + + // Kick the async processor + this.processQueue().catch((err) => { + this.log.error("[pulse-core] event queue processor failed", { error: err }); + }); + } + + private async processQueue(): Promise { + if (this.processingQueue) return; + this.processingQueue = true; + try { + while (this.eventQueue.length > 0) { + const ev = this.eventQueue.shift()!; + try { + this.route(ev); + } catch (err) { + this.log.warn("[pulse-core] watcher handler threw while routing event", { error: err }); + } + + // Clear backpressure if we're below low watermark + if (this.inBackpressure && this.eventQueue.length <= this.queueLowWaterMark) { + this.inBackpressure = false; + if (this.wePausedSourcesForBackpressure) { + try { + this.resumeSource("horizon"); + this.resumeSource("soroban"); + } catch (err) { + /* swallow */ + } + this.wePausedSourcesForBackpressure = false; + } + this.notifyWatchers("engine.backpressure", { + type: "engine.backpressure", + attempt: 0, + emittedAt: new Date().toISOString(), + active: false, + queued: this.eventQueue.length, + policy: this.queuePolicy, + } as any); + } + } + } finally { + this.processingQueue = false; + } + } + private getHorizonCursor(record: unknown): string | null { if (!this.isRecord(record)) { return null; diff --git a/packages/pulse-core/src/index.ts b/packages/pulse-core/src/index.ts index da7a78a6..31a53010 100644 --- a/packages/pulse-core/src/index.ts +++ b/packages/pulse-core/src/index.ts @@ -142,7 +142,8 @@ export type WatcherNotificationType = | "engine.rate_limited" | "engine.stopped" | "engine.cursor_store_unhealthy" - | "engine.cursor_expired"; + | "engine.cursor_expired" + | "engine.backpressure"; export type OfferEventType = "offer.created" | "offer.updated" | "offer.deleted"; export type BumpSequenceEventType = "account.bump_sequence"; @@ -455,6 +456,12 @@ export type WatcherNotification = { cursor?: string; /** The source that triggered this notification. */ source?: "horizon" | "soroban"; + /** Backpressure active flag (for `engine.backpressure`). */ + active?: boolean; + /** Number of events currently queued inside the engine. */ + queued?: number; + /** Queue policy in effect when backpressure was emitted. */ + policy?: string; /** ISO 8601 timestamp of when this notification was emitted. */ emittedAt: string; /** The cursor value that was expired or lost, if applicable. */ @@ -565,6 +572,15 @@ export type CoreConfig = { abiRegistry?: AbiRegistryClientLike | false; /** Soroban RPC configuration. Ignored when `network` is an array - set `soroban` per source instead. */ soroban?: SorobanConfig; + /** Optional internal event queue tuning. */ + queue?: { + /** High water mark for the internal engine queue. Defaults to 10000. */ + highWaterMark?: number; + /** Low water mark at which backpressure is considered cleared. Defaults to 50% of highWaterMark. */ + lowWaterMark?: number; + /** Backpressure policy: 'pause' | 'drop-oldest' | 'drop-newest'. Defaults to 'pause'. */ + policy?: "pause" | "drop-oldest" | "drop-newest"; + }; }; // Error class for invalid network validation diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 34c916f5..189d1cdd 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -43,6 +43,9 @@ importers: apps/web: dependencies: + '@orbital-stellar/abi-registry': + specifier: workspace:* + version: link:../../packages/abi-registry '@orbital-stellar/pulse-core': specifier: workspace:* version: link:../../packages/pulse-core