chore: minor refactors

api/client/src/client.rs

@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use error_stack::Result;
-use error_stack::ResultExt;
-use error_stack::bail;
 use reqwest::IntoUrl;
 use reqwest::StatusCode;
 use reqwest::Url;
@@ -71,57 +68,58 @@ impl Client {
 }
 
 async fn do_get(client: &Client, key: &str) -> Result<Option<Vec<u8>>, Error> {
-    let make_error = || Error::Other("failed to get".to_string());
-
-    let url = client.base_url.join(key).change_context_lazy(make_error)?;
-    let resp = client
-        .client
-        .get(url)
-        .send()
-        .await
-        .change_context_lazy(make_error)?;
+    let url = client
+        .base_url
+        .join(key)
+        .map_err(|e| Error::Other(e.to_string()))?;
+    let resp = client.client.get(url).send().await.map_err(Error::Http)?;
 
     match resp.status() {
-        StatusCode::NOT_FOUND => Ok(None),
+        StatusCode::NOT_FOUND | StatusCode::TOO_MANY_REQUESTS => Ok(None),
         StatusCode::OK => {
-            let body = resp.bytes().await.change_context_lazy(make_error)?;
+            let body = resp.bytes().await.map_err(Error::Http)?;
             Ok(Some(body.to_vec()))
         }
-        _ => bail!(Error::Other(resp.status().to_string())),
+        _ => Err(Error::Other(resp.status().to_string())),
     }
 }
 
 async fn do_put(client: &Client, key: &str, value: &[u8]) -> Result<(), Error> {
-    let make_error = || Error::Other("failed to put".to_string());
-
-    let url = client.base_url.join(key).change_context_lazy(make_error)?;
+    let url = client
+        .base_url
+        .join(key)
+        .map_err(|e| Error::Other(e.to_string()))?;
     let resp = client
         .client
         .put(url)
         .body(value.to_vec())
         .send()
         .await
-        .change_context_lazy(make_error)?;
+        .map_err(Error::Http)?;
 
     match resp.status() {
-        StatusCode::OK | StatusCode::CREATED => Ok(()),
-        _ => bail!(Error::Other(resp.status().to_string())),
+        StatusCode::OK
+        | StatusCode::CREATED
+        | StatusCode::NO_CONTENT
+        | StatusCode::TOO_MANY_REQUESTS => Ok(()),
+        _ => Err(Error::Other(resp.status().to_string())),
     }
 }
 
 async fn do_delete(client: &Client, key: &str) -> Result<(), Error> {
-    let make_error = || Error::Other("failed to delete".to_string());
-
-    let url = client.base_url.join(key).change_context_lazy(make_error)?;
+    let url = client
+        .base_url
+        .join(key)
+        .map_err(|e| Error::Other(e.to_string()))?;
     let resp = client
         .client
         .delete(url)
         .send()
         .await
-        .change_context_lazy(make_error)?;
+        .map_err(Error::Http)?;
 
     match resp.status() {
-        StatusCode::OK | StatusCode::NO_CONTENT => Ok(()),
-        _ => bail!(Error::Other(resp.status().to_string())),
+        StatusCode::OK | StatusCode::NO_CONTENT | StatusCode::TOO_MANY_REQUESTS => Ok(()),
+        _ => Err(Error::Other(resp.status().to_string())),
     }
 }

cmd/percas/src/start.rs

@@ -59,7 +59,8 @@ impl CommandStart {
         log::info!("Percas is starting with loaded config: {config:#?}");
 
         let server_runtime = make_server_runtime();
-        server_runtime.block_on(run_server(&server_runtime, config))
+        let gossip_runtime = make_gossip_runtime();
+        server_runtime.block_on(run_server(&server_runtime, &gossip_runtime, config))
     }
 }
 
@@ -72,6 +73,10 @@ fn make_server_runtime() -> Runtime {
     make_runtime("server_runtime", "server_thread", parallelism)
 }
 
+fn make_gossip_runtime() -> Runtime {
+    make_runtime("gossip_runtime", "gossip_thread", 1)
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 enum ServerMode {
     Standalone,
@@ -129,7 +134,7 @@ impl From<&ServerConfig> for FlattenConfig {
     }
 }
 
-async fn run_server(rt: &Runtime, config: Config) -> Result<(), Error> {
+async fn run_server(server_rt: &Runtime, gossip_rt: &Runtime, config: Config) -> Result<(), Error> {
     let make_error = || Error("failed to start server".to_string());
 
     let engine = FoyerEngine::try_new(
@@ -193,7 +198,7 @@ async fn run_server(rt: &Runtime, config: Config) -> Result<(), Error> {
         // TODO: gracefully shutdown gossip
         gossip
             .clone()
-            .start(rt, listen_peer_addr.clone())
+            .start(gossip_rt, listen_peer_addr.clone())
             .await
             .change_context_lazy(make_error)?;
 
@@ -202,12 +207,15 @@ async fn run_server(rt: &Runtime, config: Config) -> Result<(), Error> {
         None
     };
 
-    let (server, shutdown_tx) =
-        percas_server::server::start_server(rt, ctx, listen_addr, advertise_addr, cluster_proxy)
-            .await
-            .change_context_lazy(|| {
-                Error("A fatal error has occurred in server process.".to_string())
-            })?;
+    let (server, shutdown_tx) = percas_server::server::start_server(
+        server_rt,
+        ctx,
+        listen_addr,
+        advertise_addr,
+        cluster_proxy,
+    )
+    .await
+    .change_context_lazy(|| Error("A fatal error has occurred in server process.".to_string()))?;
 
     ctrlc::set_handler(move || shutdown_tx.shutdown())
         .change_context_lazy(|| Error("failed to setup ctrl-c signal handle".to_string()))?;

crates/cluster/src/proxy.rs

@@ -57,7 +57,10 @@ impl Proxy {
                 RouteDest::Local
             }
         } else {
-            log::error!("no target found for key: {key}");
+            log::debug!(
+                "no target found for key: [{key}] , current ring: {:#?}",
+                ring
+            );
             RouteDest::Local
         }
     }

crates/core/src/engine.rs

@@ -24,13 +24,12 @@ use foyer::HybridCacheBuilder;
 use foyer::HybridCachePolicy;
 use foyer::RecoverMode;
 use foyer::RuntimeOptions;
-use foyer::TokioRuntimeOptions;
 use sysinfo::Pid;
 use thiserror::Error;
 
 use crate::num_cpus;
 
-const DEFAULT_MEMORY_CAPACITY_FACTOR: f64 = 0.9;
+const DEFAULT_MEMORY_CAPACITY_FACTOR: f64 = 0.5;
 
 #[derive(Debug, Error)]
 #[error("{0}")]
@@ -76,13 +75,10 @@ impl FoyerEngine {
             .with_device_options(
                 DirectFsDeviceOptions::new(data_dir)
                     .with_capacity(disk_capacity as usize)
-                    .with_file_size(32 * 1024 * 1024),
+                    .with_file_size(64 * 1024 * 1024),
             )
             .with_recover_mode(RecoverMode::Quiet)
-            .with_runtime_options(RuntimeOptions::Unified(TokioRuntimeOptions {
-                worker_threads: parallelism,
-                max_blocking_threads: parallelism * 2,
-            }))
+            .with_runtime_options(RuntimeOptions::Unified(Default::default()))
             .build()
             .await
             .map_err(|err| report!(EngineError(err.to_string())))?;

crates/server/src/lib.rs

@@ -14,6 +14,7 @@
 
 use percas_core::FoyerEngine;
 
+pub mod middleware;
 pub mod scheduled;
 pub mod server;
 pub mod telemetry;