SpS autogenerated from Ethereum subgraph.yaml

[abourget]

Goal: go validate with customers that having a no-code 2x speed increase by just wrapping their subgraph into a SpS is desirable

We want to validate that 2x is real (stepd has done good work on that).
We want to show what the experience would look like.

Feature

1. Ability to take a stock subgraph.yaml,
2. Extract the info about the dyanmic data source,
3. Pass it to our code generator that would produce a high-speed extractor for that pattern
4. Generate protobuf structure that completely matches all of the inputs expected by the standard subgraph
5. Produce the AssemblyScript bindings for those protobufs
6. Produce a stub mapping file, that converts the protobufs into the native structures, and routes the mapping handlers to the original subgraph.
7. Profit from 2x speed increase.

Example

Original subgraph.yaml

Starting from Uniswap V3's original subgraph.yaml here:

https://github.com/Uniswap/v3-subgraph/blob/main/subgraph.yaml:

specVersion: 0.0.4
description: Uniswap is a decentralized protocol for automated token exchange on Ethereum.
repository: https://github.com/Uniswap/uniswap-v3-subgraph
schema:
  file: ./schema.graphql
...
dataSources:
  - kind: ethereum/contract
    name: Factory
    source:
      abi: Factory
      address: "0x33128a8fC17869897dcE68Ed026d694621f6FDfD"
      startBlock: 2009445
    mapping:
      kind: ethereum/events
      ...
      file: ./src/mappings/factory.ts
      entities:
        - Pool
        - Token
      abis:
        - name: Factory
          file: ./abis/factory.json
...
        - name: Pool
          file: ./abis/pool.json
      eventHandlers:
        - event: PoolCreated(indexed address,indexed address,indexed uint24,int24,address)
          handler: handlePoolCreated
templates:
  - kind: ethereum/contract
    name: Pool
    source:
      abi: Pool
    mapping:
    ...
      file: ./src/mappings/pool/index.ts
      entities:
        - Pool
        - Token
      abis:
        - name: Pool
          file: ./abis/pool.json
          ...
      eventHandlers:
        - event: Initialize(uint160,int24)
          handler: handleInitialize
        - event: Swap(indexed address,indexed address,int256,int256,uint160,uint128,int24)
          handler: handleSwap
          ...

Produced SpS manifest

# AUTO-GENERATED. DO NOT EDIT.
specVersion: 1.0.0
description: univ3sps Substreams-powered-Subgraph
schema:
  file: ./schema.graphql
dataSources:
  - kind: substreams
    name: filtered_events
    network: mainnet
    source:
      package:
        moduleName: map_events
        file: univ3sps-v0.1.0.spkg
    mapping:
      apiVersion: 0.0.7
      kind: substreams/graph-entities  # Would want to update to `substreams/triggers` ?
      file: ./src/generated/substreams_handler.ts
      handler: generatedSubstreamsHandler

and the src/generated/substreams_handler.ts looks like:

export function generatedSubstreamsHandler(bytes: Uint8Array): void {
  // NOTE: OBTAINS _all_ events from the block, in a single swift

  const events: Events = Protobuf.decode<Events>(bytes, Events.decode)
  const block = extractBlockFromMessage(events)

  events.factoryPoolCreateds.forEach((factoryPoolCreated: Factory_PoolCreated) => {
    let converted = new PoolCreated(
      Address.fromBytes(Bytes.fromUint8Array(factoryPoolCreated.pool)), // Address
      BigInt.fromI32(factoryPoolCreated.evtIndex as i32),
      BigInt.fromI32(factoryPoolCreated.evtTrxIndex as i32),
      "PoolCreated", // logType
      block,
      extractTransactionFromEvent(factoryPoolCreated),
      [
          new ethereum.EventParam("token0", ethereum.Value.fromAddress(Address.fromBytes(Bytes.fromUint8Array(factoryPoolCreated.token0)))),
          new ethereum.EventParam("token1", ethereum.Value.fromAddress(Address.fromBytes(Bytes.fromUint8Array(factoryPoolCreated.token1)))),
          new ethereum.EventParam("fee", ethereum.Value.fromI32(factoryPoolCreated.fee as i32)),
          new ethereum.EventParam("tickSpacing", ethereum.Value.fromI32(factoryPoolCreated.tickSpacing as i32)),
          new ethereum.EventParam("pool", ethereum.Value.fromAddress(Address.fromBytes(Bytes.fromUint8Array(factoryPoolCreated.pool)))),
      ],
      null, // receipt
    )

    // THIS CALLS ORIGINAL SUBGRAPH CODE:
    handlePoolCreated(converted)
  })

  events.poolInitializes.forEach((poolInitialize: Pool_Initialize) => {
      if (Pool.load("0x" + poolInitialize.evtAddress) == null) { // check if the dynamic template was created, for now we rely on pool
       return;
      }
    let converted = new Initialize(
      Address.fromString(poolInitialize.evtAddress), // Address
      BigInt.fromI32(poolInitialize.evtIndex as i32),
      BigInt.fromI32(poolInitialize.evtTrxIndex as i32),
      "Initialize",
      block,
      extractTransactionFromEvent(poolInitialize),
      [
          new ethereum.EventParam("sqrtPriceX96", ethereum.Value.fromSignedBigInt(BigInt.fromString(poolInitialize.sqrtPriceX96))),
          new ethereum.EventParam("tick", ethereum.Value.fromI32(poolInitialize.tick as i32)),
      ],
      null,
    )
    // THIS CALLS ORIGINAL SUBGRAPH CODE:
    handleInitialize(converted)
  })

  events.poolSwaps.forEach((poolSwap: Pool_Swap) => {
      if (Pool.load("0x" + poolSwap.evtAddress) == null) {
       return;
      }

    let converted = new Swap(
      Address.fromString(poolSwap.evtAddress),
      BigInt.fromI32(poolSwap.evtIndex as i32),
      BigInt.fromI32(poolSwap.evtTrxIndex as i32),
      "Swap",
      block,
      extractTransactionFromEvent(poolSwap),
      [
          new ethereum.EventParam("sender", ethereum.Value.fromAddress(Address.fromBytes(Bytes.fromUint8Array(poolSwap.sender)))),
          new ethereum.EventParam("recipient", ethereum.Value.fromAddress(Address.fromBytes(Bytes.fromUint8Array(poolSwap.recipient)))),
          new ethereum.EventParam("amount0", ethereum.Value.fromSignedBigInt(BigInt.fromString(poolSwap.amount0))),
          new ethereum.EventParam("amount1", ethereum.Value.fromSignedBigInt(BigInt.fromString(poolSwap.amount1))),
          new ethereum.EventParam("sqrtPriceX96", ethereum.Value.fromSignedBigInt(BigInt.fromString(poolSwap.sqrtPriceX96))),
          new ethereum.EventParam("liquidity", ethereum.Value.fromSignedBigInt(BigInt.fromString(poolSwap.liquidity))),
          new ethereum.EventParam("tick", ethereum.Value.fromI32(poolSwap.tick as i32)),
      ],
      null,
    )
    // THIS CALLS ORIGINAL SUBGRAPH CODE:
    handleSwap(converted)
  })

// More handlers here for burns, mint, collect...
}

alongside the protobuf-as generated models for the dynamically created protobuf messages exchanged between Substreams and the subgraph.

Result

Preliminary results show a ~2x speed increase when using this method. This avoids reconnections to firehose when a new Dynamic Data Source is added, which is potentially a source of slowdown.

Assess caveats

• The Template.create function would need to work under the substreams, and simply store that as an entity, and have no bearing on reconnections or other tweaks from the substreams source. That would require tweaks to graph-node.
• Missing eth_calls when using a substreams source. Will need to add support in graph-node, backed by another RPC endpoint.
• The substreams_handler will need to sort the events by ordinal before calling them to reproduce the handler ordering of the original subgraph. This is trivial but not represented in the code snippets here for simplicity.