store: Assign separate causality region for offchain dses

chain/ethereum/src/data_source.rs

@@ -131,6 +131,7 @@ impl blockchain::DataSource<Chain> for DataSource {
                 .as_ref()
                 .map(|ctx| serde_json::to_value(&ctx).unwrap()),
             creation_block: self.creation_block,
+            is_offchain: false,
         }
     }
 
@@ -143,8 +144,14 @@ impl blockchain::DataSource<Chain> for DataSource {
             param,
             context,
             creation_block,
+            is_offchain,
         } = stored;
 
+        ensure!(
+            !is_offchain,
+            "attempted to convert offchain data source to ethereum data source"
+        );
+
         let context = context.map(serde_json::from_value).transpose()?;
 
         let contract_abi = template.mapping.find_abi(&template.source.abi)?;

graph/src/components/store/mod.rs

@@ -799,6 +799,7 @@ pub struct StoredDynamicDataSource {
     pub param: Option<Bytes>,
     pub context: Option<serde_json::Value>,
     pub creation_block: Option<BlockNumber>,
+    pub is_offchain: bool,
 }
 
 /// An internal identifer for the specific instance of a deployment. The

graph/src/data_source/offchain.rs

@@ -86,6 +86,7 @@ impl DataSource {
             param: Some(param),
             context,
             creation_block: self.creation_block,
+            is_offchain: true,
         }
     }
 

store/postgres/src/dynds/private.rs

@@ -25,7 +25,7 @@ pub(crate) struct DataSourcesTable {
     table: DynTable,
     vid: DynColumn<Integer>,
     block_range: DynColumn<sql_types::Range<Integer>>,
-    _causality_region: DynColumn<Integer>,
+    causality_region: DynColumn<Integer>,
     manifest_idx: DynColumn<Integer>,
     param: DynColumn<Nullable<Binary>>,
     context: DynColumn<Nullable<Jsonb>>,
@@ -43,7 +43,7 @@ impl DataSourcesTable {
             namespace,
             vid: table.column("vid"),
             block_range: table.column("block_range"),
-            _causality_region: table.column("causality_region"),
+            causality_region: table.column("causality_region"),
             manifest_idx: table.column("manifest_idx"),
             param: table.column("param"),
             context: table.column("context"),
@@ -85,6 +85,7 @@ impl DataSourcesTable {
             i32,
             Option<Vec<u8>>,
             Option<serde_json::Value>,
+            i32,
         );
         let tuples = self
             .table
@@ -95,26 +96,34 @@ impl DataSourcesTable {
                 &self.manifest_idx,
                 &self.param,
                 &self.context,
+                &self.causality_region,
             ))
             .order_by(&self.vid)
             .load::<Tuple>(conn)?;
 
         let mut dses: Vec<_> = tuples
             .into_iter()
-            .map(|(block_range, manifest_idx, param, context)| {
-                let creation_block = match block_range.0 {
-                    Bound::Included(block) => Some(block),
-
-                    // Should never happen.
-                    Bound::Excluded(_) | Bound::Unbounded => unreachable!("dds with open creation"),
-                };
-                StoredDynamicDataSource {
-                    manifest_idx: manifest_idx as u32,
-                    param: param.map(|p| p.into()),
-                    context,
-                    creation_block,
-                }
-            })
+            .map(
+                |(block_range, manifest_idx, param, context, causality_region)| {
+                    let creation_block = match block_range.0 {
+                        Bound::Included(block) => Some(block),
+
+                        // Should never happen.
+                        Bound::Excluded(_) | Bound::Unbounded => {
+                            unreachable!("dds with open creation")
+                        }
+                    };
+
+                    let is_offchain = causality_region > 0;
+                    StoredDynamicDataSource {
+                        manifest_idx: manifest_idx as u32,
+                        param: param.map(|p| p.into()),
+                        context,
+                        creation_block,
+                        is_offchain,
+                    }
+                },
+            )
             .collect();
 
         // This sort is stable and `tuples` was ordered by vid, so `dses` will be ordered by `(creation_block, vid)`.
@@ -129,9 +138,6 @@ impl DataSourcesTable {
         data_sources: &[StoredDynamicDataSource],
         block: BlockNumber,
     ) -> Result<usize, StoreError> {
-        // Currently all data sources share the same causality region.
-        let causality_region = 0;
-
         let mut inserted_total = 0;
 
         for ds in data_sources {
@@ -140,6 +146,7 @@ impl DataSourcesTable {
                 param,
                 context,
                 creation_block,
+                is_offchain,
             } = ds;
 
             if creation_block != &Some(block) {
@@ -150,19 +157,32 @@ impl DataSourcesTable {
                 ));
             }
 
-            let query = format!(
-                "insert into {}(block_range, manifest_idx, causality_region, param, context) \
-                 values (int4range($1, null), $2, $3, $4, $5)",
-                self.qname
-            );
-
-            inserted_total += sql_query(query)
+            // Onchain data sources have the causality region explicitly set to 0, while offchain
+            // data sources have an unique causality region assigned from the sequence.
+            let query = match is_offchain {
+                false => format!(
+                    "insert into {}(block_range, manifest_idx, param, context, causality_region) \
+                            values (int4range($1, null), $2, $3, $4, $5)",
+                    self.qname
+                ),
+
+                true => format!(
+                    "insert into {}(block_range, manifest_idx, param, context) \
+                            values (int4range($1, null), $2, $3, $4)",
+                    self.qname
+                ),
+            };
+
+            let query = sql_query(query)
                 .bind::<Nullable<Integer>, _>(creation_block)
                 .bind::<Integer, _>(*manifest_idx as i32)
-                .bind::<Integer, _>(causality_region)
                 .bind::<Nullable<Binary>, _>(param.as_ref().map(|p| &**p))
-                .bind::<Nullable<Jsonb>, _>(context)
-                .execute(conn)?;
+                .bind::<Nullable<Jsonb>, _>(context);
+
+            inserted_total += match is_offchain {
+                false => query.bind::<Integer, _>(0).execute(conn)?,
+                true => query.execute(conn)?,
+            };
         }
 
         Ok(inserted_total)
@@ -233,8 +253,15 @@ impl DataSourcesTable {
                 param,
                 context,
                 creation_block,
+                is_offchain,
             } = ds;
 
+            if !is_offchain {
+                return Err(constraint_violation!(
+                    "called remove_offchain with onchain data sources"
+                ));
+            }
+
             let query = format!(
                 "update {} set block_range = 'empty'::int4range \
                  where manifest_idx = $1

store/postgres/src/dynds/shared.rs

@@ -81,6 +81,10 @@ pub(super) fn load(
             param: Some(address.into()),
             context: context.map(|ctx| serde_json::from_str(&ctx)).transpose()?,
             creation_block,
+
+            // The shared schema is only used for legacy deployments, and therefore not used for
+            // subgraphs that use file data sources.
+            is_offchain: false,
         };
 
         if data_sources.last().and_then(|d| d.creation_block) > data_source.creation_block {
@@ -116,7 +120,15 @@ pub(super) fn insert(
                 param,
                 context,
                 creation_block: _,
+                is_offchain,
             } = ds;
+
+            if *is_offchain {
+                return Err(constraint_violation!(
+                    "using shared data source schema with file data sources"
+                ));
+            }
+
             let address = match param {
                 Some(param) => param,
                 None => {

tests/tests/runner.rs

@@ -117,9 +117,10 @@ async fn file_data_sources() {
     let blocks = {
         let block_0 = genesis();
         let block_1 = empty_block(block_0.ptr(), test_ptr(1));
-        vec![block_0, block_1]
+        let block_2 = empty_block(block_1.ptr(), test_ptr(2));
+        vec![block_0, block_1, block_2]
     };
-    let stop_block = blocks.last().unwrap().block.ptr();
+    let stop_block = test_ptr(1);
     let chain = Arc::new(chain(blocks, &stores).await);
     let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain, None).await;
     ctx.start_and_sync_to(stop_block).await;
@@ -144,4 +145,9 @@ async fn file_data_sources() {
         query_res,
         Some(object! { ipfsFile: object!{ id: id , content: "[]" } })
     );
+
+    // Test loading offchain data sources from DB.
+    ctx.provider.stop(ctx.deployment.clone()).await.unwrap();
+    let stop_block = test_ptr(2);
+    ctx.start_and_sync_to(stop_block).await;
 }