Add initial coordinator e2e tests

Cargo.lock

@@ -7946,6 +7946,7 @@ dependencies = [
  "async-trait",
  "blake2",
  "ciphersuite",
+ "env_logger",
  "flexible-transcript",
  "futures",
  "hex",
@@ -7967,6 +7968,21 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "serai-coordinator-tests"
+version = "0.1.0"
+dependencies = [
+ "ciphersuite",
+ "dockertest",
+ "hex",
+ "serai-client",
+ "serai-docker-tests",
+ "serai-message-queue",
+ "serai-message-queue-tests",
+ "serai-processor-messages",
+ "tokio",
+]
+
 [[package]]
 name = "serai-db"
 version = "0.1.0"

Cargo.toml

@@ -56,6 +56,7 @@ members = [
   "tests/docker",
   "tests/message-queue",
   "tests/processor",
+  "tests/coordinator",
   "tests/reproducible-runtime",
 ]
 

coordinator/Cargo.toml

@@ -42,7 +42,10 @@ serai-client = { path = "../substrate/client", features = ["serai"] }
 
 hex = "0.4"
 serde_json = { version = "1", default-features = false }
+
 log = "0.4"
+env_logger = "0.10"
+
 tokio = { version = "1", features = ["rt-multi-thread", "sync", "time", "macros"] }
 
 [dev-dependencies]

coordinator/src/main.rs

@@ -74,6 +74,8 @@ async fn add_tributary<D: Db, P: P2p>(
   tributaries: &mut Tributaries<D, P>,
   spec: TributarySpec,
 ) -> TributaryReader<D, Transaction> {
+  log::info!("adding tributary {:?}", spec.set());
+
   let tributary = Tributary::<_, Transaction, _>::new(
     // TODO2: Use a db on a distinct volume
     db,
@@ -102,6 +104,8 @@ pub async fn scan_substrate<D: Db, Pro: Processors>(
   processors: Pro,
   serai: Serai,
 ) {
+  log::info!("scanning substrate");
+
   let mut db = substrate::SubstrateDb::new(db);
   let mut last_substrate_block = db.last_block();
 
@@ -146,6 +150,8 @@ pub async fn scan_tributaries<D: Db, Pro: Processors, P: P2p>(
   processors: Pro,
   tributaries: Arc<RwLock<Tributaries<D, P>>>,
 ) {
+  log::info!("scanning tributaries");
+
   let mut tributary_readers = vec![];
   for ActiveTributary { spec, tributary } in tributaries.read().await.values() {
     tributary_readers.push((spec.clone(), tributary.read().await.reader()));
@@ -669,6 +675,13 @@ pub async fn run<D: Db, Pro: Processors, P: P2p>(
 
 #[tokio::main]
 async fn main() {
+  if std::env::var("RUST_LOG").is_err() {
+    std::env::set_var("RUST_LOG", serai_env::var("RUST_LOG").unwrap_or_else(|| "info".to_string()));
+  }
+  env_logger::init();
+
+  log::info!("starting coordinator service...");
+
   let db = serai_db::new_rocksdb(&env::var("DB_PATH").expect("path to DB wasn't specified"));
 
   let key = Zeroizing::new(<Ristretto as Ciphersuite>::F::ZERO); // TODO
@@ -678,11 +691,17 @@ async fn main() {
 
   let serai = || async {
     loop {
-      let Ok(serai) = Serai::new("ws://127.0.0.1:9944").await else {
+      let Ok(serai) = Serai::new(&dbg!(format!(
+        "ws://{}:9944",
+        serai_env::var("SERAI_HOSTNAME").expect("Serai hostname wasn't provided")
+      )))
+      .await
+      else {
         log::error!("couldn't connect to the Serai node");
         sleep(Duration::from_secs(5)).await;
         continue;
       };
+      log::info!("made initial connection to Serai node");
       return serai;
     }
   };

coordinator/src/substrate/mod.rs

@@ -313,6 +313,7 @@ pub async fn handle_new_blocks<
   let mut latest = Some(latest);
 
   for b in (*last_block + 1) ..= latest_number {
+    log::info!("found substrate block {b}");
     handle_block(
       db,
       key,
@@ -331,6 +332,7 @@ pub async fn handle_new_blocks<
     .await?;
     *last_block += 1;
     db.set_last_block(*last_block);
+    log::info!("handled substrate block {b}");
   }
 
   Ok(())

deny.toml

@@ -69,6 +69,7 @@ exceptions = [
   { allow = ["AGPL-3.0"], name = "serai-docker-tests" },
   { allow = ["AGPL-3.0"], name = "serai-message-queue-tests" },
   { allow = ["AGPL-3.0"], name = "serai-processor-tests" },
+  { allow = ["AGPL-3.0"], name = "serai-coordinator-tests" },
   { allow = ["AGPL-3.0"], name = "serai-reproducible-runtime-tests" },
 ]
 

orchestration/coordinator/Dockerfile

@@ -0,0 +1,52 @@
+FROM rust:1.71-slim-bookworm as builder
+LABEL description="STAGE 1: Build"
+
+# Add files for build
+ADD common /serai/common
+ADD crypto /serai/crypto
+ADD coins /serai/coins
+ADD message-queue /serai/message-queue
+ADD processor /serai/processor
+ADD coordinator /serai/coordinator
+ADD substrate /serai/substrate
+ADD tests /serai/tests
+ADD Cargo.toml /serai
+ADD Cargo.lock /serai
+ADD AGPL-3.0 /serai
+
+WORKDIR /serai
+
+RUN apt update && apt upgrade -y && apt install -y pkg-config clang libssl-dev
+
+# Add the wasm toolchain
+RUN rustup target add wasm32-unknown-unknown
+
+# Mount the caches and build
+RUN --mount=type=cache,target=/root/.cargo \
+  --mount=type=cache,target=/usr/local/cargo/registry \
+  --mount=type=cache,target=/usr/local/cargo/git \
+  --mount=type=cache,target=/serai/target \
+  cd coordinator && \
+  cargo build --release --all-features && \
+  mkdir /serai/bin && \
+  mv /serai/target/release/serai-coordinator /serai/bin
+
+# Prepare Image
+FROM debian:bookworm-slim as image
+LABEL description="STAGE 2: Copy and Run"
+
+# Upgrade packages and install openssl
+RUN apt update && apt upgrade -y && apt install -y libssl-dev
+
+# Switch to a non-root user
+RUN useradd --system --create-home --shell /sbin/nologin coordinator
+USER coordinator
+
+WORKDIR /home/coordinator
+
+# Copy necessary files to run node
+COPY --from=builder --chown=processsor /serai/bin/serai-coordinator /bin/
+COPY --from=builder --chown=processsor /serai/AGPL-3.0 .
+
+# Run coordinator
+CMD ["serai-coordinator"]

orchestration/coordinator/scripts/entry-dev.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+
+export MESSAGE_QUEUE_KEY="0000000000000000000000000000000000000000000000000000000000000000"
+export MESSAGE_QUEUE_RPC="http://127.0.0.1:2287"
+
+export DB_PATH="./coordinator-db"
+export SERAI_HOSTNAME="127.0.0.1"
+
+serai-coordinator

orchestration/docker-compose.yml

@@ -2,7 +2,7 @@ version: "3.9"
 name: serai-dev
 
 volumes:
-  serai-node:
+  serai:
   serai-alice:
   serai-bob:
   serai-charlie:
@@ -78,6 +78,17 @@ services:
       - "./processor/scripts:/scripts"
     entrypoint: /scripts/entry-dev.sh
 
+  coordinator:
+    profiles:
+      - coordinator
+    build:
+      context: ../
+      dockerfile: ./orchestration/coordinator/Dockerfile
+    restart: unless-stopped
+    volumes:
+      - "./coordinator/scripts:/scripts"
+    entrypoint: /scripts/entry-dev.sh
+
   # Serai runtime
 
   runtime:
@@ -95,7 +106,7 @@ services:
   _serai:
     &serai_defaults
     restart: unless-stopped
-    image: serai:dev
+    # image: serai:dev
     profiles:
       - _
     build:
@@ -107,9 +118,9 @@ services:
     volumes:
       - "./serai/scripts:/scripts"
 
-  serai-node:
+  serai:
     <<: *serai_defaults
-    hostname: serai-node
+    hostname: serai
     profiles:
       - serai
     environment:

orchestration/processor/scripts/entry-dev.sh

@@ -3,7 +3,7 @@
 export MESSAGE_QUEUE_KEY="0000000000000000000000000000000000000000000000000000000000000000"
 export MESSAGE_QUEUE_RPC="http://127.0.0.1:2287"
 
-export DB_PATH="./bitcoin-db"
+export DB_PATH="./processor-bitcoin-db"
 export ENTROPY="0001020304050607080910111213141516171819202122232425262728293031"
 export NETWORK="bitcoin"
 export NETWORK_RPC_LOGIN="serai:seraidex"

tests/coordinator/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "serai-coordinator-tests"
+version = "0.1.0"
+description = "Tests for Serai's Coordinator"
+license = "AGPL-3.0-only"
+repository = "https://github.com/serai-dex/serai/tree/develop/tests/coordinator"
+authors = ["Luke Parker <lukeparker5132@gmail.com>"]
+keywords = []
+edition = "2021"
+publish = false
+
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]
+
+[dependencies]
+hex = "0.4"
+
+ciphersuite = { path = "../../crypto/ciphersuite", default-features = false, features = ["ristretto"] }
+
+messages = { package = "serai-processor-messages", path = "../../processor/messages" }
+
+serai-client = { path = "../../substrate/client" }
+serai-message-queue = { path = "../../message-queue" }
+
+tokio = { version = "1", features = ["time"] }
+
+dockertest = "0.3"
+serai-docker-tests = { path = "../docker" }
+serai-message-queue-tests = { path = "../message-queue" }

tests/coordinator/LICENSE

@@ -0,0 +1,15 @@
+AGPL-3.0-only license
+
+Copyright (c) 2023 Luke Parker
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License Version 3 as
+published by the Free Software Foundation.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program. If not, see <http://www.gnu.org/licenses/>.

tests/coordinator/src/lib.rs

@@ -0,0 +1,95 @@
+#![allow(clippy::needless_pass_by_ref_mut)] // False positives
+
+use std::sync::{OnceLock, Mutex};
+
+use ciphersuite::{group::ff::PrimeField, Ciphersuite, Ristretto};
+
+use serai_client::primitives::NetworkId;
+
+use dockertest::{PullPolicy, Image, LogAction, LogPolicy, LogSource, LogOptions, StartPolicy, Composition};
+
+#[cfg(test)]
+mod tests;
+
+static UNIQUE_ID: OnceLock<Mutex<u16>> = OnceLock::new();
+
+pub fn coordinator_instance(message_queue_key: <Ristretto as Ciphersuite>::F) -> Composition {
+  serai_docker_tests::build("coordinator".to_string());
+
+  Composition::with_image(
+    Image::with_repository("serai-dev-coordinator").pull_policy(PullPolicy::Never),
+  )
+  .with_env(
+    [
+      ("MESSAGE_QUEUE_KEY".to_string(), hex::encode(message_queue_key.to_repr())),
+      ("DB_PATH".to_string(), "./coordinator-db".to_string()),
+    ]
+    .into(),
+  )
+}
+
+pub fn serai_composition(name: &str) -> Composition {
+  serai_docker_tests::build("serai".to_string());
+
+  Composition::with_image(Image::with_repository("serai-dev-serai").pull_policy(PullPolicy::Never))
+    .with_cmd(vec![
+      "serai-node".to_string(),
+      "--unsafe-rpc-external".to_string(),
+      "--rpc-cors".to_string(),
+      "all".to_string(),
+      "--chain".to_string(),
+      "devnet".to_string(),
+      format!("--{name}"),
+    ])
+}
+
+pub type Handles = (String, String, String);
+pub fn coordinator_stack(name: &str) -> (Handles, <Ristretto as Ciphersuite>::F, Vec<Composition>) {
+  let serai_composition = serai_composition(name);
+
+  let (coord_key, message_queue_keys, message_queue_composition) =
+    serai_message_queue_tests::instance();
+
+  let coordinator_composition = coordinator_instance(message_queue_keys[&NetworkId::Bitcoin]);
+
+  // Give every item in this stack a unique ID
+  // Uses a Mutex as we can't generate a 8-byte random ID without hitting hostname length limits
+  let unique_id = {
+    let unique_id_mutex = UNIQUE_ID.get_or_init(|| Mutex::new(0));
+    let mut unique_id_lock = unique_id_mutex.lock().unwrap();
+    let unique_id = hex::encode(unique_id_lock.to_be_bytes());
+    *unique_id_lock += 1;
+    unique_id
+  };
+
+  let mut compositions = vec![];
+  let mut handles = vec![];
+  for composition in [serai_composition, message_queue_composition, coordinator_composition] {
+    let handle = composition.handle();
+    compositions.push(
+      composition
+        .with_start_policy(StartPolicy::Strict)
+        .with_container_name(format!("{handle}-{}", &unique_id))
+        .with_log_options(Some(LogOptions {
+          action: LogAction::Forward,
+          policy: if handle.contains("coordinator") {
+            LogPolicy::Always
+          } else {
+            LogPolicy::OnError
+          },
+          source: LogSource::Both,
+        })),
+    );
+    handles.push(compositions.last().unwrap().handle());
+  }
+
+  let coordinator_composition = compositions.last_mut().unwrap();
+  coordinator_composition.inject_container_name(handles.remove(0), "SERAI_HOSTNAME");
+  coordinator_composition.inject_container_name(handles.remove(0), "MESSAGE_QUEUE_RPC");
+
+  (
+    (compositions[0].handle(), compositions[1].handle(), compositions[2].handle()),
+    coord_key,
+    compositions,
+  )
+}

tests/coordinator/src/tests/mod.rs

@@ -0,0 +1,41 @@
+use std::time::Duration;
+
+use ciphersuite::{Ciphersuite, Ristretto};
+
+use dockertest::DockerTest;
+
+use crate::*;
+
+pub(crate) const COORDINATORS: usize = 4;
+// pub(crate) const THRESHOLD: usize = ((COORDINATORS * 2) / 3) + 1;
+
+fn new_test() -> (Vec<(Handles, <Ristretto as Ciphersuite>::F)>, DockerTest) {
+  let mut coordinators = vec![];
+  let mut test = DockerTest::new();
+  for i in 0 .. COORDINATORS {
+    let (handles, coord_key, compositions) = coordinator_stack(match i {
+      0 => "alice",
+      1 => "bob",
+      2 => "charlie",
+      3 => "dave",
+      4 => "eve",
+      5 => "ferdie",
+      _ => panic!("needed a 6th name for a serai node"),
+    });
+    coordinators.push((handles, coord_key));
+    for composition in compositions {
+      test.add_composition(composition);
+    }
+  }
+  (coordinators, test)
+}
+
+
+#[test]
+fn stack_test() {
+  let (_coordinators, test) = new_test();
+
+  test.run(|_ops| async move {
+    tokio::time::sleep(Duration::from_secs(30)).await;
+  });
+}

tests/docker/src/lib.rs

@@ -79,11 +79,24 @@ pub fn build(name: String) {
           meta(repo_path.join("message-queue")),
           meta(repo_path.join("processor")),
         ],
+        "coordinator" => vec![
+          meta(repo_path.join("common")),
+          meta(repo_path.join("crypto")),
+          meta(repo_path.join("coins")),
+          meta(repo_path.join("substrate")),
+          meta(repo_path.join("message-queue")),
+          meta(repo_path.join("coordinator")),
+        ],
         "runtime" => vec![
           meta(repo_path.join("common")),
           meta(repo_path.join("crypto")),
           meta(repo_path.join("substrate")),
         ],
+        "serai" => vec![
+          meta(repo_path.join("common")),
+          meta(repo_path.join("crypto")),
+          meta(repo_path.join("substrate")),
+        ],
         _ => panic!("building unrecognized docker image"),
       };