database: impl service::{Request,Response} mappings

database/Cargo.toml

@@ -25,6 +25,7 @@ cfg-if   = { workspace = true }
 # We only need the `thread` feature if `service` is enabled.
 # Figure out how to enable features of an already pulled in dependency conditionally.
 cuprate-helper = { path = "../helper", features = ["fs", "thread"] }
+cuprate-types  = { path = "../types", features = ["service"] }
 paste          = { workspace = true }
 page_size      = { version = "0.6.0" } # Needed for database resizes, they must be a multiple of the OS page size.
 thiserror      = { workspace = true }

database/src/backend/heed/env.rs

@@ -249,11 +249,11 @@ impl Env for ConcreteEnv {
         Ok(self.env.read().unwrap().force_sync()?)
     }
 
-    fn resize_map(&self, resize_algorithm: Option<ResizeAlgorithm>) {
+    fn resize_map(&self, resize_algorithm: Option<ResizeAlgorithm>) -> NonZeroUsize {
         let resize_algorithm = resize_algorithm.unwrap_or_else(|| self.config().resize_algorithm);
 
         let current_size_bytes = self.current_map_size();
-        let new_size_bytes = resize_algorithm.resize(current_size_bytes).get();
+        let new_size_bytes = resize_algorithm.resize(current_size_bytes);
 
         // SAFETY:
         // Resizing requires that we have
@@ -264,8 +264,14 @@ impl Env for ConcreteEnv {
         // <http://www.lmdb.tech/doc/group__mdb.html#gaa2506ec8dab3d969b0e609cd82e619e5>
         unsafe {
             // INVARIANT: `resize()` returns a valid `usize` to resize to.
-            self.env.write().unwrap().resize(new_size_bytes).unwrap();
+            self.env
+                .write()
+                .unwrap()
+                .resize(new_size_bytes.get())
+                .unwrap();
         }
+
+        new_size_bytes
     }
 
     #[inline]

database/src/env.rs

@@ -1,7 +1,7 @@
 //! Abstracted database environment; `trait Env`.
 
 //---------------------------------------------------------------------------------------------------- Import
-use std::{fmt::Debug, ops::Deref};
+use std::{fmt::Debug, num::NonZeroUsize, ops::Deref};
 
 use crate::{
     config::Config,
@@ -113,7 +113,7 @@ pub trait Env: Sized {
     /// This function _must_ be re-implemented if [`Env::MANUAL_RESIZE`] is `true`.
     ///
     /// Otherwise, this function will panic with `unreachable!()`.
-    fn resize_map(&self, resize_algorithm: Option<ResizeAlgorithm>) {
+    fn resize_map(&self, resize_algorithm: Option<ResizeAlgorithm>) -> NonZeroUsize {
         unreachable!()
     }
 

database/src/lib.rs

@@ -100,6 +100,8 @@
 //!     config::Config,
 //!     ConcreteEnv,
 //!     Env, Key, TxRo, TxRw,
+//! };
+//! use cuprate_types::{
 //!     service::{ReadRequest, WriteRequest, Response},
 //! };
 //!

database/src/service/mod.rs

@@ -5,7 +5,7 @@
 //! along with the reader/writer thread-pool system.
 //!
 //! The thread-pool allows outside crates to communicate with it by
-//! sending database [`Request`](ReadRequest)s and receiving [`Response`]s `async`hronously -
+//! sending database [`Request`][req_r]s and receiving [`Response`][resp]s `async`hronously -
 //! without having to actually worry and handle the database themselves.
 //!
 //! The system is managed by this crate, and only requires [`init`] by the user.
@@ -17,9 +17,9 @@
 //! - [`DatabaseReadHandle`]
 //! - [`DatabaseWriteHandle`]
 //!
-//! The 1st allows any caller to send [`ReadRequest`]s.
+//! The 1st allows any caller to send [`ReadRequest`][req_r]s.
 //!
-//! The 2nd allows any caller to send [`WriteRequest`]s.
+//! The 2nd allows any caller to send [`WriteRequest`][req_w]s.
 //!
 //! The `DatabaseReadHandle` can be shared as it is cheaply [`Clone`]able, however,
 //! the `DatabaseWriteHandle` cannot be cloned. There is only 1 place in Cuprate that
@@ -49,6 +49,14 @@
 //! An `async`hronous channel will be returned from the call.
 //! This channel can be `.await`ed upon to (eventually) receive
 //! the corresponding `Response` to your `Request`.
+//!
+//!
+//!
+//! [req_r]: cuprate_types::service::ReadRequest
+//!
+//! [req_w]: cuprate_types::service::WriteRequest
+//!
+//! [resp]: cuprate_types::service::Response
 
 mod read;
 pub use read::DatabaseReadHandle;
@@ -59,11 +67,5 @@ pub use write::DatabaseWriteHandle;
 mod free;
 pub use free::init;
 
-mod request;
-pub use request::{ReadRequest, WriteRequest};
-
-mod response;
-pub use response::Response;
-
 #[cfg(test)]
 mod tests;

database/src/service/read.rs

@@ -2,6 +2,8 @@
 
 //---------------------------------------------------------------------------------------------------- Import
 use std::{
+    collections::{HashMap, HashSet},
+    ops::Range,
     sync::Arc,
     task::{Context, Poll},
 };
@@ -14,11 +16,12 @@ use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tokio_util::sync::PollSemaphore;
 
 use cuprate_helper::asynch::InfallibleOneshotReceiver;
+use cuprate_types::service::{ReadRequest, Response};
 
 use crate::{
     config::ReaderThreads,
     error::RuntimeError,
-    service::{request::ReadRequest, response::Response},
+    types::{Amount, AmountIndex, BlockHeight, KeyImage},
     ConcreteEnv,
 };
 
@@ -155,10 +158,8 @@ impl tower::Service<ReadRequest> for DatabaseReadHandle {
         }
 
         // Acquire a permit before returning `Ready`.
-        let Some(permit) = ready!(self.semaphore.poll_acquire(cx)) else {
-            // `self` itself owns the backing semaphore, so it can't be closed.
-            unreachable!();
-        };
+        let permit = ready!(self.semaphore.poll_acquire(cx))
+            .expect("`self` itself owns the backing semaphore, so it can't be closed.");
 
         self.permit = Some(permit);
         Poll::Ready(Ok(()))
@@ -199,30 +200,45 @@ impl tower::Service<ReadRequest> for DatabaseReadHandle {
 ///
 /// This is the main entrance into all `Request` handler functions.
 /// The basic structure is:
-///
 /// 1. `Request` is mapped to a handler function
 /// 2. Handler function is called
 /// 3. [`Response`] is sent
 fn map_request(
-    _permit: OwnedSemaphorePermit,   // Permit for this request
-    env: Arc<ConcreteEnv>,           // Access to the database
-    request: ReadRequest,            // The request we must fulfill
+    _permit: OwnedSemaphorePermit, // Permit for this request, dropped at end of function
+    env: Arc<ConcreteEnv>,         // Access to the database
+    request: ReadRequest,          // The request we must fulfill
     response_sender: ResponseSender, // The channel we must send the response back to
 ) {
     /* TODO: pre-request handling, run some code for each request? */
-
-    match request {
-        ReadRequest::Example1 => example_handler_1(env, response_sender),
-        ReadRequest::Example2(x) => example_handler_2(env, response_sender, x),
-        ReadRequest::Example3(x) => example_handler_3(env, response_sender, x),
-    }
+    use ReadRequest as R;
+
+    let response = match request {
+        R::BlockExtendedHeader(block) => block_extended_header(&env, block),
+        R::BlockHash(block) => block_hash(&env, block),
+        R::BlockExtendedHeaderInRange(range) => block_extended_header_in_range(&env, range),
+        R::ChainHeight => chain_height(&env),
+        R::GeneratedCoins => generated_coins(&env),
+        R::Outputs(map) => outputs(&env, map),
+        R::NumberOutputsWithAmount(vec) => number_outputs_with_amount(&env, vec),
+        R::CheckKIsNotSpent(set) => check_k_is_not_spent(&env, set),
+        R::BlockBatchInRange(range) => block_batch_in_range(&env, range),
+    };
+
+    // TODO: what do we do if this errors?
+    response_sender.send(response).unwrap();
 
     /* TODO: post-request handling, run some code for each request? */
 }
 
 //---------------------------------------------------------------------------------------------------- Handler functions
 // These are the actual functions that do stuff according to the incoming [`Request`].
 //
+// Each function name is a 1-1 mapping (from CamelCase -> snake_case) to
+// the enum variant name, e.g: `BlockExtendedHeader` -> `block_extended_header`.
+//
+// Each function will return the [`Response`] that we
+// should send back to the caller in [`map_request()`].
+//
 // INVARIANT:
 // These functions are called above in `tower::Service::call()`
 // using a custom threadpool which means any call to `par_*()` functions
@@ -231,26 +247,63 @@ fn map_request(
 // All functions below assume that this is the case, such that
 // `par_*()` functions will not block the _global_ rayon thread-pool.
 
-/// TODO
+/// [`ReadRequest::BlockExtendedHeader`].
+#[inline]
+fn block_extended_header(env: &Arc<ConcreteEnv>, block_height: BlockHeight) -> ResponseResult {
+    todo!()
+}
+
+/// [`ReadRequest::BlockHash`].
+#[inline]
+fn block_hash(env: &Arc<ConcreteEnv>, block_height: BlockHeight) -> ResponseResult {
+    todo!()
+}
+
+/// [`ReadRequest::BlockExtendedHeaderInRange`].
+#[inline]
+fn block_extended_header_in_range(
+    env: &Arc<ConcreteEnv>,
+    range: std::ops::Range<BlockHeight>,
+) -> ResponseResult {
+    todo!()
+}
+
+/// [`ReadRequest::ChainHeight`].
+#[inline]
+fn chain_height(env: &Arc<ConcreteEnv>) -> ResponseResult {
+    todo!()
+}
+
+/// [`ReadRequest::GeneratedCoins`].
+#[inline]
+fn generated_coins(env: &Arc<ConcreteEnv>) -> ResponseResult {
+    todo!()
+}
+
+/// [`ReadRequest::Outputs`].
 #[inline]
 #[allow(clippy::needless_pass_by_value)] // TODO: remove me
-fn example_handler_1(env: Arc<ConcreteEnv>, response_sender: ResponseSender) {
-    let db_result = Ok(Response::Example1);
-    response_sender.send(db_result).unwrap();
+fn outputs(env: &Arc<ConcreteEnv>, map: HashMap<Amount, HashSet<AmountIndex>>) -> ResponseResult {
+    todo!()
 }
 
+/// [`ReadRequest::NumberOutputsWithAmount`].
 /// TODO
 #[inline]
 #[allow(clippy::needless_pass_by_value)] // TODO: remove me
-fn example_handler_2(env: Arc<ConcreteEnv>, response_sender: ResponseSender, x: usize) {
-    let db_result = Ok(Response::Example2(x));
-    response_sender.send(db_result).unwrap();
+fn number_outputs_with_amount(env: &Arc<ConcreteEnv>, vec: Vec<Amount>) -> ResponseResult {
+    todo!()
 }
 
-/// TODO
+/// [`ReadRequest::CheckKIsNotSpent`].
 #[inline]
 #[allow(clippy::needless_pass_by_value)] // TODO: remove me
-fn example_handler_3(env: Arc<ConcreteEnv>, response_sender: ResponseSender, x: String) {
-    let db_result = Ok(Response::Example3(x));
-    response_sender.send(db_result).unwrap();
+fn check_k_is_not_spent(env: &Arc<ConcreteEnv>, set: HashSet<KeyImage>) -> ResponseResult {
+    todo!()
+}
+
+/// [`ReadRequest::BlockBatchInRange`].
+#[inline]
+fn block_batch_in_range(env: &Arc<ConcreteEnv>, range: Range<BlockHeight>) -> ResponseResult {
+    todo!()
 }