Skip to content

Commit a10601b

Browse files
committed
chore: fix test-race + packages/subgraphs/README
1 parent ab3b35f commit a10601b

8 files changed

Lines changed: 126 additions & 20 deletions

File tree

packages/shared/src/db/queries/subscriptions.test.ts

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { afterAll, beforeAll, beforeEach, describe, expect, it } from "bun:test";
22
import { randomUUID } from "node:crypto";
3-
import { closeDb, getDb } from "../index.ts";
3+
import { getDb } from "../index.ts";
44
import {
55
createSubscription,
66
deleteSubscription,
@@ -32,8 +32,9 @@ beforeAll(async () => {
3232
});
3333

3434
afterAll(async () => {
35+
// Don't closeDb() — destroying the shared singleton breaks sibling
36+
// test files when `bun test` runs them together.
3537
await db.deleteFrom("subscriptions").where("account_id", "=", accountId).execute();
36-
await closeDb();
3738
});
3839

3940
beforeEach(async () => {

packages/subgraphs/README.md

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
# @secondlayer/subgraphs
2+
3+
Typed on-chain indexing for Stacks. Declare event filters + column schema with `defineSubgraph()`; the runtime decodes blocks, matches filters, runs your handlers inside a transactional context, and exposes the result as a Postgres schema you query over REST or SQL.
4+
5+
Subgraph rows fan out to HTTP subscribers through a post-flush outbox emitter — signed Standard Webhooks POSTs with retries, circuit breaker, and replay.
6+
7+
## Install
8+
9+
```bash
10+
bun add @secondlayer/subgraphs
11+
```
12+
13+
## Quick Start
14+
15+
```typescript
16+
import { defineSubgraph } from "@secondlayer/subgraphs";
17+
18+
export default defineSubgraph({
19+
name: "token-transfers",
20+
version: "1.0.0",
21+
sources: {
22+
// Named event sources — the key becomes the handler name.
23+
transfer: {
24+
type: "ft_transfer",
25+
assetIdentifier: "SP2X0TZ59D5SZ8ACQ6YMCHHNR2ZN51Z32E2CJ173.stx-token::stx",
26+
},
27+
},
28+
schema: {
29+
transfers: {
30+
columns: {
31+
sender: { type: "principal" },
32+
recipient: { type: "principal" },
33+
amount: { type: "uint" },
34+
},
35+
// Auto-added: _block_height, _tx_id, _created_at
36+
},
37+
},
38+
handlers: {
39+
async transfer(event, ctx) {
40+
ctx.insert("transfers", {
41+
sender: event.sender,
42+
recipient: event.recipient,
43+
amount: event.amount,
44+
});
45+
},
46+
},
47+
});
48+
```
49+
50+
Deploy via CLI (`sl subgraphs deploy path/to/definition.ts`), SDK (`sl.subgraphs.deploy({...})`), or MCP (`subgraphs_deploy`). The dashboard is read-only — creation always happens through an API surface.
51+
52+
## Exports
53+
54+
| Subpath | Description |
55+
| --- | --- |
56+
| `.` | `defineSubgraph`, `validateSubgraphDefinition`, `deploySchema`, `diffSchema`, `reindexSubgraph`, `backfillSubgraph`, `generateSubgraphSQL`, `pgSchemaName` |
57+
| `./types` | All schema + filter + handler types (`SubgraphDefinition`, `SubgraphFilter`, `StxTransferFilter`, etc.) |
58+
| `./schema` | Generator + deployer internals |
59+
| `./validate` | Shape + filter validation for deploys |
60+
| `./runtime/source-matcher` | Pure fn: match txs+events against a `SubgraphFilter` — used by the processor hot path |
61+
| `./runtime/replay` | `replaySubscription({ accountId, subscriptionId, fromBlock, toBlock })` — re-enqueue historical rows as outbox entries |
62+
63+
## Runtime components
64+
65+
The runtime ships behind these entrypoints (import from the package root):
66+
67+
- `startSubgraphProcessor(opts?)` — boots the block processor. LISTENs on `indexer:new_block`, matches sources, runs handlers, flushes writes inside a transaction, and emits outbox rows for matching subscriptions. Also boots the emitter worker.
68+
- `processBlock(subgraph, name, height, opts?)` — single-block entry point used by catch-up, reindex, and tests.
69+
- `catchUpSubgraph(def, name)` — drains pending blocks up to chain tip.
70+
- `reindexSubgraph(def, opts)` — drop + rebuild schema tables from a start block. Breaking schema changes trigger this automatically on deploy.
71+
72+
## Subscription emitter
73+
74+
Every row written through `ctx.insert()` / `ctx.upsert()` is atomically enqueued to `subscription_outbox` for every active subscription whose filter matches — inside the same transaction as the flush, so a processor crash rolls back both.
75+
76+
The emitter drains the outbox via `LISTEN subscriptions:new_outbox` and `FOR UPDATE SKIP LOCKED` batch claims. Live deliveries win a 90/10 split over replays. Each row dispatches through the format builder matching the subscription's `format` column (`standard-webhooks`, `inngest`, `trigger`, `cloudflare`, `cloudevents`, `raw`). Retries follow `30s → 2m → 10m → 1h → 6h → 24h → 72h`. Twenty consecutive failures trips the per-sub circuit breaker and pauses the subscription.
77+
78+
Delivery bodies and response previews land in `subscription_deliveries`. Rows whose retries exhaust mark `status = 'dead'` in the outbox and surface in the dashboard's dead-letter queue for one-click requeue.
79+
80+
## Environment
81+
82+
| Variable | Default | Description |
83+
| --- | --- | --- |
84+
| `SECONDLAYER_EMIT_OUTBOX` | `true` | Set `false` to bypass outbox emission on every block (kill-switch). |
85+
| `SECONDLAYER_ALLOW_PRIVATE_EGRESS` | `false` | Allow the emitter to deliver to private IP ranges (localhost, 10/8, 172.16/12, 192.168/16, link-local, v6 mapped). Leave off in production. |
86+
| `SECONDLAYER_SECRETS_KEY` || 32-byte hex key for the AES-GCM envelope around subscription signing secrets. OSS mode auto-generates + persists to `.env.local`. |
87+
88+
## Postgres + pool mode
89+
90+
The emitter holds a persistent `LISTEN` on `subscriptions:new_outbox` and `subscriptions:changed`, so it MUST connect through a session-mode pool. pgbouncer in transaction mode silently breaks it. Run the emitter against a session-mode port (`pool_mode = session`), or connect directly to Postgres as the default docker-compose setup does.
91+
92+
## License
93+
94+
MIT

packages/subgraphs/src/runtime/context.test.ts

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { beforeAll, describe, expect, it } from "bun:test";
22
import { randomUUID } from "node:crypto";
3-
import { closeDb, getDb, sql } from "@secondlayer/shared/db";
3+
import { getDb, sql } from "@secondlayer/shared/db";
44
import type { Kysely } from "kysely";
55
import type { Database } from "@secondlayer/shared/db";
66
import type { SubgraphSchema } from "../types.ts";
@@ -90,7 +90,5 @@ describe("SubgraphContext flush manifest", () => {
9090

9191
expect(manifest.writes[0].row.sender).toBe("SP1");
9292
expect(manifest.writes[2].row.recipient).toBe("SP1");
93-
94-
await closeDb();
9593
});
9694
});

packages/subgraphs/src/runtime/emitter-perf.test.ts

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { afterAll, beforeAll, describe, expect, it } from "bun:test";
22
import { randomUUID } from "node:crypto";
3-
import { closeDb, getDb } from "@secondlayer/shared/db";
3+
import { getDb } from "@secondlayer/shared/db";
44
import { createSubscription } from "@secondlayer/shared/db/queries/subscriptions";
55
import { SubscriptionMatcher } from "./emitter-matcher.ts";
66
import { startEmitter } from "./emitter.ts";
@@ -29,8 +29,11 @@ const db = getDb();
2929
const accountId = randomUUID();
3030
let stopEmitter: (() => Promise<void>) | null = null;
3131

32-
const SUB_COUNT = Number.parseInt(process.env.PERF_SUBS ?? "50", 10);
33-
const BLOCK_COUNT = Number.parseInt(process.env.PERF_BLOCKS ?? "200", 10);
32+
// Defaults chosen so the full `bun test` completes under 60s without
33+
// timing out beforeAll/afterAll hooks. Override for real perf runs:
34+
// PERF_SUBS=50 PERF_BLOCKS=200 bun test emitter-perf
35+
const SUB_COUNT = Number.parseInt(process.env.PERF_SUBS ?? "20", 10);
36+
const BLOCK_COUNT = Number.parseInt(process.env.PERF_BLOCKS ?? "50", 10);
3437

3538
function percentile(values: number[], p: number): number {
3639
if (values.length === 0) return 0;
@@ -47,12 +50,12 @@ beforeAll(async () => {
4750
});
4851

4952
afterAll(async () => {
53+
// Don't closeDb() — see emitter.test.ts comment.
5054
await stopEmitter?.();
5155
await db
5256
.deleteFrom("subscriptions")
5357
.where("account_id", "=", accountId)
5458
.execute();
55-
await closeDb();
5659
});
5760

5861
describe("emitter perf", () => {

packages/subgraphs/src/runtime/emitter.test.ts

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,14 @@
11
import { afterAll, beforeAll, describe, expect, it } from "bun:test";
22
import { randomUUID } from "node:crypto";
3-
import { closeDb, getDb } from "@secondlayer/shared/db";
3+
import { getDb } from "@secondlayer/shared/db";
44
import { createSubscription } from "@secondlayer/shared/db/queries/subscriptions";
55
import { verify } from "@secondlayer/shared/crypto/standard-webhooks";
66
import { startEmitter } from "./emitter.ts";
77

8+
// Per-suite afterAll only cleans up its own data. Never call `closeDb()` —
9+
// it destroys the shared Kysely singleton and breaks sibling test files
10+
// when `bun test` runs them together. Process exit handles pool cleanup.
11+
812
process.env.INSTANCE_MODE = process.env.INSTANCE_MODE ?? "oss";
913
process.env.DATABASE_URL =
1014
process.env.DATABASE_URL ??
@@ -56,7 +60,6 @@ beforeAll(async () => {
5660
afterAll(async () => {
5761
await stopEmitter?.();
5862
await db.deleteFrom("subscriptions").where("account_id", "=", accountId).execute();
59-
await closeDb();
6063
});
6164

6265
describe("startEmitter end-to-end", () => {

packages/subgraphs/src/runtime/failure-modes.test.ts

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
import { afterAll, beforeAll, describe, expect, it } from "bun:test";
22
import { randomUUID } from "node:crypto";
33
import { sign, verify } from "@secondlayer/shared/crypto/standard-webhooks";
4-
import { closeDb, getDb } from "@secondlayer/shared/db";
4+
import { getDb } from "@secondlayer/shared/db";
55
import { createSubscription } from "@secondlayer/shared/db/queries/subscriptions";
66
import { startEmitter } from "./emitter.ts";
77

@@ -20,12 +20,12 @@ beforeAll(async () => {
2020
});
2121

2222
afterAll(async () => {
23+
// Don't closeDb() — see emitter.test.ts comment.
2324
await stopEmitter?.();
2425
await db
2526
.deleteFrom("subscriptions")
2627
.where("account_id", "=", accountId)
2728
.execute();
28-
await closeDb();
2929
});
3030

3131
describe("failure modes", () => {

packages/subgraphs/src/runtime/outbox-emit.test.ts

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { afterAll, beforeAll, describe, expect, it } from "bun:test";
22
import { randomUUID } from "node:crypto";
3-
import { closeDb, getDb, sql } from "@secondlayer/shared/db";
3+
import { getDb, sql } from "@secondlayer/shared/db";
44
import { createSubscription } from "@secondlayer/shared/db/queries/subscriptions";
55
import type { Kysely } from "kysely";
66
import type { Database } from "@secondlayer/shared/db";
@@ -22,11 +22,11 @@ beforeAll(async () => {
2222
});
2323

2424
afterAll(async () => {
25+
// Don't closeDb() — see emitter.test.ts comment. Per-suite cleanup only.
2526
await db
2627
.deleteFrom("subscriptions")
2728
.where("account_id", "=", accountId)
2829
.execute();
29-
await closeDb();
3030
});
3131

3232
describe("emitSubscriptionOutbox", () => {

packages/subgraphs/src/runtime/ssrf.test.ts

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { afterAll, beforeAll, describe, expect, it } from "bun:test";
22
import { randomUUID } from "node:crypto";
3-
import { closeDb, getDb } from "@secondlayer/shared/db";
3+
import { getDb } from "@secondlayer/shared/db";
44
import { createSubscription } from "@secondlayer/shared/db/queries/subscriptions";
55
import { startEmitter } from "./emitter.ts";
66

@@ -9,24 +9,31 @@ process.env.DATABASE_URL =
99
process.env.DATABASE_URL ??
1010
"postgresql://postgres:postgres@127.0.0.1:5432/secondlayer";
1111

12-
// This test exercises the refusal path — do NOT set ALLOW_PRIVATE_EGRESS.
13-
delete process.env.SECONDLAYER_ALLOW_PRIVATE_EGRESS;
14-
1512
const db = getDb();
1613
const accountId = randomUUID();
1714
let stopEmitter: (() => Promise<void>) | null = null;
15+
// Scope the env-var mutation to this suite's lifecycle. Sibling test
16+
// files set ALLOW_PRIVATE_EGRESS=true at module load; a bare
17+
// `delete process.env.X` at this file's module load would race them
18+
// (bun loads modules in arbitrary order before running tests).
19+
let priorAllowEnv: string | undefined;
1820

1921
beforeAll(async () => {
22+
priorAllowEnv = process.env.SECONDLAYER_ALLOW_PRIVATE_EGRESS;
23+
delete process.env.SECONDLAYER_ALLOW_PRIVATE_EGRESS;
2024
stopEmitter = await startEmitter({ pollIntervalMs: 500 });
2125
});
2226

2327
afterAll(async () => {
28+
// Don't closeDb() — see emitter.test.ts comment.
2429
await stopEmitter?.();
2530
await db
2631
.deleteFrom("subscriptions")
2732
.where("account_id", "=", accountId)
2833
.execute();
29-
await closeDb();
34+
if (priorAllowEnv !== undefined) {
35+
process.env.SECONDLAYER_ALLOW_PRIVATE_EGRESS = priorAllowEnv;
36+
}
3037
});
3138

3239
describe("SSRF egress guard", () => {

0 commit comments

Comments
 (0)