Skip to content

Commit ce4dbb8

Browse files
lootheroloothero
committed
Harden Starknet subscription API
1 parent 651e47a commit ce4dbb8

10 files changed

Lines changed: 178 additions & 40 deletions

File tree

‎packages/starknet-rpc/README.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,10 @@ It does not use the Apibara DNA runtime as a block cache.
1616
Use Starknet JSON-RPC v0.10 or newer endpoints, for example URLs ending in
1717
`/rpc/v0_10` and `/ws/rpc/v0_10`.
1818

19+
The package targets modern Node.js runtimes with global `fetch` support. It
20+
ships a default WebSocket client for Node environments, and callers can still
21+
provide `webSocketFactory` when they need a custom transport.
22+
1923
This package relies on `block_number`, `transaction_index`, and `event_index`
2024
from emitted events for duplicate-safe cursoring. Older RPC versions do not
2125
include all of these fields. Pre-confirmed live indexing also requires a node

‎packages/starknet-rpc/package.json‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,11 @@
2929
},
3030
"devDependencies": {
3131
"@types/node": "^20.12.13",
32+
"@types/ws": "^8.18.1",
3233
"unbuild": "^2.0.0",
3334
"vitest": "^1.6.0"
35+
},
36+
"dependencies": {
37+
"ws": "^8.21.0"
3438
}
3539
}

‎packages/starknet-rpc/src/constants.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
import type { SubscriptionFinalityStatus } from "./types";
2+
3+
export const DEFAULT_SUBSCRIPTION_FINALITY_STATUS: SubscriptionFinalityStatus =
4+
"PRE_CONFIRMED";

‎packages/starknet-rpc/src/cursor.ts‎

Lines changed: 0 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -14,19 +14,6 @@ export function compareEventCursor(a: EventCursor, b: EventCursor): number {
1414
return a.eventIndex < b.eventIndex ? -1 : 1;
1515
}
1616

17-
const aTransactionHash = normalizeFelt(
18-
a.transactionHash,
19-
"cursor.transactionHash",
20-
);
21-
const bTransactionHash = normalizeFelt(
22-
b.transactionHash,
23-
"cursor.transactionHash",
24-
);
25-
26-
if (aTransactionHash !== bTransactionHash) {
27-
return aTransactionHash < bTransactionHash ? -1 : 1;
28-
}
29-
3017
return 0;
3118
}
3219

‎packages/starknet-rpc/src/index.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,8 @@ export type {
4949
RpcWebSocket,
5050
StreamEventsOptions,
5151
StreamMessage,
52+
SubscriptionBlockId,
53+
SubscriptionFinalityStatus,
5254
SubscribeEventsOptions,
5355
SubscribeReconnectOptions,
5456
WebSocketFactory,

‎packages/starknet-rpc/src/subscribe.ts‎

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,16 @@
1+
import { DEFAULT_SUBSCRIPTION_FINALITY_STATUS } from "./constants";
12
import { compareEventCursor, cursorEquals, eventCursorKey } from "./cursor";
23
import { normalizeFelt } from "./normalize";
34
import type {
4-
BlockId,
55
EventCursor,
66
EventSubscription,
77
Felt,
8-
FinalityStatus,
98
StreamMessage,
109
SubscribeEventsOptions,
10+
SubscriptionBlockId,
1111
} from "./types";
1212
import { TooManyBlocksBackError, connectSubscribeEvents } from "./ws";
1313

14-
const DEFAULT_FINALITY_STATUS: FinalityStatus = "PRE_CONFIRMED";
1514
const DEFAULT_MIN_RECONNECT_DELAY_MS = 500;
1615
const DEFAULT_MAX_RECONNECT_DELAY_MS = 10_000;
1716
const MAX_REMEMBERED_CURSOR_KEYS = 2_048;
@@ -163,11 +162,12 @@ function normalizeSubscribeOptions(
163162
blockId: options.blockId ? normalizeBlockId(options.blockId) : undefined,
164163
addresses: normalizeFelts(options.addresses),
165164
keys: options.keys?.map((keys) => normalizeFelts(keys) ?? []),
166-
finalityStatus: options.finalityStatus ?? DEFAULT_FINALITY_STATUS,
165+
finalityStatus:
166+
options.finalityStatus ?? DEFAULT_SUBSCRIPTION_FINALITY_STATUS,
167167
};
168168
}
169169

170-
function initialBlockId(options: SubscribeEventsOptions): BlockId {
170+
function initialBlockId(options: SubscribeEventsOptions): SubscriptionBlockId {
171171
if (options.blockId) {
172172
return options.blockId;
173173
}
@@ -179,9 +179,15 @@ function initialBlockId(options: SubscribeEventsOptions): BlockId {
179179
return "latest";
180180
}
181181

182-
function normalizeBlockId(blockId: BlockId): BlockId {
182+
function normalizeBlockId(blockId: SubscriptionBlockId): SubscriptionBlockId {
183183
if (typeof blockId === "string") {
184-
return blockId;
184+
if (blockId !== "latest") {
185+
throw new Error(
186+
`Invalid starknet_subscribeEvents blockId tag "${blockId}". Use "latest", block_number, or block_hash.`,
187+
);
188+
}
189+
190+
return "latest";
185191
}
186192

187193
if ("block_hash" in blockId) {

‎packages/starknet-rpc/src/types.ts‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,11 +11,22 @@ export type BlockId =
1111
block_hash: Felt;
1212
};
1313

14+
export type SubscriptionBlockId =
15+
| "latest"
16+
| {
17+
block_number: number;
18+
}
19+
| {
20+
block_hash: Felt;
21+
};
22+
1423
export type FinalityStatus =
1524
| "ACCEPTED_ON_L2"
1625
| "ACCEPTED_ON_L1"
1726
| "PRE_CONFIRMED";
1827

28+
export type SubscriptionFinalityStatus = "ACCEPTED_ON_L2" | "PRE_CONFIRMED";
29+
1930
export interface EventCursor {
2031
blockNumber: number;
2132
transactionIndex: number;
@@ -105,10 +116,10 @@ export interface BackfillEventsOptions extends EventFilter {
105116

106117
export interface SubscribeEventsOptions {
107118
url: string;
108-
blockId?: BlockId;
119+
blockId?: SubscriptionBlockId;
109120
addresses?: Felt[];
110121
keys?: Felt[][];
111-
finalityStatus?: FinalityStatus;
122+
finalityStatus?: SubscriptionFinalityStatus;
112123
cursor?: EventCursor;
113124
reconnect?: boolean | SubscribeReconnectOptions;
114125
signal?: AbortSignal;
@@ -130,7 +141,7 @@ export interface StreamEventsOptions extends Omit<EventFilter, "toBlock"> {
130141
wsUrl: string;
131142
cursor?: EventCursor;
132143
/** Applies to the live WebSocket subscription. Historical backfill uses accepted events. */
133-
finalityStatus?: FinalityStatus;
144+
finalityStatus?: SubscriptionFinalityStatus;
134145
signal?: AbortSignal;
135146
webSocketFactory?: WebSocketFactory;
136147
}

‎packages/starknet-rpc/src/ws.ts‎

Lines changed: 35 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,18 @@
1+
import WebSocketImpl from "ws";
2+
import { DEFAULT_SUBSCRIPTION_FINALITY_STATUS } from "./constants";
13
import { normalizeEvent, normalizeFelt, normalizeReorg } from "./normalize";
24
import type {
3-
BlockId,
45
Felt,
5-
FinalityStatus,
66
RpcEvent,
77
RpcReorg,
88
RpcWebSocket,
99
StreamMessage,
1010
SubscribeEventsOptions,
11+
SubscriptionBlockId,
12+
SubscriptionFinalityStatus,
1113
WebSocketFactory,
1214
} from "./types";
1315

14-
const DEFAULT_FINALITY_STATUS: FinalityStatus = "PRE_CONFIRMED";
1516
const TOO_MANY_BLOCKS_BACK_CODE = 68;
1617
const SUBSCRIBE_METHOD = "starknet_subscribeEvents";
1718
const EVENT_NOTIFICATION = "starknet_subscriptionEvents";
@@ -158,24 +159,22 @@ function createWebSocket(
158159
}
159160
).WebSocket;
160161

161-
if (!WebSocketCtor) {
162-
throw new Error(
163-
"No WebSocket implementation available. Pass webSocketFactory in SubscribeEventsOptions.",
164-
);
165-
}
166-
167-
return new WebSocketCtor(url);
162+
return WebSocketCtor
163+
? new WebSocketCtor(url)
164+
: (new WebSocketImpl(url) as unknown as RpcWebSocket);
168165
}
169166

170167
function buildSubscribeParams(options: SubscribeEventsOptions) {
171168
const params: {
172-
block_id: BlockId;
169+
block_id: SubscriptionBlockId;
173170
from_address?: Felt | Felt[];
174171
keys?: Felt[][];
175-
finality_status: FinalityStatus;
172+
finality_status: SubscriptionFinalityStatus;
176173
} = {
177174
block_id: normalizeBlockId(options.blockId ?? "latest"),
178-
finality_status: options.finalityStatus ?? DEFAULT_FINALITY_STATUS,
175+
finality_status: normalizeFinalityStatus(
176+
options.finalityStatus ?? DEFAULT_SUBSCRIPTION_FINALITY_STATUS,
177+
),
179178
};
180179

181180
const fromAddress = normalizeAddresses(options.addresses);
@@ -192,9 +191,15 @@ function buildSubscribeParams(options: SubscribeEventsOptions) {
192191
return params;
193192
}
194193

195-
function normalizeBlockId(blockId: BlockId): BlockId {
194+
function normalizeBlockId(blockId: SubscriptionBlockId): SubscriptionBlockId {
196195
if (typeof blockId === "string") {
197-
return blockId;
196+
if (blockId !== "latest") {
197+
throw new Error(
198+
`Invalid starknet_subscribeEvents blockId tag "${blockId}". Use "latest", block_number, or block_hash.`,
199+
);
200+
}
201+
202+
return "latest";
198203
}
199204

200205
if ("block_hash" in blockId) {
@@ -204,6 +209,21 @@ function normalizeBlockId(blockId: BlockId): BlockId {
204209
return blockId;
205210
}
206211

212+
function normalizeFinalityStatus(
213+
finalityStatus: SubscriptionFinalityStatus,
214+
): SubscriptionFinalityStatus {
215+
if (
216+
finalityStatus !== "ACCEPTED_ON_L2" &&
217+
finalityStatus !== "PRE_CONFIRMED"
218+
) {
219+
throw new Error(
220+
`Invalid starknet_subscribeEvents finalityStatus "${finalityStatus}". Use "ACCEPTED_ON_L2" or "PRE_CONFIRMED".`,
221+
);
222+
}
223+
224+
return finalityStatus;
225+
}
226+
207227
function normalizeAddresses(addresses?: Felt[]): Felt | Felt[] | undefined {
208228
if (!addresses || addresses.length === 0) {
209229
return undefined;

‎packages/starknet-rpc/tests/starknet-rpc.test.ts‎

Lines changed: 74 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { afterEach, describe, expect, it, vi } from "vitest";
22
import { backfillEvents } from "../src/backfill";
3-
import { getBlockWithTxHashes } from "../src/block-cache";
3+
import { StarknetBlockCache, getBlockWithTxHashes } from "../src/block-cache";
44
import { compareEventCursor, eventCursorKey } from "../src/cursor";
55
import { getEvents } from "../src/http";
66
import { normalizeEvent, normalizeFelt } from "../src/normalize";
@@ -37,7 +37,7 @@ describe("cursor comparison", () => {
3737
).toBeGreaterThan(0);
3838
expect(
3939
compareEventCursor(cursor(2, "0x2", 1, 0), cursor(2, "0x1", 1, 0)),
40-
).toBeGreaterThan(0);
40+
).toBe(0);
4141
});
4242

4343
it("keeps the cursor identity keyed by transaction hash and event index", () => {
@@ -47,6 +47,46 @@ describe("cursor comparison", () => {
4747
});
4848
});
4949

50+
describe("block cache", () => {
51+
it("caches blocks by number and hash and invalidates from a reorg start", async () => {
52+
const requests: unknown[] = [];
53+
mockRpcFetch((request) => {
54+
requests.push(request);
55+
return {
56+
block_hash: "0xabc",
57+
block_number: 10,
58+
timestamp: 100,
59+
transactions: [],
60+
};
61+
});
62+
63+
const cache = new StarknetBlockCache({ url: RPC_URL });
64+
65+
await expect(
66+
cache.getBlockWithTxHashes({ block_number: 10 }),
67+
).resolves.toMatchObject({
68+
block_number: 10,
69+
});
70+
await expect(
71+
cache.getBlockWithTxHashes({ block_number: 10 }),
72+
).resolves.toMatchObject({
73+
block_number: 10,
74+
});
75+
expect(requests).toHaveLength(1);
76+
expect(
77+
cache.getCachedMetadata({ block_hash: normalizeFelt("0xabc") }),
78+
).toMatchObject({
79+
blockNumber: 10,
80+
timestamp: 100,
81+
});
82+
83+
cache.invalidateFrom(10);
84+
85+
await cache.getBlockWithTxHashes({ block_number: 10 });
86+
expect(requests).toHaveLength(2);
87+
});
88+
});
89+
5090
describe("event normalization", () => {
5191
it("requires transaction_index and includes it in the cursor", () => {
5292
expect(normalizeEvent(rawEvent({ transactionIndex: 7 })).cursor).toEqual({
@@ -453,6 +493,38 @@ describe("WebSocket subscriptions", () => {
453493
await expect(next).rejects.toBeInstanceOf(TooManyBlocksBackError);
454494
expect(socket.sent).toHaveLength(1);
455495
});
496+
497+
it("rejects subscription-only block tags before sending", async () => {
498+
const sockets: MockWebSocket[] = [];
499+
const iterator = connectSubscribeEvents({
500+
url: WS_URL,
501+
blockId: "pending",
502+
webSocketFactory: mockWebSocketFactory(sockets),
503+
} as never);
504+
505+
const next = iterator.next();
506+
const socket = await waitForSocket(sockets, 0);
507+
socket.open();
508+
509+
await expect(next).rejects.toThrow(/blockId tag/);
510+
expect(socket.sent).toHaveLength(0);
511+
});
512+
513+
it("rejects unsupported subscription finality statuses before sending", async () => {
514+
const sockets: MockWebSocket[] = [];
515+
const iterator = connectSubscribeEvents({
516+
url: WS_URL,
517+
finalityStatus: "ACCEPTED_ON_L1",
518+
webSocketFactory: mockWebSocketFactory(sockets),
519+
} as never);
520+
521+
const next = iterator.next();
522+
const socket = await waitForSocket(sockets, 0);
523+
socket.open();
524+
525+
await expect(next).rejects.toThrow(/finalityStatus/);
526+
expect(socket.sent).toHaveLength(0);
527+
});
456528
});
457529

458530
describe("combined stream", () => {

0 commit comments

Comments
 (0)