fix allowed schema ID logic; add test (#603)

* fix allowed schema ID logic; add test

* Update CHANGELOG.md

* run cargo fmt; fix PR number

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Fix bug where known schemas are not replicated between nodes [#603](https://github.com/p2panda/aquadoggo/pull/603).
+
 ## [0.7.0]
 
 ### Added

aquadoggo/src/replication/ingest.rs

@@ -52,7 +52,7 @@ impl SyncIngest {
         // If the node has been configured with an allow-list of supported schema ids, check that
         // the sent operation follows one of our supported schema
         if self.schema_provider.is_allow_list_active()
-            && self
+            && !self
                 .schema_provider
                 .supported_schema_ids()
                 .await
@@ -115,6 +115,7 @@ mod tests {
     use crate::replication::errors::IngestError;
     use crate::replication::SyncIngest;
     use crate::test_utils::{test_runner_with_manager, TestNodeManager};
+    use crate::{AllowList, Configuration};
 
     #[rstest]
     fn reject_duplicate_entries(
@@ -142,4 +143,31 @@ mod tests {
             assert!(matches!(result, Err(IngestError::DuplicateEntry(_))));
         })
     }
+
+    #[rstest]
+    fn allow_supported_schema_ids(
+        schema: Schema,
+        encoded_entry: EncodedEntry,
+        encoded_operation: EncodedOperation,
+    ) {
+        test_runner_with_manager(move |manager: TestNodeManager| async move {
+            let config = Configuration {
+                allow_schema_ids: AllowList::Set(vec![schema.id().clone()]),
+                ..Configuration::default()
+            };
+            let node = manager.create_with_config(config).await;
+
+            assert!(node.context.schema_provider.is_allow_list_active());
+
+            let _ = node.context.schema_provider.update(schema.clone()).await;
+            let (tx, _rx) = broadcast::channel(8);
+            let ingest = SyncIngest::new(node.context.schema_provider.clone(), tx.clone());
+
+            let result = ingest
+                .handle_entry(&node.context.store, &encoded_entry, &encoded_operation)
+                .await;
+
+            assert!(result.is_ok());
+        });
+    }
 }

aquadoggo/src/test_utils/runner.rs

@@ -59,22 +59,20 @@ impl TestNodeManager {
     }
 
     pub async fn create(&self) -> TestNode {
+        self.create_with_config(Configuration::default()).await
+    }
+
+    pub async fn create_with_config(&self, config: Configuration) -> TestNode {
         let (_config, pool) = initialize_sqlite_db().await;
 
         // Initialise test store using pool.
         let store = SqlStore::new(pool.clone());
 
-        // Construct node config supporting any schema.
-        let cfg = Configuration::default();
+        let schema_provider = SchemaProvider::new(vec![], config.allow_schema_ids.clone());
 
         // Construct the actual test node
         let test_node = TestNode {
-            context: Context::new(
-                store.clone(),
-                KeyPair::new(),
-                cfg,
-                SchemaProvider::default(),
-            ),
+            context: Context::new(store.clone(), KeyPair::new(), config, schema_provider),
         };
 
         self.pools.lock().await.push(pool);