Allow offchain datasource creation in offchain handlers (#4713)

* core: allow creation of offchain datasources in offchain handlers

* tests: Add integration tests for offchain datasource creation in offchain handlers

* tests/runner-tests : Test failure of spawning onchan ds from offchain handler

Author: incrypto32

core/src/subgraph/runner.rs

@@ -433,7 +433,7 @@ where
         // Check for offchain events and process them, including their entity modifications in the
         // set to be transacted.
         let offchain_events = self.ctx.offchain_monitor.ready_offchain_events()?;
-        let (offchain_mods, processed_data_sources) = self
+        let (offchain_mods, processed_data_sources, persisted_off_chain_data_sources) = self
             .handle_offchain_triggers(offchain_events, &block)
             .await?;
         mods.extend(offchain_mods);
@@ -476,12 +476,13 @@ where
 
         let BlockState {
             deterministic_errors,
-            persisted_data_sources,
+            mut persisted_data_sources,
             ..
         } = block_state;
 
         let first_error = deterministic_errors.first().cloned();
 
+        persisted_data_sources.extend(persisted_off_chain_data_sources);
         store
             .transact_block_operations(
                 block_ptr,
@@ -703,9 +704,17 @@ where
         &mut self,
         triggers: Vec<offchain::TriggerData>,
         block: &Arc<C::Block>,
-    ) -> Result<(Vec<EntityModification>, Vec<StoredDynamicDataSource>), Error> {
+    ) -> Result<
+        (
+            Vec<EntityModification>,
+            Vec<StoredDynamicDataSource>,
+            Vec<StoredDynamicDataSource>,
+        ),
+        Error,
+    > {
         let mut mods = vec![];
         let mut processed_data_sources = vec![];
+        let mut persisted_data_sources = vec![];
 
         for trigger in triggers {
             // Using an `EmptyStore` and clearing the cache for each trigger is a makeshift way to
@@ -742,10 +751,17 @@ where
                 })?;
 
             anyhow::ensure!(
-                !block_state.has_created_data_sources(),
-                "Attempted to create data source in offchain data source handler. This is not yet supported.",
+                !block_state.has_created_on_chain_data_sources(),
+                "Attempted to create on-chain data source in offchain data source handler. This is not yet supported.",
             );
 
+            let (data_sources, _) =
+                self.create_dynamic_data_sources(block_state.drain_created_data_sources())?;
+
+            // Add entity operations for the new data sources to the block state
+            // and add runtimes for the data sources to the subgraph instance.
+            self.persist_dynamic_data_sources(&mut block_state, data_sources);
+
             // This propagates any deterministic error as a non-deterministic one. Which might make
             // sense considering offchain data sources are non-deterministic.
             if let Some(err) = block_state.deterministic_errors.into_iter().next() {
@@ -759,9 +775,10 @@ where
                     .modifications,
             );
             processed_data_sources.extend(block_state.processed_data_sources);
+            persisted_data_sources.extend(block_state.persisted_data_sources)
         }
 
-        Ok((mods, processed_data_sources))
+        Ok((mods, processed_data_sources, persisted_data_sources))
     }
 }
 

tests/runner-tests/file-data-sources/schema.graphql

@@ -6,4 +6,10 @@ type IpfsFile @entity {
 type IpfsFile1 @entity {
   id: ID!
   content: String!
+}
+
+type SpawnTestEntity @entity {
+  id: ID!
+  content: String!
+  context: String!
 }

tests/runner-tests/file-data-sources/src/mapping.ts

@@ -1,24 +1,30 @@
-import { ethereum, dataSource, BigInt, Bytes, DataSourceContext } from '@graphprotocol/graph-ts'
-import { TestEvent } from '../generated/Contract/Contract'
-import { IpfsFile, IpfsFile1 } from '../generated/schema'
+import {
+  ethereum,
+  dataSource,
+  BigInt,
+  Bytes,
+  DataSourceContext,
+} from "@graphprotocol/graph-ts";
+import { TestEvent } from "../generated/Contract/Contract";
+import { IpfsFile, IpfsFile1, SpawnTestEntity } from "../generated/schema";
 
 // CID of `file-data-sources/abis/Contract.abi` after being processed by graph-cli.
-const KNOWN_CID = "QmQ2REmceVtzawp7yrnxLQXgNNCtFHEnig6fL9aqE1kcWq"
+const KNOWN_CID = "QmQ2REmceVtzawp7yrnxLQXgNNCtFHEnig6fL9aqE1kcWq";
 
 export function handleBlock(block: ethereum.Block): void {
-  let entity = new IpfsFile("onchain")
-  entity.content = "onchain"
-  entity.save()
+  let entity = new IpfsFile("onchain");
+  entity.content = "onchain";
+  entity.save();
 
   // This will create the same data source twice, once at block 0 and another at block 2.
   // The creation at block 2 should be detected as a duplicate and therefore a noop.
   if (block.number == BigInt.fromI32(0) || block.number == BigInt.fromI32(2)) {
-    dataSource.create("File", [KNOWN_CID])
+    dataSource.create("File", [KNOWN_CID]);
   }
 
   if (block.number == BigInt.fromI32(1)) {
-    let entity = IpfsFile.load("onchain")!
-    assert(entity.content == "onchain")
+    let entity = IpfsFile.load("onchain")!;
+    assert(entity.content == "onchain");
 
     // The test assumes file data sources are processed in the block in which they are created.
     // So the ds created at block 0 will have been processed.
@@ -27,17 +33,16 @@ export function handleBlock(block: ethereum.Block): void {
     assert(IpfsFile.load(KNOWN_CID) == null);
 
     // Test that using an invalid CID will be ignored
-    dataSource.create("File", ["hi, I'm not valid"])
+    dataSource.create("File", ["hi, I'm not valid"]);
   }
 
-
-  // This will invoke File1 data source with same CID, which will be used 
+  // This will invoke File1 data source with same CID, which will be used
   // to test whether same cid is triggered across different data source.
   if (block.number == BigInt.fromI32(3)) {
     // Test that onchain data sources cannot read offchain data (again, but this time more likely to hit the DB than the write queue).
     assert(IpfsFile.load(KNOWN_CID) == null);
 
-    dataSource.create("File1", [KNOWN_CID])
+    dataSource.create("File1", [KNOWN_CID]);
   }
 }
 
@@ -46,17 +51,32 @@ export function handleTestEvent(event: TestEvent): void {
 
   if (command == "createFile2") {
     // Will fail the subgraph when processed due to mismatch in the entity type and 'entities'.
-    dataSource.create("File2", [KNOWN_CID])
+    dataSource.create("File2", [KNOWN_CID]);
   } else if (command == "saveConflictingEntity") {
     // Will fail the subgraph because the same entity has been created in a file data source.
-    let entity = new IpfsFile(KNOWN_CID)
-    entity.content = "empty"
-    entity.save()
+    let entity = new IpfsFile(KNOWN_CID);
+    entity.content = "empty";
+    entity.save();
   } else if (command == "createFile1") {
     // Will fail the subgraph with a conflict between two entities created by offchain data sources.
     let context = new DataSourceContext();
     context.setBytes("hash", event.block.hash);
-    dataSource.createWithContext("File1", [KNOWN_CID], context)
+    dataSource.createWithContext("File1", [KNOWN_CID], context);
+  } else if (command == "spawnOffChainHandlerTest") {
+    // Used to test the spawning of a file data source from another file data source handler.
+    // `SpawnTestHandler` will spawn a file data source that will be handled by `spawnOffChainHandlerTest`,
+    // which creates another file data source `OffChainDataSource`, which will be handled by `handleSpawnedTest`.
+    let context = new DataSourceContext();
+    context.setString("command", command);
+    dataSource.createWithContext("SpawnTestHandler", [KNOWN_CID], context);
+  } else if (command == "spawnOnChainHandlerTest") {
+    // Used to test the failure of spawning of on-chain data source from a file data source handler.
+    // `SpawnTestHandler` will spawn a file data source that will be handled by `spawnTestHandler`,
+    // which creates an `OnChainDataSource`, which should fail since spawning onchain datasources
+    // from offchain handlers is not allowed.
+    let context = new DataSourceContext();
+    context.setString("command", command);
+    dataSource.createWithContext("SpawnTestHandler", [KNOWN_CID], context);
   } else {
     assert(false, "Unknown command: " + command);
   }
@@ -66,22 +86,48 @@ export function handleFile(data: Bytes): void {
   // Test that offchain data sources cannot read onchain data.
   assert(IpfsFile.load("onchain") == null);
 
-  if (dataSource.stringParam() != "QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ") {
+  if (
+    dataSource.stringParam() != "QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ"
+  ) {
     // Test that an offchain data source cannot read from another offchain data source.
-    assert(IpfsFile.load("QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ") == null);
+    assert(
+      IpfsFile.load("QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ") == null
+    );
   }
 
-  let entity = new IpfsFile(dataSource.stringParam())
-  entity.content = data.toString()
-  entity.save()
+  let entity = new IpfsFile(dataSource.stringParam());
+  entity.content = data.toString();
+  entity.save();
 
   // Test that an offchain data source can load its own entities
-  let loaded_entity = IpfsFile.load(dataSource.stringParam())!
-  assert(loaded_entity.content == entity.content)
+  let loaded_entity = IpfsFile.load(dataSource.stringParam())!;
+  assert(loaded_entity.content == entity.content);
 }
 
 export function handleFile1(data: Bytes): void {
-  let entity = new IpfsFile1(dataSource.stringParam())
-  entity.content = data.toString()
-  entity.save()
+  let entity = new IpfsFile1(dataSource.stringParam());
+  entity.content = data.toString();
+  entity.save();
+}
+
+// Used to test spawning a file data source from another file data source handler.
+// This function spawns a file data source that will be handled by `handleSpawnedTest`.
+export function spawnTestHandler(data: Bytes): void {
+  let context = new DataSourceContext();
+  context.setString("file", "fromSpawnTestHandler");
+  let command = dataSource.context().getString("command");
+  if (command == "spawnOffChainHandlerTest") {
+    dataSource.createWithContext("OffChainDataSource", [KNOWN_CID], context);
+  } else if (command == "spawnOnChainHandlerTest") {
+    dataSource.createWithContext("OnChainDataSource", [KNOWN_CID], context);
+  }
+}
+
+// This is the handler for the data source spawned by `spawnOffChainHandlerTest`.
+export function handleSpawnedTest(data: Bytes): void {
+  let entity = new SpawnTestEntity(dataSource.stringParam());
+  let context = dataSource.context().getString("file");
+  entity.content = data.toString();
+  entity.context = context;
+  entity.save();
 }

tests/runner-tests/file-data-sources/subgraph.yaml

@@ -24,6 +24,26 @@ dataSources:
           handler: handleTestEvent
       file: ./src/mapping.ts
 templates:
+  - kind: ethereum/contract
+    name: OnChainDataSource
+    network: test
+    source:
+      abi: Contract
+    mapping:
+      kind: ethereum/events
+      apiVersion: 0.0.7
+      language: wasm/assemblyscript
+      entities:
+        - Gravatar
+      abis:
+        - name: Contract
+          file: ./abis/Contract.abi
+      blockHandlers:
+        - handler: handleBlock
+      eventHandlers:
+        - event: TestEvent(string)
+          handler: handleTestEvent
+      file: ./src/mapping.ts
   - kind: file/ipfs
     name: File
     mapping:
@@ -62,4 +82,30 @@ templates:
         - name: Contract
           file: ./abis/Contract.abi
       handler: handleFile1
+      file: ./src/mapping.ts
+  - kind: file/ipfs
+    name: SpawnTestHandler
+    mapping:
+      kind: ethereum/events
+      apiVersion: 0.0.7
+      language: wasm/assemblyscript
+      entities:
+        - SpawnTestEntity
+      abis:
+        - name: Contract
+          file: ./abis/Contract.abi
+      handler: spawnTestHandler
+      file: ./src/mapping.ts
+  - kind: file/ipfs
+    name: OffChainDataSource
+    mapping:
+      kind: ethereum/events
+      apiVersion: 0.0.7
+      language: wasm/assemblyscript
+      entities:
+        - SpawnTestEntity
+      abis:
+        - name: Contract
+          file: ./abis/Contract.abi
+      handler: handleSpawnedTest
       file: ./src/mapping.ts

tests/src/fixture/mod.rs

@@ -496,7 +496,7 @@ pub async fn wait_for_sync(
                 continue;
             }
         };
-
+        info!(logger, "TEST: sync status: {:?}", block_ptr);
         let status = store.status_for_id(deployment.id);
 
         if let Some(fatal_error) = status.fatal_error {