feat(geth-engine): implement a type inferencer for eventql

geth-domain/src/index/mem_table.rs

@@ -34,7 +34,7 @@ impl MemTable {
         self.entries_count += 1;
     }
 
-    pub fn scan_forward(&self, key: u64, start: u64, max: usize) -> ScanForward {
+    pub fn scan_forward(&self, key: u64, start: u64, max: usize) -> ScanForward<'_> {
         ScanForward {
             key,
             start,
@@ -44,7 +44,7 @@ impl MemTable {
         }
     }
 
-    pub fn scan_backward(&self, key: u64, start: u64, max: usize) -> ScanBackward {
+    pub fn scan_backward(&self, key: u64, start: u64, max: usize) -> ScanBackward<'_> {
         ScanBackward {
             key,
             start,

geth-domain/src/index/ss_table.rs

@@ -288,7 +288,7 @@ impl SsTable {
         self.iter().map(|t| (t.key, t.revision, t.position))
     }
 
-    pub fn scan_forward(&self, key: u64, start: u64, count: usize) -> ScanForward {
+    pub fn scan_forward(&self, key: u64, start: u64, count: usize) -> ScanForward<'_> {
         ScanForward {
             key,
             revision: start,
@@ -300,7 +300,7 @@ impl SsTable {
         }
     }
 
-    pub fn scan_backward(&self, key: u64, start: u64, count: usize) -> ScanBackward {
+    pub fn scan_backward(&self, key: u64, start: u64, count: usize) -> ScanBackward<'_> {
         let mut candidates = self.find_best_candidates(key, start);
 
         candidates.rotate_left(candidates.len() - 1);

geth-engine/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "geth-engine"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -17,6 +17,9 @@ path = "../geth-grpc"
 [dependencies.geth-domain]
 path = "../geth-domain"
 
+[dependencies.geth-eventql]
+path = "../geth-eventql"
+
 [dependencies.tokio]
 version = "1.20"
 features = ["full"]

geth-engine/src/lib.rs

@@ -8,26 +8,26 @@ mod options;
 mod process;
 
 use geth_mikoshi::{
-    storage::Storage, wal::chunks::ChunkContainer, FileSystemStorage, InMemoryStorage,
+    FileSystemStorage, InMemoryStorage, storage::Storage, wal::chunks::ChunkContainer,
 };
-use opentelemetry::{trace::TracerProvider, KeyValue};
+use opentelemetry::{KeyValue, trace::TracerProvider};
 use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
 use opentelemetry_otlp::WithExportConfig;
 use opentelemetry_sdk::{
-    logs::SdkLoggerProvider, metrics::PeriodicReader, trace::SdkTracerProvider, Resource,
+    Resource, logs::SdkLoggerProvider, metrics::PeriodicReader, trace::SdkTracerProvider,
 };
 pub use process::{
+    Proc, RequestContext,
     indexing::IndexClient,
-    manager::{start_process_manager_with_catalog, Catalog, CatalogBuilder, ManagerClient},
+    manager::{Catalog, CatalogBuilder, ManagerClient, start_process_manager_with_catalog},
     reading::{self, ReaderClient},
     start_process_manager,
     writing::WriterClient,
-    Proc, RequestContext,
 };
 use tokio::sync::OnceCell;
 use tracing_opentelemetry::OpenTelemetryLayer;
+use tracing_subscriber::{EnvFilter, prelude::*};
 use tracing_subscriber::{filter::filter_fn, layer::SubscriberExt};
-use tracing_subscriber::{prelude::*, EnvFilter};
 
 pub mod built_info {
     // The file has been placed there by the build script.

geth-engine/src/metrics.rs

@@ -1,5 +1,5 @@
 use std::{
-    sync::{mpsc, Arc, RwLock},
+    sync::{Arc, RwLock, mpsc},
     thread,
     time::Duration,
 };

geth-engine/src/process.rs

@@ -3,8 +3,8 @@ use std::time::Instant;
 use tokio::sync::mpsc::UnboundedSender;
 use uuid::Uuid;
 
-use crate::process::manager::{start_process_manager_with_catalog, Catalog};
 use crate::Options;
+use crate::process::manager::{Catalog, start_process_manager_with_catalog};
 
 #[cfg(test)]
 mod tests;
@@ -19,6 +19,7 @@ pub mod manager;
 mod messages;
 #[cfg(test)]
 mod panic;
+pub mod query;
 pub mod reading;
 #[cfg(test)]
 mod sink;
@@ -82,6 +83,7 @@ pub enum Proc {
     PubSub,
     Grpc,
     PyroWorker,
+    Query,
     #[cfg(test)]
     Echo,
     #[cfg(test)]

geth-engine/src/process/consumer.rs

@@ -8,8 +8,9 @@ use tokio::select;
 use tracing::instrument;
 
 use crate::{
+    IndexClient, ManagerClient, ReaderClient, RequestContext,
     process::subscription::{self, SubscriptionClient},
-    reading, IndexClient, ManagerClient, ReaderClient, RequestContext,
+    reading,
 };
 
 #[derive(Clone, Copy, Debug)]
@@ -222,10 +223,10 @@ impl Consumer {
 
                 State::Live => {
                     if let Some(event) = self.sub_streaming.next().await? {
-                        if let SubscriptionEvent::EventAppeared(temp) = &event {
-                            if temp.revision < self.end {
-                                continue;
-                            }
+                        if let SubscriptionEvent::EventAppeared(temp) = &event
+                            && temp.revision < self.end
+                        {
+                            continue;
                         }
 
                         return Ok(Some(event));

geth-engine/src/process/echo.rs

@@ -1,4 +1,4 @@
-use crate::process::{env::Managed, Item, ProcessEnv};
+use crate::process::{Item, ProcessEnv, env::Managed};
 
 pub async fn run(mut env: ProcessEnv<Managed>) -> eyre::Result<()> {
     while let Some(item) = env.recv().await {

geth-engine/src/process/env.rs

@@ -7,8 +7,8 @@ use tokio::{
 };
 
 use crate::{
-    process::{subscription::SubscriptionClient, Item},
     IndexClient, ManagerClient, Options, Proc,
+    process::{Item, subscription::SubscriptionClient},
 };
 
 pub struct Managed {

geth-engine/src/process/grpc/mod.rs

@@ -1,15 +1,15 @@
 use std::{pin::Pin, sync::Arc};
 
 use tokio::sync::Notify;
-use tonic::{transport::Server, Code, Status};
+use tonic::{Code, Status, transport::Server};
 
 use geth_grpc::generated::protocol::protocol_server::ProtocolServer;
 use tracing::instrument;
 
 use crate::{
-    metrics::{get_metrics, Metrics},
-    process::{manager::ManagerClient, Managed, ProcessEnv},
     Options,
+    metrics::{Metrics, get_metrics},
+    process::{Managed, ProcessEnv, manager::ManagerClient},
 };
 
 mod protocol;
@@ -116,13 +116,13 @@ where
 
         Box::pin(async move {
             let resp = inner.call(req).await?;
-            if let Some(status) = Status::from_header_map(resp.headers()) {
-                if status.code() != Code::Ok {
-                    if is_client_error(status.code()) {
-                        metrics.observe_client_error();
-                    } else if is_server_error(status.code()) {
-                        metrics.observe_server_error();
-                    }
+            if let Some(status) = Status::from_header_map(resp.headers())
+                && status.code() != Code::Ok
+            {
+                if is_client_error(status.code()) {
+                    metrics.observe_client_error();
+                } else if is_server_error(status.code()) {
+                    metrics.observe_server_error();
                 }
             }
 

geth-engine/src/process/grpc/protocol.rs

@@ -12,7 +12,7 @@ use tonic::{Request, Response, Status};
 use uuid::Uuid;
 
 use crate::metrics::get_metrics;
-use crate::process::consumer::{start_consumer, ConsumerResult};
+use crate::process::consumer::{ConsumerResult, start_consumer};
 use crate::process::reading::ReaderClient;
 use crate::process::subscription::SubscriptionClient;
 use crate::process::writing::WriterClient;
@@ -172,7 +172,7 @@ impl Protocol for ProtocolImpl {
                     Ok(result) => match result {
                         ConsumerResult::Success(c) => c,
                         ConsumerResult::StreamDeleted => {
-                            return Err(Status::failed_precondition("stream-deleted"))
+                            return Err(Status::failed_precondition("stream-deleted"));
                         }
                     },
                 };

geth-engine/src/process/indexing/client.rs

@@ -60,7 +60,7 @@ impl IndexClient {
                     return Ok(ReadCompleted::Success(Streaming {
                         inner,
                         batch: Some(entries.into_iter()),
-                    }))
+                    }));
                 }
 
                 _ => {
@@ -134,7 +134,9 @@ impl IndexClient {
                 }
 
                 _ => {
-                    eyre::bail!("unexpected response when fetching the latest revision from the index process");
+                    eyre::bail!(
+                        "unexpected response when fetching the latest revision from the index process"
+                    );
                 }
             }
         }

geth-engine/src/process/indexing/proc.rs

@@ -9,8 +9,8 @@ use geth_common::{Direction, IteratorIO};
 use geth_domain::index::BlockEntry;
 use geth_domain::{Lsm, LsmSettings};
 use geth_mikoshi::hashing::mikoshi_hash;
-use geth_mikoshi::wal::chunks::ChunkContainer;
 use geth_mikoshi::wal::LogReader;
+use geth_mikoshi::wal::chunks::ChunkContainer;
 use std::cmp::min;
 use std::sync::{Arc, RwLock};
 use std::{io, mem};

geth-engine/src/process/manager/catalog.rs

@@ -9,8 +9,8 @@ use tokio::sync::oneshot;
 use uuid::Uuid;
 
 use crate::{
-    process::{messages::Messages, Item, Mail, ProcId, RunningProc, SpawnError, SpawnResult},
     Proc, RequestContext,
+    process::{Item, Mail, ProcId, RunningProc, SpawnError, SpawnResult, messages::Messages},
 };
 
 #[derive(Debug)]

geth-engine/src/process/manager/client.rs

@@ -2,23 +2,23 @@ use std::time::{Duration, Instant};
 
 use geth_common::ProgramSummary;
 use tokio::sync::{
-    mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
+    mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel},
     oneshot,
 };
 use tracing::instrument;
 use uuid::Uuid;
 
 use crate::{
+    IndexClient, Proc, ReaderClient, RequestContext, WriterClient,
     process::{
+        Item, Mail, ProcId, RunningProc, SpawnResult, Stream,
         manager::{
             FindParams, ManagerCommand, ProcReadyParams, ProcTerminatedParams, SendParams,
             ShutdownNotification, ShutdownParams, TimeoutParams, TimeoutTarget, WaitForParams,
         },
         messages::Messages,
         subscription::SubscriptionClient,
-        Item, Mail, ProcId, RunningProc, SpawnResult, Stream,
     },
-    IndexClient, Proc, ReaderClient, RequestContext, WriterClient,
 };
 
 #[derive(Clone, Debug)]

geth-engine/src/process/manager/mod.rs

@@ -1,27 +1,27 @@
 use std::{
     collections::HashMap,
     sync::{
-        atomic::{AtomicBool, Ordering},
         Arc,
+        atomic::{AtomicBool, Ordering},
     },
     time::{Duration, Instant},
 };
 
 use geth_common::ProgramSummary;
-use tokio::sync::{oneshot, Notify};
+use tokio::sync::{Notify, oneshot};
 use uuid::Uuid;
 
 use crate::{
+    Options, Proc, RequestContext,
     process::{
+        Item, Mail, ProcId, RunningProc, SpawnError, SpawnResult,
         manager::{
             catalog::ProvisionResult,
             proc::process_manager,
-            spawn::{spawn_process, SpawnParams},
+            spawn::{SpawnParams, spawn_process},
         },
         messages::{Messages, Notifications, Responses},
-        Item, Mail, ProcId, RunningProc, SpawnError, SpawnResult,
     },
-    Options, Proc, RequestContext,
 };
 
 mod catalog;
@@ -187,13 +187,13 @@ impl Manager {
             }
 
             Item::Stream(stream) => {
-                if let Some(proc) = self.catalog.get_process(cmd.dest) {
-                    if !proc.mailbox.send(Item::Stream(stream)) {
-                        self.handle_terminate(ProcTerminatedParams {
-                            id: cmd.dest,
-                            error: None,
-                        });
-                    }
+                if let Some(proc) = self.catalog.get_process(cmd.dest)
+                    && !proc.mailbox.send(Item::Stream(stream))
+                {
+                    self.handle_terminate(ProcTerminatedParams {
+                        id: cmd.dest,
+                        error: None,
+                    });
                 }
             }
         }
@@ -273,20 +273,19 @@ impl Manager {
             }
 
             for dependent in running.dependents {
-                if let Some(running) = self.catalog.get_process(dependent) {
-                    if !self.closing
-                        && !running.mailbox.send(Item::Mail(Mail {
-                            context: RequestContext::new(),
-                            origin: 0,
-                            correlation: Uuid::nil(),
-                            payload: Notifications::ProcessTerminated(cmd.id).into(),
-                            created: Instant::now(),
-                        }))
-                    {
-                        // I don't want to call `handle_terminate` here because it could end up blowing up the stack.
-                        // I could rewrite `handle_terminate` to avoid recursion.
-                        tracing::warn!(id = dependent, proc = ?running.proc, closing = self.closing, "process seems to be terminated");
-                    }
+                if let Some(running) = self.catalog.get_process(dependent)
+                    && !self.closing
+                    && !running.mailbox.send(Item::Mail(Mail {
+                        context: RequestContext::new(),
+                        origin: 0,
+                        correlation: Uuid::nil(),
+                        payload: Notifications::ProcessTerminated(cmd.id).into(),
+                        created: Instant::now(),
+                    }))
+                {
+                    // I don't want to call `handle_terminate` here because it could end up blowing up the stack.
+                    // I could rewrite `handle_terminate` to avoid recursion.
+                    tracing::warn!(id = dependent, proc = ?running.proc, closing = self.closing, "process seems to be terminated");
                 }
             }
         } else if !self.closing {