fix: Prevent Coordinator from stopping V1 executors

coordinator/src/block_streams_handler.rs

@@ -1,3 +1,5 @@
+#![cfg_attr(test, allow(dead_code))]
+
 use anyhow::Context;
 use block_streamer::block_streamer_client::BlockStreamerClient;
 use block_streamer::{

coordinator/src/executors_handler.rs

@@ -1,3 +1,5 @@
+#![cfg_attr(test, allow(dead_code))]
+
 use anyhow::Context;
 use runner::runner_client::RunnerClient;
 use runner::{ExecutorInfo, ListExecutorsRequest, StartExecutorRequest, StopExecutorRequest};

coordinator/src/main.rs

@@ -17,6 +17,7 @@ mod registry;
 mod utils;
 
 const CONTROL_LOOP_THROTTLE_SECONDS: Duration = Duration::from_secs(1);
+const V1_EXECUTOR_VERSION: u64 = 0;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -81,23 +82,31 @@ async fn filter_registry_by_allowlist(
     let allowlist: AllowList =
         serde_json::from_str(&raw_allowlist).context("Failed to parse allowlist")?;
 
-    tracing::debug!("Using allowlist: {:#?}", allowlist);
-
-    Ok(indexer_registry
+    let filtered_registry = indexer_registry
         .into_iter()
         .filter(|(account_id, _)| {
             allowlist
                 .iter()
                 .any(|entry| entry.account_id == *account_id)
         })
-        .collect())
+        .collect();
+
+    tracing::debug!("Using filtered registry: {:#?}", filtered_registry);
+
+    Ok(filtered_registry)
 }
 
 async fn synchronise_executors(
     indexer_registry: &IndexerRegistry,
     executors_handler: &ExecutorsHandler,
 ) -> anyhow::Result<()> {
-    let mut active_executors = executors_handler.list().await?;
+    let active_executors = executors_handler.list().await?;
+
+    // Ignore V1 executors
+    let mut active_executors: Vec<_> = active_executors
+        .into_iter()
+        .filter(|executor| executor.version != V1_EXECUTOR_VERSION)
+        .collect();
 
     for (account_id, indexers) in indexer_registry.iter() {
         for (function_name, indexer_config) in indexers.iter() {

coordinator/src/redis.rs

@@ -1,3 +1,5 @@
+#![cfg_attr(test, allow(dead_code))]
+
 use std::fmt::Debug;
 
 use anyhow::Context;

coordinator/src/registry.rs

@@ -1,3 +1,5 @@
+#![cfg_attr(test, allow(dead_code))]
+
 use anyhow::Context;
 use std::collections::HashMap;
 

runner/src/server/runner-service.ts

@@ -102,7 +102,7 @@ function getRunnerService (executors: Map<string, StreamHandler>, StreamHandlerT
             config = {
               account_id: accountId,
               function_name: functionName,
-              version: -1, // Ensure Coordinator V2 sees version mismatch
+              version: 0, // Ensure Coordinator V2 sees version mismatch
               code: '',
               schema: '',
             };