refactor: Framework interface improvements (matter-labs#975)

Renamed:
- `IntoZkSyncTask` -> `WiringLayer` (as this trait now does more than
just creating a single task)
- `IntoZkSyncTask::create` -> `WiringLayer::wire` (layer doesn't have to
create anything)
- `ZkSyncTask` -> `Task` (was unnecessary verbose and inconsistent with
simple `Resource`)
- `ZkSyncNode` -> `ZkStackService` (1) not tied to zkSync; 2) several
people told me that it doesn't have to represent the whole node)
- `ZkSyncNode::add_task` -> `ZKStackService::add_layer`

Moved:
- `core/lib/node` -> `core/node/node_framework` (as we previously agreed
to create a separate directory for things that define the node)

Changed:
- `WiringLayer` no longer returns a task, instead, it adds any number of
tasks through the node context.

Cargo.toml

@@ -10,6 +10,8 @@ members = [
     "core/bin/system-constants-generator",
     "core/bin/verified_sources_fetcher",
     "core/bin/zksync_server",
+    # Node services
+    "core/node/node_framework",
     # Libraries
     "core/lib/zksync_core",
     "core/lib/basic_types",
@@ -27,7 +29,6 @@ members = [
     "core/lib/mempool",
     "core/lib/merkle_tree",
     "core/lib/mini_merkle_tree",
-    "core/lib/node",
     "core/lib/object_store",
     "core/lib/prometheus_exporter",
     "core/lib/prover_interface",

core/node/node_framework/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "zksync_node_framework"
+version = "0.1.0"
+edition = "2018"
+authors = ["The Matter Labs Team <[email protected]>"]
+homepage = "https://zksync.io/"
+repository = "https://github.com/matter-labs/zksync-era"
+license = "MIT OR Apache-2.0"
+keywords = ["blockchain", "zksync"]
+categories = ["cryptography"]
+
+[dependencies]
+prometheus_exporter = { path = "../../lib/prometheus_exporter" }
+zksync_types = { path = "../../lib/types" }
+zksync_health_check = { path = "../../lib/health_check" }
+zksync_dal = { path = "../../lib/dal" }
+zksync_config = { path = "../../lib/config" }
+zksync_object_store = { path = "../../lib/object_store" }
+zksync_core = { path = "../../lib/zksync_core" }
+zksync_storage = { path = "../../lib/storage" }
+
+tracing = "0.1"
+thiserror = "1"
+async-trait = "0.1"
+futures = "0.3"
+anyhow = "1"
+tokio = { version = "1", features = ["rt"] }
+
+[dev-dependencies]
+zksync_env_config = { path = "../../lib/env_config" }
+vlog = { path = "../../lib/vlog" }

core/lib/node/examples/main_node.rs → core/node/node_framework/examples/main_node.rs

@@ -6,16 +6,15 @@ use zksync_config::{configs::chain::OperationsManagerConfig, DBConfig, PostgresC
 use zksync_core::metadata_calculator::MetadataCalculatorConfig;
 use zksync_dal::ConnectionPool;
 use zksync_env_config::FromEnv;
-use zksync_node::{
+use zksync_node_framework::{
     implementations::{
-        resource::pools::MasterPoolResource,
-        task::{
-            healtcheck_server::HealthCheckTaskBuilder,
-            metadata_calculator::MetadataCalculatorTaskBuilder,
+        layers::{
+            healtcheck_server::HealthCheckLayer, metadata_calculator::MetadataCalculatorLayer,
         },
+        resources::pools::MasterPoolResource,
     },
-    node::ZkSyncNode,
     resource::{Resource, ResourceId, ResourceProvider, StoredResource},
+    service::ZkStackService,
 };
 
 /// Resource provider for the main node.
@@ -62,7 +61,7 @@ fn main() -> anyhow::Result<()> {
     // Create the node with specified resource provider. We don't need to add any resources explicitly,
     // the task will request what they actually need. The benefit here is that we won't instantiate resources
     // that are not used, which would be complex otherwise, since the task set is often dynamic.
-    let mut node = ZkSyncNode::new(MainNodeResourceProvider)?;
+    let mut node = ZkStackService::new(MainNodeResourceProvider)?;
 
     // Add the metadata calculator task.
     let merkle_tree_env_config = DBConfig::from_env()?.merkle_tree;
@@ -71,11 +70,11 @@ fn main() -> anyhow::Result<()> {
         &merkle_tree_env_config,
         &operations_manager_env_config,
     );
-    node.add_task(MetadataCalculatorTaskBuilder(metadata_calculator_config));
+    node.add_layer(MetadataCalculatorLayer(metadata_calculator_config));
 
     // Add the healthcheck server.
     let healthcheck_config = zksync_config::ApiConfig::from_env()?.healthcheck;
-    node.add_task(HealthCheckTaskBuilder(healthcheck_config));
+    node.add_layer(HealthCheckLayer(healthcheck_config));
 
     // Run the node until completion.
     node.run()?;

core/lib/node/src/implementations/task/healtcheck_server.rs → core/node/node_framework/src/implementations/layers/healtcheck_server.rs

@@ -4,29 +4,34 @@ use zksync_config::configs::api::HealthCheckConfig;
 use zksync_core::api_server::healthcheck::HealthCheckHandle;
 
 use crate::{
-    implementations::resource::healthcheck::HealthCheckResource,
-    node::{NodeContext, StopReceiver},
+    implementations::resources::healthcheck::HealthCheckResource,
     resource::ResourceCollection,
-    task::{IntoZkSyncTask, TaskInitError, ZkSyncTask},
+    service::{ServiceContext, StopReceiver},
+    task::Task,
+    wiring_layer::{WiringError, WiringLayer},
 };
 
 /// Builder for a health check server.
 ///
 /// Spawned task collects all the health checks added by different tasks to the
 /// corresponding resource collection and spawns an HTTP server exposing them.
+///
+/// This layer expects other tasks to add health checks to the `ResourceCollection<HealthCheckResource>`.
+///
+/// ## Effects
+///
+/// - Resolves `ResourceCollection<HealthCheckResource>`.
+/// - Adds `healthcheck_server` to the node.
 #[derive(Debug)]
-pub struct HealthCheckTaskBuilder(pub HealthCheckConfig);
+pub struct HealthCheckLayer(pub HealthCheckConfig);
 
 #[async_trait::async_trait]
-impl IntoZkSyncTask for HealthCheckTaskBuilder {
-    fn task_name(&self) -> &'static str {
-        "healthcheck_server"
+impl WiringLayer for HealthCheckLayer {
+    fn layer_name(&self) -> &'static str {
+        "healthcheck_layer"
     }
 
-    async fn create(
-        self: Box<Self>,
-        mut node: NodeContext<'_>,
-    ) -> Result<Box<dyn ZkSyncTask>, TaskInitError> {
+    async fn wire(self: Box<Self>, mut node: ServiceContext<'_>) -> Result<(), WiringError> {
         let healthchecks = node
             .get_resource_or_default::<ResourceCollection<HealthCheckResource>>()
             .await;
@@ -36,7 +41,8 @@ impl IntoZkSyncTask for HealthCheckTaskBuilder {
             healthchecks,
         };
 
-        Ok(Box::new(task))
+        node.add_task(Box::new(task));
+        Ok(())
     }
 }
 
@@ -54,7 +60,11 @@ impl fmt::Debug for HealthCheckTask {
 }
 
 #[async_trait::async_trait]
-impl ZkSyncTask for HealthCheckTask {
+impl Task for HealthCheckTask {
+    fn name(&self) -> &'static str {
+        "healthcheck_server"
+    }
+
     async fn run(mut self: Box<Self>, mut stop_receiver: StopReceiver) -> anyhow::Result<()> {
         let healthchecks = self.healthchecks.resolve().await;
 

core/lib/node/src/implementations/task/metadata_calculator.rs → core/node/node_framework/src/implementations/layers/metadata_calculator.rs

@@ -3,18 +3,26 @@ use zksync_dal::ConnectionPool;
 use zksync_storage::RocksDB;
 
 use crate::{
-    implementations::resource::{
+    implementations::resources::{
         healthcheck::HealthCheckResource, object_store::ObjectStoreResource,
         pools::MasterPoolResource,
     },
-    node::{NodeContext, StopReceiver},
     resource::{Resource, ResourceCollection},
-    task::{IntoZkSyncTask, TaskInitError, ZkSyncTask},
+    service::{ServiceContext, StopReceiver},
+    task::Task,
+    wiring_layer::{WiringError, WiringLayer},
 };
 
 /// Builder for a metadata calculator.
+///
+/// ## Effects
+///
+/// - Resolves `MasterPoolResource`.
+/// - Resolves `ObjectStoreResource` (optional).
+/// - Adds `tree_health_check` to the `ResourceCollection<HealthCheckResource>`.
+/// - Adds `metadata_calculator` to the node.
 #[derive(Debug)]
-pub struct MetadataCalculatorTaskBuilder(pub MetadataCalculatorConfig);
+pub struct MetadataCalculatorLayer(pub MetadataCalculatorConfig);
 
 #[derive(Debug)]
 pub struct MetadataCalculatorTask {
@@ -23,18 +31,18 @@ pub struct MetadataCalculatorTask {
 }
 
 #[async_trait::async_trait]
-impl IntoZkSyncTask for MetadataCalculatorTaskBuilder {
-    fn task_name(&self) -> &'static str {
-        "metadata_calculator"
+impl WiringLayer for MetadataCalculatorLayer {
+    fn layer_name(&self) -> &'static str {
+        "metadata_calculator_layer"
     }
 
-    async fn create(
-        self: Box<Self>,
-        mut node: NodeContext<'_>,
-    ) -> Result<Box<dyn ZkSyncTask>, TaskInitError> {
-        let pool = node.get_resource::<MasterPoolResource>().await.ok_or(
-            TaskInitError::ResourceLacking(MasterPoolResource::resource_id()),
-        )?;
+    async fn wire(self: Box<Self>, mut node: ServiceContext<'_>) -> Result<(), WiringError> {
+        let pool =
+            node.get_resource::<MasterPoolResource>()
+                .await
+                .ok_or(WiringError::ResourceLacking(
+                    MasterPoolResource::resource_id(),
+                ))?;
         let main_pool = pool.get().await.unwrap();
         let object_store = node.get_resource::<ObjectStoreResource>().await; // OK to be None.
 
@@ -56,15 +64,21 @@ impl IntoZkSyncTask for MetadataCalculatorTaskBuilder {
             ))
             .expect("Wiring stage");
 
-        Ok(Box::new(MetadataCalculatorTask {
+        let task = Box::new(MetadataCalculatorTask {
             metadata_calculator,
             main_pool,
-        }))
+        });
+        node.add_task(task);
+        Ok(())
     }
 }
 
 #[async_trait::async_trait]
-impl ZkSyncTask for MetadataCalculatorTask {
+impl Task for MetadataCalculatorTask {
+    fn name(&self) -> &'static str {
+        "metadata_calculator"
+    }
+
     async fn run(self: Box<Self>, stop_receiver: StopReceiver) -> anyhow::Result<()> {
         let result = self
             .metadata_calculator

core/lib/node/src/implementations/task/prometheus_exporter.rs → core/node/node_framework/src/implementations/layers/prometheus_exporter.rs

@@ -2,15 +2,21 @@ use prometheus_exporter::PrometheusExporterConfig;
 use zksync_health_check::{HealthStatus, HealthUpdater, ReactiveHealthCheck};
 
 use crate::{
-    implementations::resource::healthcheck::HealthCheckResource,
-    node::{NodeContext, StopReceiver},
+    implementations::resources::healthcheck::HealthCheckResource,
     resource::ResourceCollection,
-    task::{IntoZkSyncTask, TaskInitError, ZkSyncTask},
+    service::{ServiceContext, StopReceiver},
+    task::Task,
+    wiring_layer::{WiringError, WiringLayer},
 };
 
 /// Builder for a prometheus exporter.
+///
+/// ## Effects
+///
+/// - Adds prometheus health check to the `ResourceCollection<HealthCheckResource>`.
+/// - Adds `prometheus_exporter` to the node.
 #[derive(Debug)]
-pub struct PrometheusExporterTaskBuilder(pub PrometheusExporterConfig);
+pub struct PrometheusExporterLayer(pub PrometheusExporterConfig);
 
 #[derive(Debug)]
 pub struct PrometheusExporterTask {
@@ -19,15 +25,12 @@ pub struct PrometheusExporterTask {
 }
 
 #[async_trait::async_trait]
-impl IntoZkSyncTask for PrometheusExporterTaskBuilder {
-    fn task_name(&self) -> &'static str {
+impl WiringLayer for PrometheusExporterLayer {
+    fn layer_name(&self) -> &'static str {
         "prometheus_exporter"
     }
 
-    async fn create(
-        self: Box<Self>,
-        mut node: NodeContext<'_>,
-    ) -> Result<Box<dyn ZkSyncTask>, TaskInitError> {
+    async fn wire(self: Box<Self>, mut node: ServiceContext<'_>) -> Result<(), WiringError> {
         let (prometheus_health_check, prometheus_health_updater) =
             ReactiveHealthCheck::new("prometheus_exporter");
 
@@ -38,15 +41,22 @@ impl IntoZkSyncTask for PrometheusExporterTaskBuilder {
             .push(HealthCheckResource::new(prometheus_health_check))
             .expect("Wiring stage");
 
-        Ok(Box::new(PrometheusExporterTask {
+        let task = Box::new(PrometheusExporterTask {
             config: self.0,
             prometheus_health_updater,
-        }))
+        });
+
+        node.add_task(task);
+        Ok(())
     }
 }
 
 #[async_trait::async_trait]
-impl ZkSyncTask for PrometheusExporterTask {
+impl Task for PrometheusExporterTask {
+    fn name(&self) -> &'static str {
+        "prometheus_exporter"
+    }
+
     async fn run(self: Box<Self>, stop_receiver: StopReceiver) -> anyhow::Result<()> {
         let prometheus_task = self.config.run(stop_receiver.0);
         self.prometheus_health_updater

core/lib/node/src/implementations/mod.rs → core/node/node_framework/src/implementations/mod.rs

@@ -2,5 +2,5 @@
 //! These are temporarily provided by the framework crate itself, but will be moved to the separate crates
 //! in the future.
 
-pub mod resource;
-pub mod task;
+pub mod layers;
+pub mod resources;

core/lib/node/src/lib.rs → core/node/node_framework/src/lib.rs

@@ -4,21 +4,22 @@
 //!
 //! This crate provides core abstractions that allow one to compose a ZK Stack node.
 //! Main concepts used in this crate are:
-//! - [`IntoZkSyncTask`](task::IntoZkSyncTask) - builder interface for tasks.
-//! - [`ZkSyncTask`](task::ZkSyncTask) - a unit of work that can be executed by the node.
+//! - [`WiringLayer`](wiring_layer::WiringLayer) - builder interface for tasks.
+//! - [`Task`](task::Task) - a unit of work that can be executed by the node.
 //! - [`Resource`](resource::Resource) - a piece of logic that can be shared between tasks. Most resources are
 //!   represented by generic interfaces and also serve as points of customization for tasks.
 //! - [`ResourceProvider`](resource::ResourceProvider) - a trait that allows one to provide resources to the node.
-//! - [`ZkSyncNode`](node::ZkSyncNode) - a container for tasks and resources that takes care of initialization, running
+//! - [`ZkStackService`](service::ZkStackService) - a container for tasks and resources that takes care of initialization, running
 //!   and shutting down.
 //!
 //! The general flow to compose a node is as follows:
 //! - Create a [`ResourceProvider`](resource::ResourceProvider) that can provide all the resources that the node needs.
-//! - Create a [`ZkSyncNode`](node::ZkSyncNode) with that [`ResourceProvider`](resource::ResourceProvider).
+//! - Create a [`ZkStackService`](node::ZkStackService) with that [`ResourceProvider`](resource::ResourceProvider).
 //! - Add tasks to the node.
 //! - Run it.
 
 pub mod implementations;
-pub mod node;
 pub mod resource;
+pub mod service;
 pub mod task;
+pub mod wiring_layer;