diff --git a/README.md b/README.md index 1b52fdd..254a9a4 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,43 @@ pnpm add @morpho-org/viem-dlc Also available on the [GitHub Package Registry](https://npm.pkg.github.com). +## Observability (optional) + +This library can emit structured events through a logger you provide. The expected +shape is a structural subset of [`loglayer`](https://www.npmjs.com/package/loglayer) +— a `LogLayer` instance satisfies it directly — but `loglayer` is **not** a declared +peer dependency, so it isn't installed transitively and isn't required to typecheck. +Pass any value matching the exported `Logger` interface (`child`, `withContext`, +`withMetadata`, `withError`, `info`, `warn`, `error`, `metadataOnly`). + +```bash +pnpm add loglayer # only if you want to use it as the logger +``` + +If you don't call `withLogging`, the library emits nothing and the dep is irrelevant. + +```ts +import { withLogging } from '@morpho-org/viem-dlc' + +await withLogging(() => client.request({ method: 'eth_getLogs', params: [filter] }), { + logger, // anything satisfying the `Logger` interface, e.g. a LogLayer instance + service: 'indexer', // extra opts become context fields on every event +}) +``` + +Each outermost `client.request` made inside a `withLogging` scope emits a single +`"concluded"` wide event. Transports contribute flat, queryable fields under their +key — e.g. `viem-dlc-failover.succeeded_index`, `viem-dlc-logs-divider.logs_fetched` — +and layers crossed many times per call (e.g. once per chunk under the divider) +accumulate totals there (e.g. `viem-dlc-logs-sieve.logs_dropped`). If a call crosses +several *instances* of the same transport — say, one cache per failover branch — later +instances are suffixed `.1`, `.2`, ... in first-touch order, which is stable for a +given composition. Every layer also stamps a per-instance `crossings` count, so the +event records which transports the call traversed and how many times each. Call-level +fields are `call_id`, `duration_ms`, and `status` (`"ok"` or `"error"`). Failed calls +emit at `error` level with the error attached via `withError`, so hosts that forward +`withError` entries to an error reporter (e.g. Sentry) capture them automatically. + ## Transports ### `deployless` @@ -55,6 +92,15 @@ const result = await call(client, { If `policy.cache` is present, `deployless(...)` ignores it and still behaves as split-only mode. Use `cache(...)` when you want the same marked calls to populate and read from a backing store. +With observability enabled, batching reports `elements_requested` / `elements_fetched`, +`nominal_batches`, `batch_bytes` (sizes of the initial packing, so bisected and continued +chunks are not resampled), and `splits_*` for chunks bisected after a size or timeout error. +Paged lenses get their own fields, since stopping early is normal rather than a failure — +and only paged calls emit them, so their presence marks a paged run: `pages_continued` +(responses that stopped early, each of which may be repacked into several requests), +`pages_waves`, and `elements_missing` (elements the lens declined, plus any single element +that exhausted the frame — the same count carried on `DeploylessPartialResultError.missing`). + ### `cache` All-in-one caching transport for `eth_getLogs` and `eth_call`. Internally composes five layers: @@ -71,7 +117,7 @@ import { LruStore } from '@morpho-org/viem-dlc/stores' const transport = cache(http(rpcUrl), [ { binSize: 10_000, - store: new LruStore(100_000_000), + store: new LruStore({ maxBytes: 100_000_000 }), invalidationStrategy: createSimpleInvalidation(), gasLimit: 30_000_000, }, @@ -121,7 +167,7 @@ import { failover } from '@morpho-org/viem-dlc/transports' import { cache, createSimpleInvalidation } from '@morpho-org/viem-dlc/transports/cache' import { LruStore } from '@morpho-org/viem-dlc/stores' -const store = new LruStore(100_000_000) +const store = new LruStore({ maxBytes: 100_000_000 }) const sharedConfig = { binSize: 10_000, store, invalidationStrategy: createSimpleInvalidation() } const transport = failover([ @@ -239,7 +285,10 @@ const client = createPublicClient({ transport }) ### `rateLimiter` -Token-bucket rate limiting with concurrency limiting and priority scheduling: +Token-bucket rate limiting with concurrency limiting and priority scheduling. When +observability is enabled it reports `queue_wait_ms` (admission wait, summarized over +every crossing in the call), which separates time spent queued behind your own limits +from time spent waiting on the upstream RPC: ```ts import { createPublicClient, http } from 'viem' @@ -319,7 +368,7 @@ front would pin the stale copy for the whole process lifetime): import { HierarchicalStore, LruStore, TtlStore } from '@morpho-org/viem-dlc/stores' const store = new HierarchicalStore( - [new TtlStore(new LruStore(100_000_000), { ttlMs: 60_000 }), remote], + [new TtlStore(new LruStore({ maxBytes: 100_000_000 }), { ttlMs: 60_000 }), remote], { populateOnMiss: true }, ) ``` @@ -580,7 +629,7 @@ Exported from `@morpho-org/viem-dlc/utils`: - `divideBlockRange` / `mergeBlockRanges` / `halveBlockRange` — block range manipulation - `resolveBlockNumber` / `extractRangeFromFilter` / `isInBlockRange` — block number helpers -- `isErrorCausedByBlockRange` — detect RPC "block range too large" errors +- `classifyBlockRangeError` — classify RPC errors as range-related, timeout-like, or neither - `createCoalescingMutex` — per-resource leader/follower batching - `createTokenBucket` / `createRateLimit` — rate limiting primitives - `cyrb64Hash` — fast string hashing diff --git a/package.json b/package.json index 9b4947a..59de825 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@morpho-org/viem-dlc", - "version": "0.0.14", + "version": "0.0.15", "description": "A collection of flexible viem extensions with a focus on intelligent caching.", "license": "MIT", "keywords": [ @@ -94,6 +94,7 @@ "@types/node": "25.2.0", "@upstash/redis": "1.38.1", "@vercel/blob": "^2.6.1", + "loglayer": "^9.0.0", "tsx": "4.23.1", "typescript": "7.0.2", "viem": "2.55.10", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 938a0f7..8a92bcc 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -30,6 +30,9 @@ importers: '@vercel/blob': specifier: ^2.6.1 version: 2.6.1 + loglayer: + specifier: ^9.0.0 + version: 9.4.0 tsx: specifier: 4.23.1 version: 4.23.1 @@ -264,6 +267,26 @@ packages: '@jridgewell/sourcemap-codec@1.5.5': resolution: {integrity: sha512-cYQ9310grqxueWbl+WuIUIaiUaDcj7WOq5fVhEljNVgRfOUhY9fy2zTvfoqWsnebh8Sl70VScFbICvJnLKB0Og==} + '@loglayer/context-manager@2.3.0': + resolution: {integrity: sha512-bStCSDKUk88ZHTloOR+eE+Jfnfcs+jfI2hNSrcchorwi2UHS8AYHWMm9o2Ov3tSQ4v64je9JDeTVLgoeL+JHvw==} + engines: {node: '>=18'} + + '@loglayer/log-level-manager@2.3.0': + resolution: {integrity: sha512-tkiKKa5yBnWqCLUWFZraNAotaHxGlb8yqTUpYS1o5Qv3XeTPoNkI87oDwI8V8m63yvKLPrXsZtNas2pIiFldCQ==} + engines: {node: '>=18'} + + '@loglayer/plugin@3.3.0': + resolution: {integrity: sha512-qSEgukxOXrQsUPPUbcrOBGymowBfobTK+vC3+eDRx1dEqlhCl1YxTn8zIJjQKV+nZJyCRTzgPTvEfvceFO+QnQ==} + engines: {node: '>=18'} + + '@loglayer/shared@4.4.0': + resolution: {integrity: sha512-9C6zpO9RXjfjddR+Bbk+h5NiLDL2wNlno5jLplzhw+Jaj2gO2Dv53rxrgdRmBP5/DgSHNCZu6FdOq8ankSYdGA==} + engines: {node: '>=18'} + + '@loglayer/transport@3.3.0': + resolution: {integrity: sha512-fFh4XaX24FiyZtpzOjBP2PScA3sQk1rfka287FsYXkGMm2/n74BOVj6/D/oLaiI+cJMy7fv3dtyy5sUtNil5UQ==} + engines: {node: '>=18'} + '@noble/ciphers@1.3.0': resolution: {integrity: sha512-2I0gnIVPtfnMw9ee9h1dJG7tp81+8Ob3OJb3Mv37rx5L40/b0i7djjCVvGOVqc9AEIQyvyu1i6ypKdFw8R8gQw==} engines: {node: ^14.21.3 || >=16} @@ -700,6 +723,10 @@ packages: jose@5.10.0: resolution: {integrity: sha512-s+3Al/p9g32Iq+oqXxkW//7jk2Vig6FF1CFqzVXoTUXt2qz89YWbL+OwS17NFYEvxC35n0FKeGO2LGYSxeM2Gg==} + loglayer@9.4.0: + resolution: {integrity: sha512-abAZY0TkGw1LOA8n0yGJbRi9R0upRWymYBZItWfiktXMTHqdgCpHukONoX/Byfbt1RfRdw/GjT/PwrztePYOcw==} + engines: {node: '>=18'} + magic-string@0.30.21: resolution: {integrity: sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ==} @@ -1081,6 +1108,24 @@ snapshots: '@jridgewell/sourcemap-codec@1.5.5': {} + '@loglayer/context-manager@2.3.0': + dependencies: + '@loglayer/shared': 4.4.0 + + '@loglayer/log-level-manager@2.3.0': + dependencies: + '@loglayer/shared': 4.4.0 + + '@loglayer/plugin@3.3.0': + dependencies: + '@loglayer/shared': 4.4.0 + + '@loglayer/shared@4.4.0': {} + + '@loglayer/transport@3.3.0': + dependencies: + '@loglayer/shared': 4.4.0 + '@noble/ciphers@1.3.0': {} '@noble/curves@1.9.1': @@ -1418,6 +1463,14 @@ snapshots: jose@5.10.0: {} + loglayer@9.4.0: + dependencies: + '@loglayer/context-manager': 2.3.0 + '@loglayer/log-level-manager': 2.3.0 + '@loglayer/plugin': 3.3.0 + '@loglayer/shared': 4.4.0 + '@loglayer/transport': 3.3.0 + magic-string@0.30.21: dependencies: '@jridgewell/sourcemap-codec': 1.5.5 @@ -1656,6 +1709,7 @@ time: '@types/node@25.2.0': '2026-02-01T15:38:51.767Z' '@upstash/redis@1.38.1': '2026-07-31T12:11:53.818Z' '@vercel/blob@2.6.1': '2026-07-08T10:32:05.586Z' + loglayer@9.4.0: '2026-07-13T20:15:07.161Z' tsx@4.23.1: '2026-07-13T03:13:33.471Z' typescript@7.0.2: '2026-07-08T15:55:18.431Z' valibot@1.4.2: '2026-06-28T15:16:53.427Z' diff --git a/src/index.ts b/src/index.ts index bd3ab73..50aa2e5 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,4 +1,5 @@ export * from "./actions/index.js"; +export * from "./observability.js"; export * from "./transports/index.js"; export type * from "./types.js"; export * from "./utils/index.js"; diff --git a/src/observability.ts b/src/observability.ts new file mode 100644 index 0000000..0144f50 --- /dev/null +++ b/src/observability.ts @@ -0,0 +1,353 @@ +import { estimateUtf8Bytes } from "./utils/json.js"; +import { deepTransform } from "./utils/objects.js"; + +/** + * Minimal structural slice of `loglayer`'s `LogLayer`, which satisfies it directly. + * Declared locally so `loglayer` stays a *true* optional dep: the emitted `.d.ts` + * refers to `Logger`, not `import("loglayer").LogLayer`, so consumers who don't use + * it needn't install it to typecheck. + */ +export interface Logger { + child(): Logger; + withContext(context: Record): Logger; + withMetadata(metadata: Record): Logger; + withError(error: unknown): Logger; + info(message?: string): void; + warn(message?: string): void; + error(message?: string): void; + metadataOnly(metadata: Record): void; +} + +/** + * Accumulates one transport instance's fields on the call's wide event, under the + * prefix its {@link FacetId} claims. + * + * Re-allocating a facet for the same id returns the same slot, so a layer crossed + * many times per call (e.g. once per chunk under a divider fan-out) aggregates + * naturally. Writes are valid at any point in the operation's lifetime, including + * after `await`s. All facets share one byte budget per event; if it is exceeded, + * the largest fields are dropped and named in `truncated_fields`. + */ +export interface Facet { + /** + * Merges fields into this facet's slot; last write per field wins. Reserve it for + * once-per-call facts — on a layer crossed repeatedly, prefer `add`/`stat`/`push`. + */ + set(fields: Record): void; + /** Adds `n` (default 1) to a numeric accumulator field. */ + add(field: string, n?: number): void; + /** + * Records a sample into a streaming summary. Emitted at conclusion as + * `${field}.count`, `${field}.min`, `${field}.max`, and `${field}.avg`. + */ + stat(field: string, sample: number): void; + /** + * Appends to a bounded array (default limit 10). Values pushed past the limit + * are dropped and counted in `${field}_truncated`. + */ + push(field: string, value: unknown, limit?: number): void; + /** Returns a facet writing under `prefix` within this same slot. */ + sub(prefix: string): Facet; +} + +/** + * Identity of one transport instance, naming its fields on the wide event. + * + * The first id of a given key touched during a call writes bare `${key}.${field}` + * fields; later ids sharing that key (e.g. one cache per failover branch) get + * `${key}.1.*`, `${key}.2.*`, ... in first-touch order — stable across calls + * because transports traverse in a fixed order for a given composition. + * + * Create exactly one per composition node, at transport-factory scope, and pass it + * to both {@link observe} and {@link Observability.facet}: identity is by object + * reference, so an id created per request would claim a fresh label every call. + */ +export interface FacetId { + readonly key: string; +} + +export function createFacetId(key: string): FacetId { + return { key }; +} + +export type Observability = { + logger: Logger; + call_id: string; + facet(id: FacetId): Facet; +}; + +/** Streaming summary accumulator backing `Facet["stat"]`. */ +interface StatAcc { + count: number; + sum: number; + min: number; + max: number; +} + +/** Per-call state, shared by reference across the call's ALS scope; every facet writes here. */ +interface RootState { + /** First-touch labels: facet key → (id → field prefix). */ + labels: Map>; + /** Flattened `${prefix}.${field}` → value. */ + fields: Record; + /** Streaming summaries, keyed by full dotted field path. */ + stats: Map; +} + +class FacetImpl implements Facet { + constructor( + private readonly root: RootState, + private readonly prefix: string, + ) {} + + set(fields: Record): void { + for (const [k, v] of Object.entries(fields)) this.root.fields[`${this.prefix}.${k}`] = v; + } + + add(field: string, n = 1): void { + const fk = `${this.prefix}.${field}`; + const current = this.root.fields[fk]; + this.root.fields[fk] = (typeof current === "number" ? current : 0) + n; + } + + stat(field: string, sample: number): void { + const fk = `${this.prefix}.${field}`; + let acc = this.root.stats.get(fk); + if (!acc) { + acc = { count: 0, sum: 0, min: Number.POSITIVE_INFINITY, max: Number.NEGATIVE_INFINITY }; + this.root.stats.set(fk, acc); + } + acc.count += 1; + acc.sum += sample; + if (sample < acc.min) acc.min = sample; + if (sample > acc.max) acc.max = sample; + } + + push(field: string, value: unknown, limit = 10): void { + const fk = `${this.prefix}.${field}`; + let arr = this.root.fields[fk] as unknown[]; + if (!Array.isArray(arr)) { + arr = []; + this.root.fields[fk] = arr; + } + if (arr.length < limit) { + arr.push(value); + } else { + const tk = `${fk}_truncated`; + this.root.fields[tk] = ((this.root.fields[tk] as number) ?? 0) + 1; + } + } + + sub(prefix: string): Facet { + return new FacetImpl(this.root, `${this.prefix}.${prefix}`); + } +} + +/** Suffixes count per key ({@link FacetId}), so unrelated keys never shift each other's labels. */ +function resolvePrefix(root: RootState, id: FacetId): string { + let labels = root.labels.get(id.key); + if (!labels) { + labels = new Map(); + root.labels.set(id.key, labels); + } + let prefix = labels.get(id); + if (prefix === undefined) { + prefix = labels.size === 0 ? id.key : `${id.key}.${labels.size}`; + labels.set(id, prefix); + } + return prefix; +} + +/** + * Per-operation ALS scope. `parentLogger` and `context` are the seed captured by + * `withLogging`; `obs` and `root` are derived once by the outermost {@link observe} + * and reused by every inner boundary, so one call produces one wide event. + */ +interface Scope { + parentLogger: Logger; + context: Record; + obs?: Observability; + root?: RootState; +} + +type AlsLike = { + getStore(): Scope | undefined; + run(store: Scope, fn: () => R): R; +}; + +let als: AlsLike | undefined; +let alsLoad: Promise | undefined; + +// Imported lazily so bundles that never call `withLogging` don't pull in +// `node:async_hooks`, and so environments lacking it (e.g. unpolyfilled browsers) +// degrade to the no-op path rather than failing to load. Memoizing the in-flight +// promise (not just its result) is load-bearing: scopes racing the first import must +// share one instance, or `withLogging` would open a scope on a storage that `observe` +// isn't reading, and that call would silently emit nothing. +function loadAls(): Promise { + alsLoad ??= import("node:async_hooks").then( + ({ AsyncLocalStorage }) => { + als = new AsyncLocalStorage(); + return als; + }, + () => undefined, + ); + return alsLoad; +} + +export interface WithLoggingOpts { + logger: Logger; + /** Additional context fields, stamped onto every event emitted in this scope. */ + [key: string]: unknown; +} + +/** + * Opens an ALS scope seeding `logger` and `opts` for the duration of `fn`. Each + * outermost viem-dlc transport call made inside `fn` (synchronously or via awaits) + * derives its own child logger and emits one wide event; the transport layers it + * nests through contribute fields to that same event. Outside the scope, the + * library emits nothing. + * + * Parallel `client.request` calls inside one `withLogging` scope are fully + * isolated from one another. + * + * In environments without `AsyncLocalStorage`, this is a no-op pass-through. + */ +export async function withLogging(fn: () => Promise | T, opts: WithLoggingOpts): Promise { + const storage = await loadAls(); + if (!storage) return fn(); + + const { logger, ...rest } = opts; + + return storage.run({ parentLogger: logger, context: rest }, fn); +} + +/** Soft ceiling on accumulated facet bytes emitted on one wide event. */ +const MAX_FIELDS_BYTES = 32 * 1024; + +function fieldSize(key: string, value: unknown): number { + try { + return key.length + estimateUtf8Bytes(value); + } catch { + // Unestimable (e.g. circular) values are treated as oversized so they're dropped first. + return Number.POSITIVE_INFINITY; + } +} + +/** + * Resolves stat accumulators into scalar fields and enforces `MAX_FIELDS_BYTES`, + * dropping the largest fields first and recording their names in `truncated_fields`. + */ +function finalizeFields(root: RootState): Record { + const { fields } = root; + + for (const [fk, acc] of root.stats) { + fields[`${fk}.count`] = acc.count; + fields[`${fk}.min`] = acc.min; + fields[`${fk}.max`] = acc.max; + fields[`${fk}.avg`] = acc.sum / acc.count; + } + + let total = 0; + for (const [k, v] of Object.entries(fields)) total += fieldSize(k, v); + if (total <= MAX_FIELDS_BYTES) return fields; + + // Over budget (rare): re-measure so the largest fields can be dropped first. + const sizes = Object.entries(fields) + .map(([k, v]) => [k, fieldSize(k, v)] as const) + .sort((a, b) => b[1] - a[1]); + const dropped: string[] = []; + for (const [k, size] of sizes) { + if (total <= MAX_FIELDS_BYTES) break; + delete fields[k]; + dropped.push(k); + total -= size; + } + fields.truncated_fields = dropped; + + return fields; +} + +/** + * Inherit-or-originate primitive used by every viem-dlc transport's `request` fn. + * + * The outermost boundary for a call derives the per-call child logger and facet + * accumulator, then emits one `"concluded"` wide event when the call settles: + * `info`-level on success, `error`-level with `withError` on rejection. Inner + * boundaries contribute to that same event; no new ALS store is created past the + * outermost one. + * + * Every crossing (outermost included) increments `id`'s `crossings` field, so the + * event records which transports this call traversed and how many times each. + * Running on boundary entry also pins `id`'s label before any handler writes. + */ +export function observe Promise>(fn: F, id: FacetId): F { + const countCrossing = (root: RootState) => { + const fk = `${resolvePrefix(root, id)}.crossings`; + root.fields[fk] = ((root.fields[fk] as number) ?? 0) + 1; + }; + const wrapped = (req: Parameters[0]) => { + const scope = als?.getStore(); + if (!als || !scope) return fn(req); + if (scope.root) { + countCrossing(scope.root); + return fn(req); + } + + const call_id = crypto.randomUUID(); + const logger = scope.parentLogger.child().withContext({ + // Seeded context first, so the canonical fields below can't be overwritten. + ...scope.context, + library: "viem-dlc", + // Trimmed so a large calldata or filter payload can't dominate the event. + req: deepTransform(req, { + transformLeaf: (v: T) => (typeof v === "string" && v.length > 100 ? v.slice(0, 97).concat("...") : v) as T, + }), + call_id, + }); + const root: RootState = { labels: new Map(), fields: {}, stats: new Map() }; + countCrossing(root); + const obs: Observability = { + logger, + call_id, + facet: (facetId) => new FacetImpl(root, resolvePrefix(root, facetId)), + }; + return als.run({ ...scope, obs, root }, () => { + const t0 = performance.now(); + const conclude = (status: "ok" | "error", error?: unknown) => { + const fields = finalizeFields(root); + fields.status = status; + fields.duration_ms = performance.now() - t0; + // `withError` on the error path so hosts wired like Morpho's `@repo/observability` + // (LogLayer plugin forwarding `.withError()` entries to an ErrorReporter) capture it. + const enriched = logger.withContext(fields); + if (status === "ok") enriched.info("concluded"); + else enriched.withError(error).error("concluded"); + }; + return fn(req).then( + (result) => { + conclude("ok"); + return result; + }, + (error) => { + conclude("error", error); + throw error; + }, + ); + }); + }; + return wrapped as F; +} + +/** + * Reads the active per-request observability scope from ambient ALS. Returns + * `undefined` when called outside a `withLogging` scope, in environments + * without `AsyncLocalStorage`, or before `observe` has derived a per-call + * child logger. + * + * ALS context flows through `await`s, so this may be called at any point in a + * transport's lifetime. + */ +export function getObservability(): Observability | undefined { + return als?.getStore()?.obs; +} diff --git a/src/stores/hierarchical.ts b/src/stores/hierarchical.ts index dcafd39..b8212cf 100644 --- a/src/stores/hierarchical.ts +++ b/src/stores/hierarchical.ts @@ -1,3 +1,4 @@ +import type { Logger } from "../observability.js"; import type { Store } from "../types.js"; /** @@ -11,19 +12,23 @@ import type { Store } from "../types.js"; export class HierarchicalStore implements Store { constructor( private readonly stores: readonly Store[], - private readonly options?: { populateOnMiss?: boolean }, + private readonly options?: { populateOnMiss?: boolean; logger?: Logger }, ) {} async get(key: string) { for (let i = 0; i < this.stores.length; i++) { const value = await this.stores[i]!.get(key); if (value !== null) { + this.options?.logger + ?.withMetadata({ class: HierarchicalStore.name, method: "get", level: i, key }) + .info("cache hit"); if (this.options?.populateOnMiss) { void Promise.all(this.stores.slice(0, i).map((store) => store.set(key, value))); } return value; } } + this.options?.logger?.withMetadata({ class: HierarchicalStore.name, method: "get", key }).info("cache miss"); return null; } diff --git a/src/stores/lru.ts b/src/stores/lru.ts index 6cf9d22..e6e6852 100644 --- a/src/stores/lru.ts +++ b/src/stores/lru.ts @@ -1,18 +1,34 @@ +import type { Logger } from "../observability.js"; import type { Store } from "../types.js"; function sizeOf(buffers: Buffer[]) { return buffers.reduce((acc, b) => acc + b.byteLength, 0); } +export type LruStoreOptions = { + maxBytes: number; + /** Optional Logger for non-request-bound emissions (e.g. oversized-value drops). */ + logger?: Logger; +}; + /** LRU cache with byte-based size limit (only values counted, keys assumed negligible). */ export class LruStore implements Store { private readonly maxBytes: number; + private readonly logger?: Logger; private readonly map = new Map(); private bytes = 0; - constructor(maxBytes: number) { - if (maxBytes < 1) throw new Error("[LruStore] maxBytes must be at least 1"); + constructor({ maxBytes, logger }: LruStoreOptions) { + // Rejects a non-number outright so the superseded `new LruStore(bytes)` form fails + // loudly; left to the `< 1` check alone it would yield `undefined`, making every + // size comparison false and turning this into an unbounded map. + if (typeof maxBytes !== "number" || !Number.isFinite(maxBytes) || maxBytes < 1) { + const err = new Error(`[LruStore] maxBytes must be at least 1 (got ${String(maxBytes)})`); + logger?.withMetadata({ class: LruStore.name, method: "constructor" }).withError(err).error(); + throw err; + } this.maxBytes = maxBytes; + this.logger = logger; } get(key: string) { @@ -28,20 +44,27 @@ export class LruStore implements Store { const size = sizeOf(value); if (size > this.maxBytes) { - console.warn(`[LruStore] Value exceeds maxBytes (${size} > ${this.maxBytes}), skipping`); + this.logger + ?.withMetadata({ class: LruStore.name, method: "set", key, size, max_bytes: this.maxBytes }) + .warn("value exceeds maxBytes, skipping"); return; } + const evicted: string[] = []; while (this.bytes + size > this.maxBytes) { // Non-null assertion is safe because map has entries until `this.bytes === 0`, // and once it's zero, the loop condition breaks because `size <= this.maxBytes`. const [oldestKey, oldest] = this.map.entries().next().value!; this.bytes -= sizeOf(oldest); this.map.delete(oldestKey); + + evicted.push(oldestKey); } this.map.set(key, value); this.bytes += size; + + this.logger?.metadataOnly({ class: LruStore.name, method: "set", key, size, max_bytes: this.maxBytes, evicted }); } delete(key: string) { diff --git a/src/stores/throttled.ts b/src/stores/throttled.ts index 9145a35..811a698 100644 --- a/src/stores/throttled.ts +++ b/src/stores/throttled.ts @@ -1,5 +1,6 @@ import { withTimeout } from "viem"; +import type { Logger } from "../observability.js"; import type { Store } from "../types.js"; import { createRateLimit, RateLimitGateError } from "../utils/with-rate-limit.js"; @@ -16,6 +17,12 @@ export type ThrottledStoreOptions = { maxStalenessMs: number; /** Optional: handle write errors (default: ignore) -- MUST NOT THROW. */ onWriteError?: (key: string, err: unknown, durationMs: number) => void; + /** + * Optional logger for non-request-bound emissions (e.g. background flush boundaries). + * Per-`set`/`get`/`delete` events are read from the ambient ALS scope via `getCurrentLog()` + * — this field is only needed for events that fire outside any caller's async scope. + */ + logger?: Logger; }; /** @@ -35,6 +42,7 @@ export class ThrottledStore implements Store { private readonly rateLimiter: ReturnType; private readonly maxStalenessMs: number; private readonly onWriteError?: (key: string, err: unknown, durationMs: number) => void; + protected readonly logger?: Logger; /** Latest pending op per key. Written at call time, read lazily at admission time, deleted at completion. */ private readonly pending = new Map(); @@ -50,6 +58,7 @@ export class ThrottledStore implements Store { this.rateLimiter = createRateLimit(opts.maxWritesBurst, opts.maxWritesPerSecond, opts.maxConcurrent); this.maxStalenessMs = opts.maxStalenessMs; this.onWriteError = opts.onWriteError; + this.logger = opts.logger; } get(key: string) { @@ -108,6 +117,16 @@ export class ThrottledStore implements Store { }, ); } catch (err) { + this.logger + ?.withMetadata({ + class: ThrottledStore.name, + method: "ensureQueued", + key, + kind: entry.op.kind, + duration_ms: Date.now() - t0, + }) + .withError(err) + .warn("upstream write failed"); this.onWriteError?.(key, err, Date.now() - t0); } @@ -123,6 +142,15 @@ export class ThrottledStore implements Store { const isValid = Date.now() - entry.lastUpdatedAt <= this.maxStalenessMs; if (!isValid) { + this.logger + ?.withMetadata({ + class: ThrottledStore.name, + method: "ensureQueued", + key, + kind: entry.op.kind, + age_ms: Date.now() - entry.lastUpdatedAt, + }) + .info("dropped stale pending op"); this.pending.delete(key); this.resolveFlushBoundaries(key, entry.version); } diff --git a/src/stores/ttl.ts b/src/stores/ttl.ts index 8e2e406..c5ab759 100644 --- a/src/stores/ttl.ts +++ b/src/stores/ttl.ts @@ -35,7 +35,7 @@ export type TtlStoreOptions = { * * The stamp is set on `set` and never refreshed on `get`, bounding how long this tier may diverge from * a fresher source behind it. The intended shape is an in-memory front atop a {@link HierarchicalStore} - * (e.g. `new TtlStore(new LruStore(maxBytes), { ttlMs })`): a plain `LruStore` would pin a warm copy for + * (e.g. `new TtlStore(new LruStore({ maxBytes: maxBytes }), { ttlMs })`): a plain `LruStore` would pin a warm copy for * the whole process lifetime, whereas this expires it so the next read picks up the source of truth. * * Best-effort and non-throwing, per the `Store` contract, and passes the wrapped store's sync/async diff --git a/src/stores/upstash.ts b/src/stores/upstash.ts index a815b7c..25d668d 100644 --- a/src/stores/upstash.ts +++ b/src/stores/upstash.ts @@ -3,6 +3,7 @@ import { randomBytes, randomUUID } from "crypto"; import { Redis, type RedisConfigNodejs } from "@upstash/redis"; +import type { Logger } from "../observability.js"; import type { Store } from "../types.js"; import { createInFlightBarrier } from "../utils/in-flight.js"; import { shardString } from "../utils/strings.js"; @@ -15,6 +16,8 @@ export type UpstashStoreOptions = { maxRequestBytes: number; ttl?: number; redis?: Omit; + /** Optional logger for non-request-bound emissions (e.g. background I/O errors). */ + logger?: Logger; }; class WriteId { @@ -69,13 +72,17 @@ export class UpstashStore implements Store { constructor(options: UpstashStoreOptions) { if (!Number.isSafeInteger(options.maxRequestBytes) || options.maxRequestBytes! <= WriteId.LENGTH) { - throw new Error( - `[UpstashStore] maxRequestBytes must be a safe integer > ${WriteId.LENGTH} (got ${options.maxRequestBytes})`, + const err = new Error( + `maxRequestBytes must be a safe integer > ${WriteId.LENGTH} (got ${options.maxRequestBytes})`, ); + options.logger?.withMetadata({ class: UpstashStore.name, method: "constructor" }).withError(err).error(); + throw err; } if (options.ttl !== undefined && (!Number.isSafeInteger(options.ttl) || options.ttl! <= 0)) { - throw new Error(`[UpstashStore] ttl must be a positive safe integer (got ${options.ttl})`); + const err = new Error(`ttl must be a positive safe integer (got ${options.ttl})`); + options.logger?.withMetadata({ class: UpstashStore.name, method: "constructor" }).withError(err).error(); + throw err; } this.options = options; @@ -103,6 +110,9 @@ export class UpstashStore implements Store { // If shard is null, array must've been shortened after our initial read (non-atomic inconsistency). if (shardWithId === null) { + this.options.logger + ?.withMetadata({ class: UpstashStore.name, method: "_get", key }) + .info("non-atomic inconsistency: skewed length"); return { value: null, motivatesRetry: true }; } @@ -110,12 +120,17 @@ export class UpstashStore implements Store { // If writeId doesn't match, array must've been overwritten after our initial read (non-atomic inconsistency). if (writeId !== writeId0) { + this.options.logger + ?.withMetadata({ class: UpstashStore.name, method: "_get", key }) + .info("non-atomic inconsistency: skewed shard"); return { value: null, motivatesRetry: true }; } result += shard; // appending like this lets V8 engine defer memory copy until `result` is consumed } + this.options.logger?.metadataOnly({ class: UpstashStore.name, method: "_get", key, len }); + return { value: [Buffer.from(result, "base64")], motivatesRetry: false }; } @@ -132,6 +147,7 @@ export class UpstashStore implements Store { } } + this.options?.logger?.withMetadata({ class: UpstashStore.name, method: "get", key }).warn("exhausted retries"); return null; } @@ -178,7 +194,10 @@ export class UpstashStore implements Store { try { await this.inFlight.track(this._set(key, value)); } catch (err) { - console.warn(`[UpstashStore] Failed to set key "${key}":`, err); + this.options.logger + ?.withMetadata({ class: UpstashStore.name, method: "set", key }) + .withError(err) + .warn("set failed"); } } @@ -186,7 +205,10 @@ export class UpstashStore implements Store { try { await this.inFlight.track(this.redis.unlink(key)); } catch (err) { - console.warn(`[UpstashStore] Failed to delete key "${key}":`, err); + this.options.logger + ?.withMetadata({ class: UpstashStore.name, method: "delete", key }) + .withError(err) + .warn("delete failed"); } } @@ -194,7 +216,10 @@ export class UpstashStore implements Store { try { await this.inFlight.flush(); } catch (err) { - console.warn("[UpstashStore] Failed to flush:", err); + this.options.logger + ?.withMetadata({ class: UpstashStore.name, method: "flush" }) + .withError(err) + .warn("flush failed"); } } } @@ -210,14 +235,15 @@ export function createOptimizedUpstashStore(options: UpstashStoreOptions) { // We coalesce writes per key and rate-limit remote persistence. return new HierarchicalStore( [ - new LruStore(1 << 30), // 1 GB + new LruStore({ maxBytes: 1 << 30, logger: options.logger }), // 1 GB new ThrottledStore(remote, { maxStalenessMs: 60_000, // defend against serverless freeze/thaw cycles maxWritesBurst, maxWritesPerSecond, maxConcurrent: Infinity, + logger: options.logger, }), ], - { populateOnMiss: true }, + { populateOnMiss: true, logger: options.logger }, ); } diff --git a/src/stores/vercel.ts b/src/stores/vercel.ts index f33b877..02e8190 100644 --- a/src/stores/vercel.ts +++ b/src/stores/vercel.ts @@ -3,6 +3,7 @@ import { createHash } from "crypto"; import { del, get, put } from "@vercel/blob"; +import type { Logger } from "../observability.js"; import type { Store } from "../types.js"; import { createInFlightBarrier } from "../utils/in-flight.js"; @@ -46,6 +47,8 @@ export type VercelStoreOptions = { * eventual consistency in exchange for fewer Simple Operations. */ useCdnCache?: boolean; + /** Optional logger for non-request-bound emissions (e.g. background I/O errors). */ + logger?: Logger; }; /** @@ -99,7 +102,9 @@ export class VercelStore implements Store { options.cacheControlMaxAge !== undefined && (!Number.isFinite(options.cacheControlMaxAge) || options.cacheControlMaxAge < 60) ) { - throw new Error(`[VercelStore] cacheControlMaxAge must be >= 60 seconds (got ${options.cacheControlMaxAge})`); + const err = new Error(`cacheControlMaxAge must be >= 60 seconds (got ${options.cacheControlMaxAge})`); + options.logger?.withMetadata({ class: VercelStore.name, method: "constructor" }).withError(err).error(); + throw err; } this.options = options; @@ -145,7 +150,10 @@ export class VercelStore implements Store { try { return await this._get(key); } catch (err) { - console.warn(`[VercelStore] Failed to get key "${key}":`, err); + this.options.logger + ?.withMetadata({ class: VercelStore.name, method: "get", key }) + .withError(err) + .warn("get failed"); return null; } } @@ -168,7 +176,10 @@ export class VercelStore implements Store { try { await this.inFlight.track(this._set(key, value)); } catch (err) { - console.warn(`[VercelStore] Failed to set key "${key}":`, err); + this.options.logger + ?.withMetadata({ class: VercelStore.name, method: "set", key }) + .withError(err) + .warn("set failed"); } } @@ -176,7 +187,10 @@ export class VercelStore implements Store { try { await this.inFlight.track(Promise.resolve(del(this.resolvePathname(key), this.tokenOption))); } catch (err) { - console.warn(`[VercelStore] Failed to delete key "${key}":`, err); + this.options.logger + ?.withMetadata({ class: VercelStore.name, method: "delete", key }) + .withError(err) + .warn("delete failed"); } } @@ -184,7 +198,10 @@ export class VercelStore implements Store { try { await this.inFlight.flush(); } catch (err) { - console.warn("[VercelStore] Failed to flush:", err); + this.options.logger + ?.withMetadata({ class: VercelStore.name, method: "flush" }) + .withError(err) + .warn("flush failed"); } } } @@ -204,14 +221,15 @@ export function createOptimizedVercelStore(options: VercelStoreOptions = {}) { // and shields the remote tier from network/operation cost on hot keys. return new HierarchicalStore( [ - new LruStore(1 << 30), // 1 GB + new LruStore({ maxBytes: 1 << 30, logger: options.logger }), // 1 GB new ThrottledStore(remote, { maxStalenessMs: 60_000, // defend against serverless freeze/thaw cycles maxWritesBurst, maxWritesPerSecond, maxConcurrent: Infinity, + logger: options.logger, }), ], - { populateOnMiss: true }, + { populateOnMiss: true, logger: options.logger }, ); } diff --git a/src/transports/cache/eth-call/handler.ts b/src/transports/cache/eth-call/handler.ts index 9325fd4..efb1170 100644 --- a/src/transports/cache/eth-call/handler.ts +++ b/src/transports/cache/eth-call/handler.ts @@ -1,6 +1,7 @@ import type { Hex } from "viem"; import { LazyNdjsonMap } from "../../../internal/lazy-ndjson-map.js"; +import { getObservability } from "../../../observability.js"; import type { EIP1193Parameters } from "../../../types.js"; import { cyrb64Hash } from "../../../utils/hash.js"; import { @@ -21,7 +22,7 @@ import type { HandlerContext } from "../types.js"; import type { CachedEthCallEntry } from "./types.js"; export async function handleEthCall( - { store, coalesce, requestFn, chainId, gasLimit }: HandlerContext, + { store, coalesce, requestFn, chainId, gasLimit, facetId }: HandlerContext, req: EIP1193Parameters, ): Promise { const extracted = extractEthCallPolicy(req.params[2]); @@ -29,6 +30,8 @@ export async function handleEthCall( return requestFn(req); } + const facet = getObservability()?.facet(facetId).sub("eth_call"); + const [txn, ...restOfEthCallParams] = req.params; if (txn.data === undefined) { throw new Error("[cache] eth_call with policy requires `data`"); @@ -55,6 +58,8 @@ export async function handleEthCall( const solidity = resolveArrayFunction(extracted.policy.abi, extracted.policy.paged); const inputElements = calldataToArray(solidity, targetData); + facet?.set({ input_elements: inputElements.length }); + if (inputElements.length === 0) { return arrayToHex(solidity.outputLayout, []); } @@ -72,10 +77,12 @@ export async function handleEthCall( batch: extracted.policy.batch, gasLimit, restOfEthCallParams, + facet, }); return arrayToHex(solidity.outputLayout, outputs); } + facet?.set({ blob_key: blobKey, ttl_ms: ttl, delta_ms: delta }); return coalesce(blobKey, req, async (_leaderReq, collectFollowers) => { /*////////////////////////////////////////////////////////////// LEADER OPS @@ -97,9 +104,13 @@ export async function handleEthCall( keyToInfo.set(ek, { indices: [i], element }); } }); + facet?.set({ input_elements_unique: keyToInfo.size }); // Open blob lazily — read once, buffer writes, flush when done. + const t0 = performance.now(); let buffers = (await store.get(blobKey)) ?? []; + const t1 = performance.now(); + const ndjson = new LazyNdjsonMap( { toJson: stringify, fromJson: parse }, { @@ -116,6 +127,7 @@ export async function handleEthCall( const misses: { entryKey: string; indices: number[]; element: Hex }[] = []; const now = Date.now(); + const t2 = performance.now(); await ndjson.scan((record) => { const match = keyToInfo.get(record.key); if (!match) return; @@ -131,11 +143,16 @@ export async function handleEthCall( if (keyToInfo.size === 0) return false; }); + const t3 = performance.now(); for (const [entryKey, info] of keyToInfo) { misses.push({ entryKey, ...info }); } + // `factorisedFactoryCall` overwrites both when it runs; these cover the zero-misses + // (full cache hit) case, where it doesn't run at all. + facet?.set({ elements_requested: misses.length, elements_fetched: 0 }); + // Fetch misses if (misses.length > 0) { const fetchedAt = Date.now(); @@ -148,6 +165,7 @@ export async function handleEthCall( batch: extracted.policy.batch, gasLimit, restOfEthCallParams, + facet, // Buffer per chunk, so a later chunk failing doesn't discard the siblings that landed. onResolved: (entries) => { ndjson.upsert( @@ -163,6 +181,9 @@ export async function handleEthCall( // Rebase onto the caller's input: `missing` indexes deduped misses, `data` omits hits. if (isDeploylessPartialResultError(e)) { const missing = e.missing.flatMap((i) => misses[i]!.indices).sort((a, b) => a - b); + // Deduping means one unservable entry can stand for several caller inputs; restamp + // so the field matches the `missing` carried by the error we actually throw. + facet?.set({ elements_missing: missing.length }); throw new DeploylessPartialResultError({ // Sparse exactly at the missing indices, and `filter` skips holes. data: arrayToHex( @@ -175,7 +196,9 @@ export async function handleEthCall( } throw e; } finally { + const t4 = performance.now(); await ndjson.flush(); + facet?.set({ fetch_cache_ms: t1 - t0, read_cache_ms: t3 - t2, flush_cache_ms: performance.now() - t4 }); } } @@ -188,6 +211,7 @@ export async function handleEthCall( const leaderHash = cyrb64Hash(JSON.stringify(req.params)); const collected = collectFollowers(); const matching = collected.filter((f) => cyrb64Hash(JSON.stringify(f.args.params)) === leaderHash); + facet?.set({ n_followers: matching.length }); return { leader: { action: "resolve", result }, diff --git a/src/transports/cache/eth-get-logs/handler.ts b/src/transports/cache/eth-get-logs/handler.ts index e4926d9..38522b7 100644 --- a/src/transports/cache/eth-get-logs/handler.ts +++ b/src/transports/cache/eth-get-logs/handler.ts @@ -2,6 +2,7 @@ import { hexToBigInt, type RpcLog, toHex } from "viem"; import { LazyNdjsonMap } from "../../../internal/lazy-ndjson-map.js"; import type { LazyEntry } from "../../../internal/ndjson-map.js"; +import { getObservability } from "../../../observability.js"; import type { BlockRange, EIP1193Parameters } from "../../../types.js"; import { divideBlockRange, extractRangeFromFilter, isInBlockRange, mergeBlockRanges } from "../../../utils/blocks.js"; import { tryCatch } from "../../../utils/errors.js"; @@ -41,9 +42,11 @@ function shouldFetchRange( } export async function handleEthGetLogs( - { binSize, invalidationStrategy, store, coalesce, requestFn, chainId }: HandlerContext, + ctx: HandlerContext, req: EIP1193Parameters, ): Promise { + const { binSize, invalidationStrategy, store, coalesce, requestFn, chainId } = ctx; + const facet = getObservability()?.facet(ctx.facetId).sub("eth_getLogs"); const blobKey = keychain.blobKey(chainId, req); return coalesce(blobKey, req, async (args, collectFollowers) => { @@ -119,11 +122,15 @@ export async function handleEthGetLogs( // Start fetching all gaps. `logsDivider` and `rateLimiter` handle splitting, concurrency, and rate limits. // viem also provides request deduplication at each layer, and at this point we've already normalized it. + let gapsFetchedCount = 0; + let fetchMs = 0; if (gaps.length > 0) { const rangesToFetch = mergeBlockRanges(gaps); + gapsFetchedCount = rangesToFetch.length; const sink = createSink({ chainId, binSize, ndjson }); + const tFetchStart = performance.now(); try { await Promise.all( rangesToFetch.map((range) => @@ -154,6 +161,8 @@ export async function handleEthGetLogs( throw error; } throw new Error(`${context} ${String(error)}`); + } finally { + fetchMs = performance.now() - tFetchStart; } } @@ -167,6 +176,17 @@ export async function handleEthGetLogs( const leader = { slot: -1, args }; const followers = collectFollowers(); + // TODO(observability): once `coalesce` carries a `meta: { call_id }` per + // caller, stamp `served_by` onto followers' facets and record + // `follower_call_ids` here. For now the event only describes leader-side work. + facet?.set({ + blob_key: blobKey, + gaps_fetched: gapsFetchedCount, + n_followers: followers.length, + blob_bytes_written: buffers.reduce((s, b) => s + b.byteLength, 0), + fetch_ms: fetchMs, + }); + const leaderFilterJson = JSON.stringify(filter); const participants = [leader, ...followers] .filter((f) => JSON.stringify(f.args.params[0]) === leaderFilterJson) diff --git a/src/transports/cache/index.ts b/src/transports/cache/index.ts index b1b03cd..6e71693 100644 --- a/src/transports/cache/index.ts +++ b/src/transports/cache/index.ts @@ -1,5 +1,6 @@ import { createTransport, type EIP1193RequestFn, type PublicRpcSchema, type Transport } from "viem"; +import { createFacetId, observe } from "../../observability.js"; import type { EIP1193Parameters, Store } from "../../types.js"; import { createCoalescingMutex } from "../../utils/coalescing-mutex.js"; import { type LogsDividerConfig, logsDivider } from "../logs-divider/index.js"; @@ -10,11 +11,12 @@ import type { RateLimiterConfig } from "../rate-limiter/index.js"; import { handleEthCall } from "./eth-call/handler.js"; import { handleEthGetLogs } from "./eth-get-logs/handler.js"; import { normalize } from "./normalization.js"; -import type { CachedMethod, CacheSchema } from "./schema.js"; +import { type CachedMethod, type CacheSchema, cacheTransportKey } from "./schema.js"; import type { CacheConfig, HandlerContext, InvalidationStrategy } from "./types.js"; export type * from "./schema.js"; export type * from "./types.js"; +export { cacheTransportKey }; /** * @param alphaAge Exponential growth rate w.r.t cache entry age (in time). @default 1/8 @@ -74,8 +76,6 @@ export function createSimpleInvalidation( }; } -export const cacheTransportKey = "viem-dlc-cache" as const; - /** * Creates an all-in-one caching transport for eth_getLogs calls. * @@ -100,7 +100,7 @@ export const cacheTransportKey = "viem-dlc-cache" as const; * const transport = cache( * http(rpcUrl), * [ - * { binSize: 10_000, store: new LruStore(), invalidationStrategy: createSimpleInvalidation() }, + * { binSize: 10_000, store: new LruStore({ maxBytes: 1 << 30 }), invalidationStrategy: createSimpleInvalidation() }, * { maxBlockRange: 100_000 }, * { maxRequestsPerSecond: 10, maxConcurrentRequests: 5 } * ] @@ -118,6 +118,8 @@ export function cache( RateLimiterConfig, ], ): Transport> { + const facetId = createFacetId(cacheTransportKey); + return (params) => { if (params.chain === undefined) { throw new Error("You must pass a chain to the cache transport."); @@ -137,6 +139,7 @@ export function cache( chainId, requestFn: transport.request, coalesce, + facetId, }; const request = (req: EIP1193Parameters) => { @@ -162,7 +165,7 @@ export function cache( { key: cacheTransportKey, name: "[viem-dlc] cache", - request: request as EIP1193RequestFn, + request: observe(request, facetId) as EIP1193RequestFn, retryCount: 0, type: cacheTransportKey, }, diff --git a/src/transports/cache/schema.ts b/src/transports/cache/schema.ts index abd2606..52a25ab 100644 --- a/src/transports/cache/schema.ts +++ b/src/transports/cache/schema.ts @@ -34,3 +34,5 @@ export type CacheSchema = SafelyExtendRpcSchema< export const cachedMethods = ["eth_call", "eth_getLogs"] as const satisfies EIP1193Parameters["method"][]; export type CachedMethod = (typeof cachedMethods)[number]; + +export const cacheTransportKey = "viem-dlc-cache" as const; diff --git a/src/transports/cache/types.ts b/src/transports/cache/types.ts index 6c2cf4d..faec4b1 100644 --- a/src/transports/cache/types.ts +++ b/src/transports/cache/types.ts @@ -1,5 +1,6 @@ import type { EIP1193RequestFn } from "viem"; +import type { FacetId } from "../../observability.js"; import type { Store } from "../../types.js"; import type { createCoalescingMutex } from "../../utils/coalescing-mutex.js"; import type { LogsDividerSchema } from "../logs-divider/schema.js"; @@ -36,4 +37,6 @@ export type HandlerContext = CacheConfig & { chainId: number; requestFn: EIP1193RequestFn; coalesce: ReturnType["coalesce"]; + /** Owning transport's facet identity; see {@link FacetId}. */ + facetId: FacetId; }; diff --git a/src/transports/deployless/index.ts b/src/transports/deployless/index.ts index c16b987..105b2ea 100644 --- a/src/transports/deployless/index.ts +++ b/src/transports/deployless/index.ts @@ -1,5 +1,6 @@ import { createTransport, type EIP1193RequestFn, type PublicRpcSchema, type Transport } from "viem"; +import { createFacetId, type FacetId, getObservability, observe } from "../../observability.js"; import type { EIP1193Parameters, SafelyExtendedRpcSchema } from "../../types.js"; import { factorisedFactoryCall } from "../../utils/deployless/call.js"; import { unwrapDeploylessFactoryCall } from "../../utils/deployless/codec.envelope.js"; @@ -29,6 +30,8 @@ export function deployless( baseTransportFn: Transport>, { gasLimit }: DeploylessConfig, ): Transport> { + const facetId = createFacetId(deploylessTransportKey); + return (params) => { const requestFn = baseTransportFn(params).request; @@ -37,14 +40,14 @@ export function deployless( return requestFn(args); } - return handleEthCall(requestFn, args as EIP1193Parameters, gasLimit); + return handleEthCall(requestFn, args as EIP1193Parameters, gasLimit, facetId); }; return createTransport( { key: deploylessTransportKey, name: "[viem-dlc] deployless", - request: request as EIP1193RequestFn, + request: observe(request, facetId) as EIP1193RequestFn, retryCount: 0, type: deploylessTransportKey, }, @@ -57,6 +60,7 @@ async function handleEthCall( requestFn: EIP1193RequestFn, req: EIP1193Parameters, gasLimit: number, + facetId: FacetId, ) { const extracted = extractEthCallPolicy(req.params[2]); if (!extracted) { @@ -89,6 +93,9 @@ async function handleEthCall( const solidity = resolveArrayFunction(extracted.policy.abi, extracted.policy.paged); const inputElements = calldataToArray(solidity, targetData); + const facet = getObservability()?.facet(facetId).sub("eth_call"); + facet?.set({ input_elements: inputElements.length }); + if (inputElements.length === 0) { return arrayToHex(solidity.outputLayout, []); } @@ -100,6 +107,7 @@ async function handleEthCall( batch: extracted.policy.batch, gasLimit, restOfEthCallParams, + facet, }); return arrayToHex(solidity.outputLayout, outputs); } diff --git a/src/transports/failover/index.ts b/src/transports/failover/index.ts index 1fa4dfe..a8c8ccb 100644 --- a/src/transports/failover/index.ts +++ b/src/transports/failover/index.ts @@ -1,7 +1,8 @@ import type { EIP1193RequestFn, RpcSchema, Transport } from "viem"; +import { createFacetId, getObservability, observe } from "../../observability.js"; import type { EIP1193Parameters } from "../../types.js"; -import { isTerminalError } from "../../utils/errors.js"; +import { isTerminalError, serializeError } from "../../utils/errors.js"; export const failoverTransportKey = "viem-dlc-failover" as const; @@ -25,8 +26,10 @@ export interface FailoverConfig { * * Each request tries `transports[0].request` first; on an error that is not * classified as "should throw," tries `transports[1].request`, and so on. If - * every branch fails, the last error is rethrown. Halving for range/size errors - * should be handled inside each branch — `failover` only sees errors that escape. + * every branch fails, the last error is rethrown. Errors carrying a partial payload are + * terminal and propagate immediately, ahead of and not overridable by `shouldThrow`, since + * falling over would discard what was already fetched. Halving for range/size errors should + * be handled inside each branch — `failover` only sees errors that escape. * * @example * const transport = failover([ @@ -42,23 +45,68 @@ export function failover( throw new Error("[failover] requires at least one transport"); } + const facetId = createFacetId(failoverTransportKey); + return (params) => { const requestFns = transports.map((t) => t(params).request); const request = async (args: EIP1193Parameters) => { - let lastErr: unknown; - for (const requestFn of requestFns) { - try { - return await requestFn(args); - } catch (err) { - // Ahead of `shouldThrow` and not overridable: falling over discards the payload. - if (isTerminalError(err) || shouldThrow(err)) throw err; - lastErr = err; + const facet = getObservability()?.facet(facetId); + const stats: { + branchErrors: unknown[]; + branchDurationsMs: number[]; + succeededIndex: number; + terminatedByShouldThrow: boolean; + terminatedByTerminalError: boolean; + } = { + branchErrors: [], + branchDurationsMs: [], + succeededIndex: -1, + terminatedByShouldThrow: false, + terminatedByTerminalError: false, + }; + + try { + for (let i = 0; i < requestFns.length; i++) { + const t0 = performance.now(); + try { + const result = await requestFns[i]!(args); + stats.branchDurationsMs.push(performance.now() - t0); + stats.succeededIndex = i; + return result; + } catch (err) { + stats.branchDurationsMs.push(performance.now() - t0); + stats.branchErrors.push(err); + // Ahead of `shouldThrow` and not overridable: falling over discards the payload. + if (isTerminalError(err)) { + stats.terminatedByTerminalError = true; + throw err; + } + if (shouldThrow(err)) { + stats.terminatedByShouldThrow = true; + throw err; + } + } + } + throw stats.branchErrors.at(-1); + } finally { + facet?.set({ + branches_attempted: stats.branchErrors.length + (stats.succeededIndex >= 0 ? 1 : 0), + succeeded_index: stats.succeededIndex, + branch_durations_ms: stats.branchDurationsMs, + }); + if (stats.branchErrors.length > 0) { + facet?.set({ + branch_errors: stats.branchErrors.map(serializeError), + terminated_by_should_throw: stats.terminatedByShouldThrow, + terminated_by_terminal_error: stats.terminatedByTerminalError, + }); } } - throw lastErr; }; + const observed = observe(request, facetId) as EIP1193RequestFn; + // Bypass `createTransport` so we don't add a redundant `buildRequest` layer. // Failover doesn't classify errors, retry, or dedupe — wrapping here would only // re-wrap errors already classified by each branch's own `buildRequest`. @@ -67,10 +115,10 @@ export function failover( key: failoverTransportKey, name: "[viem-dlc] failover", type: failoverTransportKey, - request: request as EIP1193RequestFn, + request: observed, retryCount: 0, }, - request: request as EIP1193RequestFn, + request: observed, }; }; } diff --git a/src/transports/logs-divider/handlers.ts b/src/transports/logs-divider/handlers.ts index f7028a4..d2a9c60 100644 --- a/src/transports/logs-divider/handlers.ts +++ b/src/transports/logs-divider/handlers.ts @@ -1,5 +1,6 @@ import { type EIP1193RequestFn, hexToBigInt, type RpcLog, toHex } from "viem"; +import { type Facet, type FacetId, getObservability } from "../../observability.js"; import type { BlockRange, EthGetLogsHashlessFilter, RpcSignature } from "../../types.js"; import { augment } from "../../utils/arrays.js"; import { @@ -9,6 +10,7 @@ import { isInBlockRange, resolveBlockNumber, } from "../../utils/blocks.js"; +import { serializeError } from "../../utils/errors.js"; import { min } from "../../utils/math.js"; import type { RateLimiterSchema } from "../rate-limiter/schema.js"; @@ -22,6 +24,14 @@ interface ProcessContext { onLogsResponseOnly?: boolean; baseFilter: EthGetLogsHashlessFilter; latestBlockNumber: bigint; + facet?: Facet; + stats: { + logsFetched: number; + /** Halving stats, surfaced as flat `splits_*` fields on the terminal wide event. */ + splits: { count: number; range: number; timeout: number; maxDepth: number }; + /** Counts of leaf `requestFn` durations (success or failure), keyed by 100ms-bin lower bound in ms. */ + fetchDurationsMs: Record; + }; } /** Fetches logs for a single range with automatic retry and range halving on range-related failure. */ @@ -30,7 +40,10 @@ async function fetchRangeWithRetry( range: BlockRange, priority?: number, timeoutSplitsRemaining = 1, + depth = 0, ): Promise { + if (depth > ctx.stats.splits.maxDepth) ctx.stats.splits.maxDepth = depth; + // Constrain toBlock to chain tip (range may span past it due to alignment) const constrainedRange: BlockRange = { fromBlock: range.fromBlock, @@ -49,14 +62,21 @@ async function fetchRangeWithRetry( }; try { - const logs = await ctx.requestFn( - { - method: "eth_getLogs", - params: [filter, { __rateLimiter: true, priority }], - }, - // `retryCount: 0` so that we fail fast on block range errors - { retryCount: 0 }, - ); + let logs: RpcLog[]; + const t0 = performance.now(); + try { + logs = await ctx.requestFn( + { + method: "eth_getLogs", + params: [filter, { __rateLimiter: true, priority }], + }, + // `retryCount: 0` so that we fail fast on block range errors + { retryCount: 0 }, + ); + } finally { + const bin = Math.floor((performance.now() - t0) / 100) * 100; + ctx.stats.fetchDurationsMs[bin] = (ctx.stats.fetchDurationsMs[bin] ?? 0) + 1; + } // Success - invoke callback ctx.onLogsResponse?.({ @@ -66,6 +86,7 @@ async function fetchRangeWithRetry( fetchedAtBlock: ctx.latestBlockNumber, fetchedAt: Date.now(), }); + ctx.stats.logsFetched += logs.length; return ctx.onLogsResponseOnly ? [] : logs; } catch (error) { @@ -76,18 +97,25 @@ async function fetchRangeWithRetry( if (halves) { const nextBudget = cause === "timeout" ? timeoutSplitsRemaining - 1 : timeoutSplitsRemaining; - const logs = await Promise.all(halves.map((half) => fetchRangeWithRetry(ctx, half, priority, nextBudget))); + ctx.stats.splits.count += 1; + ctx.stats.splits[cause] += 1; + + const logs = await Promise.all( + halves.map((half) => fetchRangeWithRetry(ctx, half, priority, nextBudget, depth + 1)), + ); return ctx.onLogsResponseOnly ? [] : logs.flat(); } } - // Add range context to non-range errors for easier debugging - const rangeContext = `[fetchRangeWithRetry [${range.fromBlock}n, ${range.toBlock}n]]`; - if (error instanceof Error) { - error.message = `${rangeContext} ${error.message}`; - throw error; - } - throw new Error(`${rangeContext} ${String(error)}`); + // Record on the wide event rather than emitting a separate log entry. `Promise.all` + // upstream surfaces only the first rejection, so this bounded list is the only + // record of sibling chunks that failed in parallel. + ctx.facet?.push("failed_ranges", { + from_block: Number(range.fromBlock), + to_block: Number(range.toBlock), + error: serializeError(error), + }); + throw error; } } @@ -99,12 +127,15 @@ export async function handleEthGetLogs( requestFn: EIP1193RequestFn, [filter, ...params]: RpcSignature["Parameters"], config: LogsDividerConfig, + facetId: FacetId, ): Promise { // blockHash queries cannot be divided - pass through if (filter.blockHash) { return requestFn({ method: "eth_getLogs", params: params[0] ? [filter, params[0]] : [filter] }); } + const facet = getObservability()?.facet(facetId); + // Get extra params const priority = params[0]?.priority ?? 0; const latestBlockNumber = hexToBigInt(params[1]?.latestBlock ?? (await requestFn({ method: "eth_blockNumber" }))); @@ -114,6 +145,7 @@ export async function handleEthGetLogs( const toBlock = min(resolveBlockNumber(filter.toBlock ?? "latest", latestBlockNumber), latestBlockNumber); if (fromBlock > toBlock) { + facet?.set({ short_circuit: "empty_range" }); return []; } @@ -123,25 +155,49 @@ export async function handleEthGetLogs( onLogsResponseOnly: params[1]?.onLogsResponseOnly, baseFilter: filter, latestBlockNumber, + facet, + stats: { + logsFetched: 0, + splits: { count: 0, range: 0, timeout: 0, maxDepth: 0 }, + fetchDurationsMs: {}, + }, }; const range: BlockRange = { fromBlock, toBlock }; const chunks = divideBlockRange(range, config.maxBlockRange, config.alignTo); - const logs = await augment(chunks).mapAsync( - async (chunk, i) => { - // Take chunks to be [A, B, ..., Z] -- if we make requests without specifying priority, the queue - // is FIFO, so *retries* for chunk A are queued after the *initial* request for chunk Z. This isn't - // a problem here, since we need to fetch all ranges anyway, but it can produce unexpected - // mental-model-overhead for `onLogsResponse` consumers. By using the chunk index as the priority, - // we ensure that *if we're rate/concurrency limited*, chunks are processed roughly in order. - const result = await fetchRangeWithRetry(ctx, chunk, priority + i / chunks.length); - // Filter out logs outside original range (in case alignment extended the range). - // We do this per-chunk to avoid creating an extra copy of the final flattened array, which could be large. - return result.filter(isInBlockRange(range)); - }, - // NOTE: Defensive upper bound to avoid flooding EventLoop. Request concurrency is managed by `rateLimiter`. - { maxConcurrent: 1000 }, - ); - return logs.flat(); + facet?.set({ + from_block: Number(fromBlock), + to_block: Number(toBlock), + latest_block: Number(latestBlockNumber), + nominal_ranges: chunks.length, + }); + + try { + const logs = await augment(chunks).mapAsync( + async (chunk, i) => { + // Take chunks to be [A, B, ..., Z] -- if we make requests without specifying priority, the queue + // is FIFO, so *retries* for chunk A are queued after the *initial* request for chunk Z. This isn't + // a problem here, since we need to fetch all ranges anyway, but it can produce unexpected + // mental-model-overhead for `onLogsResponse` consumers. By using the chunk index as the priority, + // we ensure that *if we're rate/concurrency limited*, chunks are processed roughly in order. + const result = await fetchRangeWithRetry(ctx, chunk, priority + i / chunks.length); + // Filter out logs outside original range (in case alignment extended the range). + // We do this per-chunk to avoid creating an extra copy of the final flattened array, which could be large. + return result.filter(isInBlockRange(range)); + }, + // NOTE: Defensive upper bound to avoid flooding EventLoop. Request concurrency is managed by `rateLimiter`. + { maxConcurrent: 1000 }, + ); + return logs.flat(); + } finally { + facet?.set({ + logs_fetched: ctx.stats.logsFetched, + fetch_durations_ms: ctx.stats.fetchDurationsMs, + splits_count: ctx.stats.splits.count, + splits_range: ctx.stats.splits.range, + splits_timeout: ctx.stats.splits.timeout, + splits_max_depth: ctx.stats.splits.maxDepth, + }); + } } diff --git a/src/transports/logs-divider/index.ts b/src/transports/logs-divider/index.ts index c21dccb..653bcb6 100644 --- a/src/transports/logs-divider/index.ts +++ b/src/transports/logs-divider/index.ts @@ -1,5 +1,6 @@ import { createTransport, type EIP1193RequestFn, type PublicRpcSchema, type Transport } from "viem"; +import { createFacetId, observe } from "../../observability.js"; import type { EIP1193Parameters } from "../../types.js"; import { isRevertExpected } from "../../utils/deployless/codec.envelope.js"; import { logsEnricher } from "../logs-enricher/index.js"; @@ -8,13 +9,12 @@ import { type LogsSieveConfig, logsSieve } from "../logs-sieve/index.js"; import { type RateLimiterConfig, rateLimiter } from "../rate-limiter/index.js"; import { handleEthGetLogs } from "./handlers.js"; -import type { LogsDividerSchema } from "./schema.js"; +import { type LogsDividerSchema, logsDividerTransportKey } from "./schema.js"; import type { LogsDividerConfig } from "./types.js"; export type * from "./schema.js"; export type * from "./types.js"; - -export const logsDividerTransportKey = "viem-dlc-logs-divider" as const; +export { logsDividerTransportKey }; /** * Creates a transport wrapper that divides large eth_getLogs requests into smaller chunks. @@ -70,23 +70,25 @@ export function logsDivider( throw new Error(`[logsDivider] maxBlockRange must be >= 1 (got ${logsDividerConfig.maxBlockRange})`); } + const facetId = createFacetId(logsDividerTransportKey); + return (params) => { const transport = logsEnricher(logsSieve(rateLimiter(baseTransportFn, [rateLimiterConfig]), [logsSieveConfig]), [ logsEnricherConfig, ])(params); - const request = (args: EIP1193Parameters) => { - if (args.method !== "eth_getLogs") { - return transport.request(args, isRevertExpected(args) ? { retryCount: 0 } : undefined); + const request = (req: EIP1193Parameters) => { + if (req.method !== "eth_getLogs") { + return transport.request(req, isRevertExpected(req) ? { retryCount: 0 } : undefined); } - return handleEthGetLogs(transport.request, args.params, logsDividerConfig); + return handleEthGetLogs(transport.request, req.params, logsDividerConfig, facetId); }; return createTransport({ key: logsDividerTransportKey, name: "[viem-dlc] logs-divider", - request: request as EIP1193RequestFn, + request: observe(request, facetId) as EIP1193RequestFn, retryCount: 0, type: logsDividerTransportKey, }); diff --git a/src/transports/logs-divider/schema.ts b/src/transports/logs-divider/schema.ts index 40ebf13..5765dd1 100644 --- a/src/transports/logs-divider/schema.ts +++ b/src/transports/logs-divider/schema.ts @@ -23,3 +23,5 @@ export type LogsDividerSchema = SafelyExtendRpcSchema< }, ] >; + +export const logsDividerTransportKey = "viem-dlc-logs-divider" as const; diff --git a/src/transports/logs-enricher/index.ts b/src/transports/logs-enricher/index.ts index 6975a7b..a459085 100644 --- a/src/transports/logs-enricher/index.ts +++ b/src/transports/logs-enricher/index.ts @@ -1,5 +1,6 @@ import { createTransport, type EIP1193RequestFn, type Hex, type PublicRpcSchema, type Transport } from "viem"; +import { createFacetId, getObservability, observe } from "../../observability.js"; import type { EIP1193Parameters, SafelyExtendedRpcSchema } from "../../types.js"; import { isRevertExpected } from "../../utils/deployless/codec.envelope.js"; @@ -16,6 +17,8 @@ export function logsEnricher( baseTransportFn: Transport>, [{ retryCount, retryDelay, blockTimestamp }]: [LogsEnricherConfig], ): Transport> { + const facetId = createFacetId(logsEnricherTransportKey); + return (params) => { const requestFn = baseTransportFn(params).request as EIP1193RequestFn; @@ -24,6 +27,9 @@ export function logsEnricher( return requestFn(args, isRevertExpected(args) ? { retryCount: 0 } : undefined); } + // Crossed once per chunk under a divider fan-out; `add` accumulates the + // per-call totals on this transport's slot. + const facet = getObservability()?.facet(facetId); const logs = await requestFn(args as EIP1193Parameters); if (!blockTimestamp) return logs; @@ -51,7 +57,7 @@ export function logsEnricher( ); // Enrich logs, dropping any whose block was reorged away - return logs.reduce((acc, log) => { + const enriched = logs.reduce((acc, log) => { if (log.blockTimestamp !== undefined || log.blockNumber === null) { acc.push(log); return acc; @@ -63,12 +69,17 @@ export function logsEnricher( } return acc; }, []); + + facet?.add("blocks_fetched", blockNumbers.size); + facet?.add("logs_dropped", logs.length - enriched.length); + + return enriched; }; return createTransport({ key: logsEnricherTransportKey, name: "[viem-dlc] logs-enricher", - request: request as EIP1193RequestFn, + request: observe(request, facetId) as EIP1193RequestFn, retryCount: 0, type: logsEnricherTransportKey, }); diff --git a/src/transports/logs-sieve/index.ts b/src/transports/logs-sieve/index.ts index 7853c4f..2b82e73 100644 --- a/src/transports/logs-sieve/index.ts +++ b/src/transports/logs-sieve/index.ts @@ -1,5 +1,6 @@ import { createTransport, type EIP1193RequestFn, type PublicRpcSchema, type Transport } from "viem"; +import { createFacetId, getObservability, observe } from "../../observability.js"; import type { EIP1193Parameters, SafelyExtendedRpcSchema } from "../../types.js"; import { isRevertExpected } from "../../utils/deployless/codec.envelope.js"; import { estimateUtf8Bytes } from "../../utils/json.js"; @@ -26,6 +27,8 @@ export function logsSieve( throw new Error(`[logsSieve] maxBytes must be a safe integer >= 1 (got ${maxBytes})`); } + const facetId = createFacetId(logsSieveTransportKey); + return (params) => { const requestFn = baseTransportFn(params).request as EIP1193RequestFn; @@ -34,14 +37,29 @@ export function logsSieve( return requestFn(args, isRevertExpected(args) ? { retryCount: 0 } : undefined); } + // Crossed once per chunk under a divider fan-out, so `add`/`stat` accumulate + // per-call totals on this transport's slot. + const facet = getObservability()?.facet(facetId); const logs = await requestFn(args as EIP1193Parameters); - return logs.filter((log) => estimateUtf8Bytes(log) <= maxBytes); + const kept = logs.filter((log) => { + const bytes = estimateUtf8Bytes(log); + if (bytes <= maxBytes) return true; + // Sizes of the dropped logs, so `maxBytes` can be tuned against what it rejects. + facet?.stat("dropped_log_bytes", bytes); + return false; + }); + + if (kept.length < logs.length) { + facet?.add("logs_dropped", logs.length - kept.length); + } + + return kept; }; return createTransport({ key: logsSieveTransportKey, name: "[viem-dlc] logs-sieve", - request: request as EIP1193RequestFn, + request: observe(request, facetId) as EIP1193RequestFn, retryCount: 0, type: logsSieveTransportKey, }); diff --git a/src/transports/rate-limiter/index.ts b/src/transports/rate-limiter/index.ts index 3f852b8..e364520 100644 --- a/src/transports/rate-limiter/index.ts +++ b/src/transports/rate-limiter/index.ts @@ -1,5 +1,6 @@ import { createTransport, type EIP1193RequestFn, type PublicRpcSchema, type Transport } from "viem"; +import { createFacetId, getObservability, observe } from "../../observability.js"; import type { EIP1193Parameters } from "../../types.js"; import { isRevertExpected } from "../../utils/deployless/codec.envelope.js"; import { hash } from "../../utils/hash.js"; @@ -42,6 +43,8 @@ export function rateLimiter( RateLimiterConfig, ], ): Transport> { + const facetId = createFacetId(rateLimiterTransportKey); + return (params) => { const transport = baseTransportFn(params); const { withRateLimit } = createRateLimit(maxBurstRequests, maxRequestsPerSecond, maxConcurrentRequests); @@ -49,9 +52,14 @@ export function rateLimiter( const request = (req: EIP1193Parameters) => { const [baseReq, additional] = stripAdditionalParameters(req); + // Captured here rather than looked up inside `onAdmitted`, so the sample lands on + // this call's slot no matter which job's completion drained the queue. Crossed once + // per chunk under a divider fan-out, so `stat` summarizes the whole call's waiting. + const facet = getObservability()?.facet(facetId); const inner = () => withRateLimit(() => transport.request(baseReq, isRevertExpected(baseReq) ? { retryCount: 0 } : undefined), { priority: additional?.[0].priority, + onAdmitted: (waitMs) => facet?.stat("queue_wait_ms", waitMs), }); return dedupe ? withDedupe(inner, { key: hash(baseReq) }) : inner(); @@ -60,7 +68,7 @@ export function rateLimiter( return createTransport({ key: rateLimiterTransportKey, name: "[viem-dlc] rate-limiter", - request: request as EIP1193RequestFn, + request: observe(request, facetId) as EIP1193RequestFn, retryCount: 0, type: rateLimiterTransportKey, }); diff --git a/src/utils/deployless/call.ts b/src/utils/deployless/call.ts index a4e8eca..53f7a4d 100644 --- a/src/utils/deployless/call.ts +++ b/src/utils/deployless/call.ts @@ -1,5 +1,6 @@ import { BaseError, type EIP1193RequestFn, type Hex, type PublicRpcSchema } from "viem"; +import type { Facet } from "../../observability.js"; import type { EIP1193Parameters } from "../../types.js"; import { isTimeoutLikeError } from "../errors.js"; import type { Tail } from "../tuples.js"; @@ -40,6 +41,7 @@ type FactorisedFactoryCallParams = { * awaited — so a caller's results survive a later chunk failing. */ onResolved?: (entries: readonly ResolvedElement[]) => void | Promise; + facet?: Facet; }; /** An input element's index paired with the raw output bytes fetched for it. */ @@ -59,7 +61,7 @@ type MeasureBytes = (start: number, end: number) => number; */ export async function factorisedFactoryCall( requestFn: EIP1193RequestFn, - { target, elements, solidity, batch, gasLimit, restOfEthCallParams, onResolved }: FactorisedFactoryCallParams, + { target, elements, solidity, batch, gasLimit, restOfEthCallParams, onResolved, facet }: FactorisedFactoryCallParams, ): Promise { const compress = batch?.compress ?? false; const wrap = (els: readonly Hex[]): Hex => @@ -120,8 +122,20 @@ export async function factorisedFactoryCall( }); const outputs = new Array(elements.length); + facet?.set({ elements_requested: elements.length, nominal_batches: ranges.length }); + // Sizes of the *initial* packing, to compare realized utilization against `batchSize`. + // Bisected children and paged continuations are not resampled. Guarded rather than + // `facet?.stat(...)` so unobserved calls skip re-measuring. + if (facet) for (const [start, end] of ranges) facet.stat("batch_bytes", measureBytes(start, end)); + let fetched = 0; + const splits = { count: 0, size: 0, timeout: 0, maxDepth: 0 }; + // Paged lenses stop early instead of failing, so continuations are counted apart from + // `splits_*`: those mean a chunk was too big, these mean the lens served what it could. + const pages = { continued: 0, waves: 0 }; + const commit = async (entries: readonly ResolvedElement[]) => { for (const { index, output } of entries) outputs[index] = output; + fetched += entries.length; if (entries.length > 0) await onResolved?.(entries); }; @@ -142,7 +156,9 @@ export async function factorisedFactoryCall( nextWave: BatchRange[], precomputed?: Hex, timeoutSplitsRemaining = 1, + depth = 0, ): Promise => { + if (depth > splits.maxDepth) splits.maxDepth = depth; const count = end - start; const wrapped = precomputed ?? wrap(elements.slice(start, end)); @@ -154,10 +170,12 @@ export async function factorisedFactoryCall( const cause = classifyBatchSizeError(e); if (cause === "size" || (cause === "timeout" && timeoutSplitsRemaining > 0)) { const nextBudget = cause === "timeout" ? timeoutSplitsRemaining - 1 : timeoutSplitsRemaining; + splits.count += 1; + splits[cause] += 1; const mid = start + Math.floor(count / 2); return settleAll([ - fetchRecursive([start, mid], nextWave, undefined, nextBudget), - fetchRecursive([mid, end], nextWave, undefined, nextBudget), + fetchRecursive([start, mid], nextWave, undefined, nextBudget, depth + 1), + fetchRecursive([mid, end], nextWave, undefined, nextBudget, depth + 1), ]); } } else if (solidity.paged && isOutOfGasRevert(e)) { @@ -187,17 +205,39 @@ export async function factorisedFactoryCall( } await commit(entries); - if (attempted < count) nextWave.push(...packRange(start + attempted, end, attempted)); + if (attempted < count) { + pages.continued += 1; + nextWave.push(...packRange(start + attempted, end, attempted)); + } }; let wave: BatchRange[] = ranges; - while (wave.length > 0) { - const nextWave: BatchRange[] = []; - const isWholeInput = wave.length === 1 && wave[0]![0] === 0 && wave[0]![1] === elements.length; - await settleAll( - wave.map((range) => fetchRecursive(range, nextWave, isWholeInput ? getReferenceWrapped() : undefined)), - ); - wave = nextWave; + try { + while (wave.length > 0) { + pages.waves += 1; + const nextWave: BatchRange[] = []; + const isWholeInput = wave.length === 1 && wave[0]![0] === 0 && wave[0]![1] === elements.length; + await settleAll( + wave.map((range) => fetchRecursive(range, nextWave, isWholeInput ? getReferenceWrapped() : undefined)), + ); + wave = nextWave; + } + } finally { + facet?.set({ + elements_fetched: fetched, + splits_count: splits.count, + splits_size: splits.size, + splits_timeout: splits.timeout, + splits_max_depth: splits.maxDepth, + }); + // Paged-only, so their presence is what distinguishes a paged run from an ordinary one. + if (solidity.paged) { + facet?.set({ + pages_continued: pages.continued, + pages_waves: pages.waves, + elements_missing: missing.length, + }); + } } if (missing.length > 0) { diff --git a/src/utils/errors.ts b/src/utils/errors.ts index 8577823..0490d09 100644 --- a/src/utils/errors.ts +++ b/src/utils/errors.ts @@ -8,6 +8,16 @@ export function tryCatch(fn: () => T) { } } +export function serializeError(e: unknown) { + if (!(e instanceof Error)) return { value: String(e) }; + return { + name: e.name, + message: e.message, + code: (e as { code?: unknown }).code, + data: (e as { data?: unknown }).data, + }; +} + /** * Detects timeout-shaped errors anywhere in the BaseError cause chain: viem's TimeoutError, * HTTP 408 / 504 / 524, and "timed out" / "timeout" messages. diff --git a/src/utils/index.ts b/src/utils/index.ts index 213dcbf..6ca4eec 100644 --- a/src/utils/index.ts +++ b/src/utils/index.ts @@ -12,6 +12,7 @@ export * from "./math.js"; export * from "./objects.js"; export * from "./omit.js"; export * from "./pick.js"; +export * from "./sleep.js"; export * from "./strings.js"; export * from "./tuples.js"; export * from "./with-dedupe.js"; diff --git a/src/utils/with-rate-limit.ts b/src/utils/with-rate-limit.ts index a2b2dcf..757dc12 100644 --- a/src/utils/with-rate-limit.ts +++ b/src/utils/with-rate-limit.ts @@ -227,6 +227,11 @@ export function createRateLimit(maxTokens: number, refillRate: number, maxConcur * Jobs with the same priority are processed FIFO. * Use `gate` to defer decision on whether to run job (doesn't consume tokens if false). * + * `onAdmitted` reports how long this call waited for admission. It runs after this + * caller's own `await`, so it observes the caller's async context — not that of + * whichever job's completion happened to drain the queue. It is skipped entirely + * when admission is denied by `gate`. + * * @example * const { withRateLimit } = createRateLimit(5, 10, 2) // 5 burst, 10/sec refill, 2 concurrent * @@ -240,8 +245,13 @@ export function createRateLimit(maxTokens: number, refillRate: number, maxConcur */ async withRateLimit( fn: () => Promise, - { priority = Infinity, gate }: { priority?: number; gate?: () => boolean }, + { + priority = Infinity, + gate, + onAdmitted, + }: { priority?: number; gate?: () => boolean; onAdmitted?: (waitMs: number) => void }, ): Promise { + const queuedAt = performance.now(); await new Promise((resolve, reject) => { ctx.queue.push({ resolve, @@ -252,6 +262,7 @@ export function createRateLimit(maxTokens: number, refillRate: number, maxConcur }); drainQueue(ctx); }); + onAdmitted?.(performance.now() - queuedAt); try { return await fn(); diff --git a/test/helpers/logger.ts b/test/helpers/logger.ts new file mode 100644 index 0000000..d27c0c9 --- /dev/null +++ b/test/helpers/logger.ts @@ -0,0 +1,66 @@ +export interface StubEvent { + name: string; + context: Record; + metadata: Record; + error?: unknown; +} + +/** + * Hand-rolled LogLayer stub. Implements just enough of the API for the library + * to call into without importing the real package. Each method returns `this` + * so chaining works; emissions are captured into `events`. + */ +export function createStubLogger() { + const events: StubEvent[] = []; + + // biome-ignore lint/suspicious/noExplicitAny: stub LogLayer surface + function makeLayer(parentContext: Record): any { + let context = { ...parentContext }; + let pendingMetadata: Record = {}; + let pendingError: unknown; + + const emit = (name: string) => { + events.push({ name, context: { ...context }, metadata: pendingMetadata, error: pendingError }); + pendingMetadata = {}; + pendingError = undefined; + return layer; + }; + + const layer = { + child() { + return makeLayer(context); + }, + withContext(extra: Record) { + context = { ...context, ...extra }; + return layer; + }, + withMetadata(extra: Record) { + pendingMetadata = { ...pendingMetadata, ...extra }; + return layer; + }, + withError(err: unknown) { + pendingError = err; + return layer; + }, + info: emit, + warn: emit, + error: emit, + metadataOnly(extra: Record) { + pendingMetadata = { ...pendingMetadata, ...extra }; + emit(""); + }, + }; + return layer; + } + + return { logger: makeLayer({}), events }; +} + +/** + * Reads the `.` entry in a wide event's context. The first + * instance of a key touched in a call writes under the bare key, which is all + * these tests exercise. `field` may be dotted (e.g. "eth_call.input_elements"). + */ +export function findDotted(context: Record, transportKey: string, field: string) { + return context[`${transportKey}.${field}`]; +} diff --git a/test/observability.test.ts b/test/observability.test.ts new file mode 100644 index 0000000..223243d --- /dev/null +++ b/test/observability.test.ts @@ -0,0 +1,480 @@ +import { custom } from "viem"; +import { describe, expect, it, vi } from "vitest"; + +import { createFacetId, getObservability, observe, withLogging } from "../src/observability.js"; +import { failover } from "../src/transports/failover/index.js"; +import { handleEthGetLogs } from "../src/transports/logs-divider/handlers.js"; +import { logsDividerTransportKey } from "../src/transports/logs-divider/schema.js"; +import { logsSieve } from "../src/transports/logs-sieve/index.js"; +import { rateLimiter } from "../src/transports/rate-limiter/index.js"; + +import { createStubLogger, findDotted } from "./helpers/logger.js"; + +describe("observability", () => { + /** Stands in for a transport's own facet id at the outermost boundary. */ + const ROOT = createFacetId("test-root"); + const T = createFacetId("t"); + + describe("withLogging + observe", () => { + it("derives a per-call child carrying seeded context, library, call_id, req, and emits one wide event", async () => { + const { logger, events } = createStubLogger(); + + const observed = observe(async (req: { method: string }) => { + const obs = getObservability(); + expect(obs?.logger).toBeDefined(); + expect(obs?.call_id).toMatch(/^[0-9a-f-]{36}$/); + expect(typeof obs?.facet).toBe("function"); + return `ok:${req.method}`; + }, ROOT); + + const result = await withLogging(() => observed({ method: "eth_blockNumber" }), { + logger, + request_id: "rq-1", + tag: "user-op", + }); + expect(result).toBe("ok:eth_blockNumber"); + + expect(events).toHaveLength(1); + expect(events[0]!.name).toBe("concluded"); + expect(events[0]!.context).toMatchObject({ + library: "viem-dlc", + request_id: "rq-1", + tag: "user-op", + req: { method: "eth_blockNumber" }, + }); + expect(events[0]!.context.call_id).toMatch(/^[0-9a-f-]{36}$/); + expect(events[0]!.context.status).toBe("ok"); + expect(typeof events[0]!.context.duration_ms).toBe("number"); + }); + + it("emits exactly one error-status wide event when the wrapped fn rejects", async () => { + const { logger, events } = createStubLogger(); + + const observed = observe(async () => { + throw new Error("boom"); + }, ROOT); + + await expect(withLogging(() => observed({}), { logger })).rejects.toThrow("boom"); + + expect(events).toHaveLength(1); + expect(events[0]!.name).toBe("concluded"); + expect(events[0]!.context.status).toBe("error"); + expect(events[0]!.error).toBeInstanceOf(Error); + expect(typeof events[0]!.context.duration_ms).toBe("number"); + }); + + it("nested observe boundaries reuse the per-call logger and call_id", async () => { + const { logger, events } = createStubLogger(); + + let outer: ReturnType; + let inner: ReturnType; + + const innerObserved = observe(async () => { + inner = getObservability(); + }, createFacetId("inner")); + const outerObserved = observe(async () => { + outer = getObservability(); + await innerObserved({}); + await innerObserved({}); + }, ROOT); + + await withLogging(() => outerObserved({ method: "eth_getLogs" }), { logger }); + + expect(inner?.call_id).toBe(outer?.call_id); + // Same LogLayer instance, so every layer's facet writes accumulate onto one wide event. + expect(inner?.logger).toBe(outer?.logger); + + // Only the outermost boundary emits. + expect(events).toHaveLength(1); + expect(events[0]!.name).toBe("concluded"); + }); + + it("parallel calls inside one withLogging scope get distinct call_ids and isolated contexts", async () => { + const { logger, events } = createStubLogger(); + + const observed = observe(async (req: { which: string }) => { + getObservability()?.facet(ROOT).set({ which: req.which }); + }, ROOT); + + await withLogging(() => Promise.all([observed({ which: "a" }), observed({ which: "b" })]), { logger }); + + expect(events).toHaveLength(2); + expect(events.map((e) => e.name)).toEqual(["concluded", "concluded"]); + expect(events[0]!.context.call_id).not.toBe(events[1]!.context.call_id); + expect(new Set(events.map((e) => e.context["test-root.which"]))).toEqual(new Set(["a", "b"])); + }); + + it("parallel withLogging scopes do not bleed context", async () => { + const { logger: a, events: aEvents } = createStubLogger(); + const { logger: b, events: bEvents } = createStubLogger(); + + const observed = observe(async () => {}, ROOT); + + await Promise.all([ + withLogging(() => observed({ method: "eth_blockNumber" }), { logger: a, request_id: "A" }), + withLogging(() => observed({ method: "eth_chainId" }), { logger: b, request_id: "B" }), + ]); + + expect(aEvents).toHaveLength(1); + expect(bEvents).toHaveLength(1); + expect(aEvents[0]!.context.request_id).toBe("A"); + expect(bEvents[0]!.context.request_id).toBe("B"); + expect(aEvents[0]!.context.req).toEqual({ method: "eth_blockNumber" }); + expect(bEvents[0]!.context.req).toEqual({ method: "eth_chainId" }); + }); + + it("trims long strings in the per-call `req` context", async () => { + const { logger, events } = createStubLogger(); + const long = "0x".concat("ab".repeat(200)); + const short = "0x1234"; + + const observed = observe(async (_req: unknown) => {}, ROOT); + await withLogging(() => observed({ method: "eth_call", params: [{ data: long, to: short }] }), { logger }); + + const req = events[0]!.context.req as { params: [{ data: string; to: string }] }; + expect(req.params[0].data).toHaveLength(100); + expect(req.params[0].data).toBe(long.slice(0, 97).concat("...")); + expect(req.params[0].to).toBe(short); + }); + + it("shares one AsyncLocalStorage across scopes that race the cold-start import", async () => { + // Fresh module registry, so both calls below hit `loadAls` before it resolves. + vi.resetModules(); + const fresh = await import("../src/observability.js"); + const id = fresh.createFacetId("racer"); + const observed = fresh.observe(async () => {}, id); + + const a = createStubLogger(); + const b = createStubLogger(); + await Promise.all([ + fresh.withLogging(() => observed({ method: "eth_blockNumber" }), { logger: a.logger }), + fresh.withLogging(() => observed({ method: "eth_chainId" }), { logger: b.logger }), + ]); + + // If each call built its own storage, only the last-assigned one would be + // visible to `observe` and the other scope would emit nothing. + expect(a.events).toHaveLength(1); + expect(b.events).toHaveLength(1); + }); + + it("emits nothing and reports no observability when no scope is active", async () => { + const { logger, events } = createStubLogger(); + + expect(getObservability()).toBeUndefined(); + + const observed = observe(async (req: { method: string }) => { + expect(getObservability()).toBeUndefined(); + return `ok:${req.method}`; + }, ROOT); + expect(await observed({ method: "eth_blockNumber" })).toBe("ok:eth_blockNumber"); + + // ...and a real transport still works outside a scope. + const transport = failover([custom({ request: vi.fn().mockResolvedValue("0x42") }, { retryCount: 0 })])( + {} as never, + ); + expect(await transport.request({ method: "eth_blockNumber" })).toBe("0x42"); + + expect(events).toHaveLength(0); + // Reference `logger` so it isn't reported unused. + void logger; + }); + }); + + describe("facets", () => { + /** Runs `fn` inside a fresh withLogging + observe scope and returns the wide event's context. */ + async function concluded(fn: () => Promise | void) { + const { logger, events } = createStubLogger(); + const observed = observe(async () => fn(), ROOT); + await withLogging(() => observed({ method: "test" }), { logger }); + expect(events).toHaveLength(1); + return events[0]!.context; + } + + it("labels ids sharing a key in first-touch order, independent across keys", async () => { + const a = createFacetId("alpha"); + const b = createFacetId("alpha"); + const b2 = createFacetId("beta"); + const context = await concluded(() => { + const obs = getObservability()!; + obs.facet(a).set({ x: 1 }); + obs.facet(b).set({ x: 2 }); + // Re-allocating with the same id returns the same slot. + obs.facet(a).set({ y: 3 }); + obs.facet(b2).set({ y: 4 }); + }); + + expect(context["alpha.x"]).toBe(1); + expect(context["alpha.1.x"]).toBe(2); + expect(context["alpha.y"]).toBe(3); + // Each key labels independently: beta's first id also gets a bare label. + expect(context["beta.y"]).toBe(4); + }); + + it("set merges per field, and writes remain valid after awaits", async () => { + const context = await concluded(async () => { + const facet = getObservability()!.facet(T); + facet.set({ a: 1, b: "old" }); + await Promise.resolve(); + facet.set({ b: "new" }); + }); + + expect(context["t.a"]).toBe(1); + expect(context["t.b"]).toBe("new"); + }); + + it("add accumulates and stat summarizes", async () => { + const context = await concluded(() => { + const facet = getObservability()!.facet(T); + facet.add("hits"); + facet.add("hits", 4); + facet.stat("ms", 10); + facet.stat("ms", 30); + facet.stat("ms", 20); + }); + + expect(context["t.hits"]).toBe(5); + expect(context["t.ms.count"]).toBe(3); + expect(context["t.ms.min"]).toBe(10); + expect(context["t.ms.max"]).toBe(30); + expect(context["t.ms.avg"]).toBe(20); + }); + + it("push bounds the array and counts the overflow", async () => { + const context = await concluded(() => { + const facet = getObservability()!.facet(T); + facet.push("ids", "a", 2); + facet.push("ids", "b", 2); + facet.push("ids", "c", 2); + facet.push("ids", "d", 2); + }); + + expect(context["t.ids"]).toEqual(["a", "b"]); + expect(context["t.ids_truncated"]).toBe(2); + }); + + it("drops oversized fields at conclusion and records them in truncated_fields", async () => { + const context = await concluded(() => { + const facet = getObservability()!.facet(T); + facet.set({ huge: "x".repeat(64 * 1024), small: 1 }); + }); + + expect(context["t.huge"]).toBeUndefined(); + expect(context["t.small"]).toBe(1); + expect(context.truncated_fields).toEqual(["t.huge"]); + }); + + it("a facet that never writes contributes no fields", async () => { + const context = await concluded(() => { + getObservability()!.facet(createFacetId("silent")); + }); + + expect(Object.keys(context).some((k) => k.startsWith("silent."))).toBe(false); + }); + + it("sub scopes field names within the same slot", async () => { + const context = await concluded(() => { + const facet = getObservability()!.facet(T); + facet.sub("eth_call").set({ x: 1 }); + facet.set({ y: 2 }); + }); + + expect(context["t.eth_call.x"]).toBe(1); + expect(context["t.y"]).toBe(2); + }); + + it("repeated facet lookups for one id aggregate on its slot", async () => { + const context = await concluded(() => { + // Simulates a per-chunk layer allocating its facet on every crossing. + getObservability()!.facet(T).add("n", 2); + getObservability()!.facet(T).add("n", 3); + }); + + expect(context["t.n"]).toBe(5); + // No second label was claimed. + expect(Object.keys(context).some((k) => /^t\.\d+\./.test(k))).toBe(false); + }); + + it("observe stamps a crossings count per id", async () => { + const { logger, events } = createStubLogger(); + + const innerA = createFacetId("inner"); + const innerB = createFacetId("inner"); + const innerObservedA = observe(async () => {}, innerA); + const innerObservedB = observe(async () => {}, innerB); + const outerObserved = observe(async () => { + await innerObservedA({}); + await innerObservedA({}); + await innerObservedB({}); + }, createFacetId("outer")); + + await withLogging(() => outerObserved({ method: "eth_getLogs" }), { logger }); + + const { context } = events[0]!; + expect(context["outer.crossings"]).toBe(1); + expect(context["inner.crossings"]).toBe(2); + // A second id sharing that key gets its own first-touch label. + expect(context["inner.1.crossings"]).toBe(1); + }); + }); + + describe("canonical enrichment through real transports", () => { + it("failover stamps branch stats onto the wide event", async () => { + const { logger, events } = createStubLogger(); + const a = vi.fn().mockRejectedValue(new Error("a-failed")); + const b = vi.fn().mockResolvedValue("0x42"); + + const transport = failover([ + custom({ request: a }, { retryCount: 0 }), + custom({ request: b }, { retryCount: 0 }), + ])({} as never); + + const result = await withLogging(() => transport.request({ method: "eth_blockNumber" }), { logger }); + expect(result).toBe("0x42"); + + expect(events).toHaveLength(1); + const { context } = events[0]!; + expect(events[0]!.name).toBe("concluded"); + + const key = "viem-dlc-failover"; + expect(findDotted(context, key, "succeeded_index")).toBe(1); + expect(findDotted(context, key, "branches_attempted")).toBe(2); + expect(findDotted(context, key, "terminated_by_should_throw")).toBe(false); + + const errs = findDotted(context, key, "branch_errors") as { message: string }[]; + expect(errs).toHaveLength(1); + expect(errs[0]!.message).toContain("a-failed"); + + const branchDurations = findDotted(context, key, "branch_durations_ms") as number[]; + expect(branchDurations).toHaveLength(2); + expect(branchDurations.every((d) => typeof d === "number" && d >= 0)).toBe(true); + }); + + it("logs-divider stamps split stats when a range error triggers a halving", async () => { + const { logger, events } = createStubLogger(); + let firstFailure = true; + const requestFn = vi.fn().mockImplementation(async () => { + if (firstFailure) { + firstFailure = false; + throw Object.assign(new Error("query returned more than 10000 results"), { code: -32005 }); + } + return []; + }); + + // Call the handler directly to bypass viem's `buildRequest` retry layers, but wrap it + // in `observe` so it sees an ambient per-call scope (which it reads itself). + const dividerId = createFacetId(logsDividerTransportKey); + const observed = observe( + () => + handleEthGetLogs( + requestFn, + [{ fromBlock: "0x0", toBlock: "0x10" }, undefined, { latestBlock: "0x20" }], + { maxBlockRange: 100, alignTo: 1 }, + dividerId, + ), + dividerId, + ); + + const logs = await withLogging(() => observed({ method: "eth_getLogs" }), { logger }); + expect(logs).toEqual([]); + // 1 failed full-range fetch + 2 successful halves. + expect(requestFn).toHaveBeenCalledTimes(3); + + expect(events).toHaveLength(1); + const { context } = events[0]!; + expect(events[0]!.name).toBe("concluded"); + + const key = "viem-dlc-logs-divider"; + expect(findDotted(context, key, "from_block")).toBe(0); + expect(findDotted(context, key, "to_block")).toBe(0x10); + expect(findDotted(context, key, "latest_block")).toBe(0x20); + expect(findDotted(context, key, "nominal_ranges")).toBe(1); + expect(findDotted(context, key, "logs_fetched")).toBe(0); + expect(findDotted(context, key, "splits_count")).toBe(1); + expect(findDotted(context, key, "splits_range")).toBe(1); + expect(findDotted(context, key, "splits_timeout")).toBe(0); + expect(findDotted(context, key, "splits_max_depth")).toBe(1); + + // Histogram of leaf fetch durations, keyed by 100ms-bin lower bound. + const durations = findDotted(context, key, "fetch_durations_ms") as Record; + expect(Object.values(durations).reduce((a, b) => a + b, 0)).toBe(3); + }); + + it("logs-divider records unhalvable failures in failed_ranges and the event carries error status", async () => { + const { logger, events } = createStubLogger(); + const failure = new Error("connection refused"); + const requestFn = vi.fn().mockRejectedValue(failure); + + const dividerId = createFacetId(logsDividerTransportKey); + const observed = observe( + () => + handleEthGetLogs( + requestFn, + [{ fromBlock: "0x0", toBlock: "0x10" }, undefined, { latestBlock: "0x20" }], + { maxBlockRange: 100, alignTo: 1 }, + dividerId, + ), + dividerId, + ); + + await expect(withLogging(() => observed({ method: "eth_getLogs" }), { logger })).rejects.toThrow( + "connection refused", + ); + + expect(events).toHaveLength(1); + const { context } = events[0]!; + expect(context.status).toBe("error"); + expect(events[0]!.error).toBe(failure); + + const failedRanges = findDotted(context, "viem-dlc-logs-divider", "failed_ranges") as { + from_block: number; + to_block: number; + error: { message: string }; + }[]; + expect(failedRanges).toHaveLength(1); + expect(failedRanges[0]).toMatchObject({ from_block: 0, to_block: 0x10 }); + expect(failedRanges[0]!.error.message).toContain("connection refused"); + }); + + it("logs-sieve summarizes the sizes of the logs it drops", async () => { + const { logger, events } = createStubLogger(); + const small = { address: "0x1", data: "0x" }; + const big = { address: "0x2", data: `0x${"ab".repeat(400)}` }; + const requestFn = vi.fn().mockResolvedValue([small, big, big]); + + const transport = logsSieve(custom({ request: requestFn }, { retryCount: 0 }), [{ maxBytes: 128 }])({} as never); + const kept = await withLogging(() => transport.request({ method: "eth_getLogs", params: [{}] }), { logger }); + expect(kept).toHaveLength(1); + + const { context } = events[0]!; + const key = "viem-dlc-logs-sieve"; + expect(findDotted(context, key, "logs_dropped")).toBe(2); + expect(findDotted(context, key, "dropped_log_bytes.count")).toBe(2); + expect(findDotted(context, key, "dropped_log_bytes.min")).toBeGreaterThan(128); + expect(findDotted(context, key, "dropped_log_bytes.avg")).toBe(findDotted(context, key, "dropped_log_bytes.max")); + }); + + it("rate-limiter summarizes queue wait, attributing each sample to the call that waited", async () => { + const { logger, events } = createStubLogger(); + const requestFn = vi.fn().mockResolvedValue("0x1"); + + // One token, no refill: the first call is admitted immediately and the second + // only after the first releases — so the two calls must record different waits. + const transport = rateLimiter(custom({ request: requestFn }, { retryCount: 0 }), [ + { maxRequestsPerSecond: 1000, maxBurstRequests: 1, maxConcurrentRequests: 1 }, + ])({} as never); + + await Promise.all([ + withLogging(() => transport.request({ method: "eth_blockNumber" }), { logger, tag: "first" }), + withLogging(() => transport.request({ method: "eth_chainId" }), { logger, tag: "second" }), + ]); + + expect(events).toHaveLength(2); + const key = "viem-dlc-rate-limiter"; + for (const event of events) { + // Each event carries exactly its own call's single wait sample, not both. + expect(findDotted(event.context, key, "queue_wait_ms.count")).toBe(1); + expect(findDotted(event.context, key, "queue_wait_ms.max")).toBeGreaterThanOrEqual(0); + } + }); + }); +}); diff --git a/test/stores/lru.test.ts b/test/stores/lru.test.ts index 2281bd6..28cce8b 100644 --- a/test/stores/lru.test.ts +++ b/test/stores/lru.test.ts @@ -4,43 +4,48 @@ import { LruStore } from "../../src/stores/index.js"; describe("LruStore", () => { it("throws if maxBytes is less than 1", () => { - expect(() => new LruStore(0)).toThrow("[LruStore] maxBytes must be at least 1"); - expect(() => new LruStore(-1)).toThrow("[LruStore] maxBytes must be at least 1"); + expect(() => new LruStore({ maxBytes: 0 })).toThrow("maxBytes must be at least 1"); + expect(() => new LruStore({ maxBytes: -1 })).toThrow("maxBytes must be at least 1"); + }); + + it("rejects the superseded positional form rather than silently going unbounded", () => { + // @ts-expect-error deliberately exercising the pre-options `new LruStore(bytes)` call. + expect(() => new LruStore(1000)).toThrow("maxBytes must be at least 1"); }); it("returns null for missing keys", async () => { - const store = new LruStore(1000); + const store = new LruStore({ maxBytes: 1000 }); expect(await store.get("missing")).toBeNull(); }); it("stores and retrieves values", async () => { - const store = new LruStore(1000); + const store = new LruStore({ maxBytes: 1000 }); await store.set("key", [Buffer.from("value")]); expect(await store.get("key")).toEqual([Buffer.from("value")]); }); it("overwrites existing values", async () => { - const store = new LruStore(1000); + const store = new LruStore({ maxBytes: 1000 }); await store.set("key", [Buffer.from("first")]); await store.set("key", [Buffer.from("second")]); expect(await store.get("key")).toEqual([Buffer.from("second")]); }); it("deletes values", async () => { - const store = new LruStore(1000); + const store = new LruStore({ maxBytes: 1000 }); await store.set("key", [Buffer.from("value")]); await store.delete("key"); expect(await store.get("key")).toBeNull(); }); it("handles empty string values", async () => { - const store = new LruStore(1000); + const store = new LruStore({ maxBytes: 1000 }); await store.set("key", [Buffer.from("")]); expect(await store.get("key")).toEqual([Buffer.from("")]); }); it("isolates keys from each other", async () => { - const store = new LruStore(1000); + const store = new LruStore({ maxBytes: 1000 }); await store.set("a", [Buffer.from("1")]); await store.set("b", [Buffer.from("2")]); expect(await store.get("a")).toEqual([Buffer.from("1")]); @@ -52,7 +57,7 @@ describe("LruStore", () => { it("evicts oldest entries when byte limit is exceeded", async () => { // Only value bytes are counted (keys assumed negligible) - const store = new LruStore(2); + const store = new LruStore({ maxBytes: 2 }); await store.set("a", [Buffer.from("1")]); // 1 byte await store.set("b", [Buffer.from("2")]); // 1 byte, total 2 await store.set("c", [Buffer.from("3")]); // 1 byte, would be 3, evicts 'a', total 2 @@ -63,7 +68,7 @@ describe("LruStore", () => { }); it("updates access order on get", async () => { - const store = new LruStore(2); + const store = new LruStore({ maxBytes: 2 }); await store.set("a", [Buffer.from("1")]); // 1 byte await store.set("b", [Buffer.from("2")]); // 1 byte, total 2 @@ -79,7 +84,7 @@ describe("LruStore", () => { }); it("updates access order on set of existing key", async () => { - const store = new LruStore(2); + const store = new LruStore({ maxBytes: 2 }); await store.set("a", [Buffer.from("1")]); // 1 byte await store.set("b", [Buffer.from("2")]); // 1 byte, total 2 @@ -95,7 +100,7 @@ describe("LruStore", () => { }); it("evicts multiple entries if needed for a large value", async () => { - const store = new LruStore(10); + const store = new LruStore({ maxBytes: 10 }); await store.set("a", [Buffer.from("11")]); // 2 bytes await store.set("b", [Buffer.from("22")]); // 2 bytes await store.set("c", [Buffer.from("33")]); // 2 bytes, total 6 @@ -110,13 +115,13 @@ describe("LruStore", () => { }); it("handles deleting non-existent keys", async () => { - const store = new LruStore(1000); + const store = new LruStore({ maxBytes: 1000 }); await store.delete("nonexistent"); expect(await store.get("nonexistent")).toBeNull(); }); it("correctly tracks bytes when updating with different sized values", async () => { - const store = new LruStore(10); + const store = new LruStore({ maxBytes: 10 }); await store.set("a", [Buffer.from("123456789")]); // 9 bytes await store.set("a", [Buffer.from("1")]); // now 1 byte @@ -127,22 +132,22 @@ describe("LruStore", () => { }); it("warns and skips values that exceed maxBytes", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); - const store = new LruStore(5); + const warn = vi.fn(); + const logger = { withMetadata: () => ({ warn }), metadataOnly: () => {} } as unknown as ConstructorParameters< + typeof LruStore + >[0]["logger"]; + const store = new LruStore({ maxBytes: 5, logger }); await store.set("a", [Buffer.from("1")]); // 1 byte, fits await store.set("big", [Buffer.from("123456")]); // 6 bytes, exceeds 5 - expect(warnSpy).toHaveBeenCalledWith("[LruStore] Value exceeds maxBytes (6 > 5), skipping"); + expect(warn).toHaveBeenCalledWith("value exceeds maxBytes, skipping"); expect(await store.get("a")).toEqual([Buffer.from("1")]); // original entry still there expect(await store.get("big")).toBeNull(); // oversized value was not stored - - warnSpy.mockRestore(); }); it("does not evict existing entries when new value exceeds maxBytes", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); - const store = new LruStore(5); + const store = new LruStore({ maxBytes: 5 }); await store.set("a", [Buffer.from("12")]); // 2 bytes await store.set("b", [Buffer.from("34")]); // 2 bytes, total 4 @@ -152,12 +157,10 @@ describe("LruStore", () => { expect(await store.get("a")).toEqual([Buffer.from("12")]); expect(await store.get("b")).toEqual([Buffer.from("34")]); expect(await store.get("huge")).toBeNull(); - - warnSpy.mockRestore(); }); it("evicts other entries when updating existing key with larger value", async () => { - const store = new LruStore(10); + const store = new LruStore({ maxBytes: 10 }); await store.set("a", [Buffer.from("12345")]); // 5 bytes await store.set("b", [Buffer.from("12345")]); // 5 bytes, total 10 diff --git a/test/stores/ttl.test.ts b/test/stores/ttl.test.ts index 7fa1ebf..17b9b29 100644 --- a/test/stores/ttl.test.ts +++ b/test/stores/ttl.test.ts @@ -41,33 +41,33 @@ afterEach(() => { describe("TtlStore", () => { it("throws if ttlMs is not a finite number >= 1", () => { const msg = "[TtlStore] ttlMs must be a finite number >= 1"; - expect(() => new TtlStore(new LruStore(1024), { ttlMs: 0 })).toThrow(msg); - expect(() => new TtlStore(new LruStore(1024), { ttlMs: -1 })).toThrow(msg); - expect(() => new TtlStore(new LruStore(1024), { ttlMs: Number.NaN })).toThrow(msg); - expect(() => new TtlStore(new LruStore(1024), { ttlMs: Number.POSITIVE_INFINITY })).toThrow(msg); + expect(() => new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 0 })).toThrow(msg); + expect(() => new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: -1 })).toThrow(msg); + expect(() => new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: Number.NaN })).toThrow(msg); + expect(() => new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: Number.POSITIVE_INFINITY })).toThrow(msg); }); it("returns null for missing keys", () => { - const store = new TtlStore(new LruStore(1024), { ttlMs: 1000 }); + const store = new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }); expect(store.get("missing")).toBeNull(); }); it("serves a value within ttlMs", () => { - const store = new TtlStore(new LruStore(1024), { ttlMs: 1000 }); + const store = new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }); store.set("k", bytes("v")); vi.advanceTimersByTime(999); expect(store.get("k")).toEqual(bytes("v")); }); it("misses once past ttlMs", () => { - const store = new TtlStore(new LruStore(1024), { ttlMs: 1000 }); + const store = new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }); store.set("k", bytes("v")); vi.advanceTimersByTime(1001); expect(store.get("k")).toBeNull(); }); it("serves at exactly the ttlMs deadline and expires only strictly past it", () => { - const store = new TtlStore(new LruStore(1024), { ttlMs: 1000 }); + const store = new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }); store.set("k", bytes("v")); vi.advanceTimersByTime(1000); // age === ttlMs → still fresh expect(store.get("k")).toEqual(bytes("v")); @@ -76,7 +76,7 @@ describe("TtlStore", () => { }); it("never refreshes the ttl on get (absolute expiry from set)", () => { - const store = new TtlStore(new LruStore(1024), { ttlMs: 1000 }); + const store = new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }); store.set("k", bytes("v")); // A read just before expiry must not extend the entry's life. vi.advanceTimersByTime(999); @@ -86,7 +86,7 @@ describe("TtlStore", () => { }); it("re-stamps the ttl on set", () => { - const store = new TtlStore(new LruStore(1024), { ttlMs: 1000 }); + const store = new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }); store.set("k", bytes("v1")); vi.advanceTimersByTime(600); store.set("k", bytes("v2")); // resets the clock @@ -97,7 +97,7 @@ describe("TtlStore", () => { it("reports a miss past ttlMs without deleting from the wrapped store", () => { // Expiry must NOT issue a delete: with an async store it could race a concurrent write and // clobber a fresh value. Spy is attached after `set` (LruStore.set self-calls delete internally). - const inner = new LruStore(1024); + const inner = new LruStore({ maxBytes: 1024 }); const store = new TtlStore(inner, { ttlMs: 1000 }); store.set("k", bytes("v")); const deleteSpy = vi.spyOn(inner, "delete"); @@ -107,7 +107,7 @@ describe("TtlStore", () => { }); it("returns value buffers by reference over a framing-preserving store (no copy)", () => { - const store = new TtlStore(new LruStore(1024), { ttlMs: 1000 }); + const store = new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }); const a = Buffer.from("ab"); const b = Buffer.from("cd"); store.set("k", [a, b]); @@ -126,7 +126,7 @@ describe("TtlStore", () => { }); it("treats an entry too short to carry the header as a miss", () => { - const inner = new LruStore(1024); + const inner = new LruStore({ maxBytes: 1024 }); inner.set("k", [Buffer.from("short")]); // 5 bytes < header — could not have been written by TtlStore const store = new TtlStore(inner, { ttlMs: 1000 }); expect(store.get("k")).toBeNull(); @@ -136,7 +136,7 @@ describe("TtlStore", () => { // A value written to the wrapped store by something other than TtlStore (a pre-existing entry, or // another consumer sharing the store). Long enough to look header-sized, but lacks the magic — so // it must NOT be served as a stamped value with its leading bytes stripped. - const inner = new LruStore(1024); + const inner = new LruStore({ maxBytes: 1024 }); const foreign = Buffer.from("a plain value that was never written through the TtlStore wrapper"); inner.set("k", [foreign]); const store = new TtlStore(inner, { ttlMs: 1000 }); @@ -144,7 +144,7 @@ describe("TtlStore", () => { }); it("delete removes the entry and flush does not throw", () => { - const store = new TtlStore(new LruStore(1024), { ttlMs: 1000 }); + const store = new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }); store.set("k", bytes("v")); store.delete("k"); expect(store.get("k")).toBeNull(); @@ -153,7 +153,7 @@ describe("TtlStore", () => { it("delegates byte-cap eviction to a wrapped LruStore (the header counts toward the cap)", () => { // Each stored entry is a 12-byte header + 3-byte payload = 15 bytes; a cap of 18 holds one, not two. - const store = new TtlStore(new LruStore(12 + 3 + 3), { ttlMs: 1_000_000 }); + const store = new TtlStore(new LruStore({ maxBytes: 12 + 3 + 3 }), { ttlMs: 1_000_000 }); store.set("a", bytes("xxx")); // 15 bytes store.set("b", bytes("yyy")); // 15 more → total 30 > 18, evicts 'a' expect(store.get("a")).toBeNull(); @@ -189,7 +189,7 @@ describe("TtlStore", () => { }); it("passes through the wrapped store's synchronous nature", () => { - const store = new TtlStore(new LruStore(1024), { ttlMs: 1000 }); + const store = new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }); expect(store.set("k", bytes("v"))).toBeUndefined(); const getResult = store.get("k"); expect(getResult).not.toBeInstanceOf(Promise); @@ -214,7 +214,7 @@ describe("TtlStore fronting a shared remote (HierarchicalStore)", () => { it("masks a fresher remote write within ttlMs, then serves it after the cap", async () => { const remote = makeRemote(); - const store = new HierarchicalStore([new TtlStore(new LruStore(1024), { ttlMs: 1000 }), remote], { + const store = new HierarchicalStore([new TtlStore(new LruStore({ maxBytes: 1024 }), { ttlMs: 1000 }), remote], { populateOnMiss: true, }); diff --git a/test/transports/cache/eth-call/handler.test.ts b/test/transports/cache/eth-call/handler.test.ts index fead5c2..4a974b7 100644 --- a/test/transports/cache/eth-call/handler.test.ts +++ b/test/transports/cache/eth-call/handler.test.ts @@ -15,11 +15,13 @@ import { import { describe, expect, it, vi } from "vitest"; import { LazyNdjsonMap } from "../../../../src/internal/index.js"; +import { createFacetId, observe, withLogging } from "../../../../src/observability.js"; import { MemoryStore } from "../../../../src/stores/memory.js"; import { handleEthCall } from "../../../../src/transports/cache/eth-call/handler.js"; import type { CachedEthCallEntry } from "../../../../src/transports/cache/eth-call/types.js"; import { keychain } from "../../../../src/transports/cache/keychain.js"; import type { CacheSchema } from "../../../../src/transports/cache/schema.js"; +import { cacheTransportKey } from "../../../../src/transports/cache/schema.js"; import type { HandlerContext } from "../../../../src/transports/cache/types.js"; import { ETH_CALL_POLICY_ADDRESS } from "../../../../src/transports/state-overrides.js"; import type { EIP1193Parameters } from "../../../../src/types.js"; @@ -28,6 +30,7 @@ import { OK_SENTINEL, unwrapDeploylessFactoryCall } from "../../../../src/utils/ import { isDeploylessPartialResultError } from "../../../../src/utils/deployless/errors.js"; import { flzDecompress } from "../../../../src/utils/deployless/flz.js"; import { parse, stringify } from "../../../../src/utils/json.js"; +import { createStubLogger, findDotted } from "../../../helpers/logger.js"; type EthCallRequest = EIP1193Parameters; @@ -110,6 +113,7 @@ function ctx(requestFn: HandlerContext["requestFn"], store = new MemoryStore()): binSize: 10_000, invalidationStrategy: () => 0, gasLimit: 30_000_000, + facetId: createFacetId(cacheTransportKey), }; } @@ -547,6 +551,22 @@ describe("handleEthCall", () => { expect(keys).toEqual([1, 3].map((n) => entryKeyFor(pad(toHex(n), { size: 32 }), pageAbi))); }); + it("reports elements_missing against caller inputs, not deduped entries", async () => { + // addr(2) appears twice, so one declined cache entry stands for two caller indices. + const req = pagedRequest([addr(1), addr(2), addr(3), addr(2)]); + + const { logger, events } = createStubLogger(); + const context = ctx(mockPagedFn([2]), new MemoryStore()); + // Same id on the boundary as the handler uses, so both write the bare key. + const observed = observe(() => handleEthCall(context, req), context.facetId); + const error = await withLogging(() => observed({ method: "eth_call" }).catch((e) => e), { logger }); + + expect(isDeploylessPartialResultError(error)).toBe(true); + expect(error.missing).toEqual([1, 3]); + // The field has to match the error the caller actually receives. + expect(findDotted(events[0]!.context, cacheTransportKey, "eth_call.elements_missing")).toBe(error.missing.length); + }); + it("waits for a slow sibling chunk to commit before a failing chunk's error escapes", async () => { const store = new MemoryStore(); const addrs = [addr(1), addr(2), addr(3), addr(4)]; diff --git a/test/transports/cache/eth-get-logs/handler.test.ts b/test/transports/cache/eth-get-logs/handler.test.ts index e56856e..26a58df 100644 --- a/test/transports/cache/eth-get-logs/handler.test.ts +++ b/test/transports/cache/eth-get-logs/handler.test.ts @@ -1,10 +1,11 @@ import { type RpcLog, toHex } from "viem"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createFacetId } from "../../../../src/observability.js"; import { MemoryStore } from "../../../../src/stores/memory.js"; import { handleEthGetLogs } from "../../../../src/transports/cache/eth-get-logs/handler.js"; import { keychain } from "../../../../src/transports/cache/keychain.js"; -import type { CacheSchema } from "../../../../src/transports/cache/schema.js"; +import { type CacheSchema, cacheTransportKey } from "../../../../src/transports/cache/schema.js"; import type { HandlerContext, InvalidationStrategy } from "../../../../src/transports/cache/types.js"; import type { EIP1193Parameters } from "../../../../src/types.js"; import { createCoalescingMutex } from "../../../../src/utils/coalescing-mutex.js"; @@ -67,6 +68,7 @@ describe("handleEthGetLogs", () => { gasLimit: 30_000_000, requestFn: requestFn as unknown as HandlerContext["requestFn"], coalesce, + facetId: createFacetId(cacheTransportKey), }, { method: "eth_getLogs", params } as unknown as EthGetLogsRequest, ); diff --git a/test/transports/deployless-paged.test.ts b/test/transports/deployless-paged.test.ts index 855cd28..6b7e3f4 100644 --- a/test/transports/deployless-paged.test.ts +++ b/test/transports/deployless-paged.test.ts @@ -15,11 +15,13 @@ import { } from "viem"; import { describe, expect, it, vi } from "vitest"; +import { withLogging } from "../../src/observability.js"; import { deployless } from "../../src/transports/deployless/index.js"; import { ETH_CALL_POLICY_ADDRESS } from "../../src/transports/state-overrides.js"; import type { EIP1193Parameters } from "../../src/types.js"; import { OK_SENTINEL, OOG_SENTINEL, unwrapDeploylessFactoryCall } from "../../src/utils/deployless/codec.envelope.js"; import { isDeploylessPartialResultError } from "../../src/utils/deployless/errors.js"; +import { createStubLogger, findDotted } from "../helpers/logger.js"; type EthCallRequest = EIP1193Parameters; @@ -204,6 +206,26 @@ describe("deployless (paged)", () => { expect(error.missing).toEqual([1, 3]); }); + it("stamps paged continuations and unservable elements onto the wide event", async () => { + // Serves 2 per call and declines the element valued 3, so the run both continues and + // ends up short: one continuation, two waves, one element the lens refused. + const requestFn = mockPagedLens({ pageSize: 2, decline: [3] }); + const transport = createTransport(requestFn); + + const { logger, events } = createStubLogger(); + await withLogging(() => transport.request(createRequest([1, 2, 3, 4, 5].map(addr))), { logger }).catch(() => {}); + + expect(events).toHaveLength(1); + const { context } = events[0]!; + const field = (name: string) => findDotted(context, "viem-dlc-deployless", `eth_call.${name}`); + expect(context.status).toBe("error"); + expect(field("elements_missing")).toBe(1); + expect(field("pages_continued")).toBe(1); + expect(field("pages_waves")).toBe(2); + // A lens stopping early is a continuation, not a bisect. + expect(field("splits_count")).toBe(0); + }); + it("propagates an ordinary lens revert instead of treating it as unservable", async () => { const requestFn = vi.fn().mockRejectedValue(revertWith("0xdeadbeef")); const transport = createTransport(requestFn); diff --git a/test/transports/deployless.test.ts b/test/transports/deployless.test.ts index 84bf928..95f6c93 100644 --- a/test/transports/deployless.test.ts +++ b/test/transports/deployless.test.ts @@ -15,6 +15,7 @@ import { } from "viem"; import { describe, expect, it, vi } from "vitest"; +import { withLogging } from "../../src/observability.js"; import { deployless } from "../../src/transports/deployless/index.js"; import { ETH_CALL_POLICY_ADDRESS } from "../../src/transports/state-overrides.js"; import type { EIP1193Parameters } from "../../src/types.js"; @@ -26,6 +27,7 @@ import { wrapDeploylessFactoryCall, } from "../../src/utils/deployless/codec.envelope.js"; import { flzDecompress } from "../../src/utils/deployless/flz.js"; +import { createStubLogger, findDotted } from "../helpers/logger.js"; type EthCallRequest = EIP1193Parameters; @@ -202,6 +204,37 @@ describe("deployless", () => { expect(decoded).toEqual([addr(1), addr(2), addr(3), addr(4), addr(5)].map((a) => BigInt(a))); }); + it("stamps input_elements, nominal_batches, and splits onto the wide event", async () => { + const batchSize = 520; + const requestFn = mockBalancesOfFn(); + const transport = createTransport(requestFn); + const req = createRequest([addr(1), addr(2), addr(3), addr(4), addr(5)], { batch: { batchSize } }); + + const { logger, events } = createStubLogger(); + await withLogging(() => transport.request(req), { logger }); + + expect(events).toHaveLength(1); + const { context } = events[0]!; + const field = (name: string) => findDotted(context, "viem-dlc-deployless", `eth_call.${name}`); + expect(field("input_elements")).toBe(5); + expect(field("elements_requested")).toBe(5); + expect(field("elements_fetched")).toBe(5); + expect(field("nominal_batches")).toBeGreaterThan(1); + expect(field("splits_count")).toBe(0); + expect(field("splits_size")).toBe(0); + expect(field("splits_timeout")).toBe(0); + expect(field("splits_max_depth")).toBe(0); + // Non-paged lens: the paged-only fields are absent, which is how a consumer tells + // a paged run from an ordinary one. + expect(field("pages_waves")).toBeUndefined(); + expect(field("pages_continued")).toBeUndefined(); + expect(field("elements_missing")).toBeUndefined(); + + // One packed-size sample per batch, none exceeding the budget they were packed under. + expect(field("batch_bytes.count")).toBe(field("nominal_batches")); + expect(field("batch_bytes.max")).toBeLessThanOrEqual(batchSize); + }); + it("forwards block, cleaned stateOverride, and blockOverride upstream", async () => { const requestFn = mockBalancesOfFn(); const transport = createTransport(requestFn);