better support of null storage (profile) engine

Author: shortishly

Cargo.toml

@@ -159,6 +159,10 @@ url = { version = "2.5.4", features = ["serde"] }
 uuid = { version = "1.13.2", features = ["serde", "v4", "v7"] }
 zstd = "0.13.2"
 
-[profile.release]
-codegen-units = 1
-lto = "thin"
+# [profile.release]
+# codegen-units = 1
+# lto = "thin"
+
+[profile.profiling]
+inherits = "release"
+debug = true

justfile

@@ -17,6 +17,8 @@ build-examples: (cargo-build "--examples")
 
 release: (cargo-build "--release" "--bin" "tansu" "--no-default-features" "--features" "delta,dynostore,iceberg,libsql,parquet,postgres")
 
+release-sqlite: (cargo-build "--release" "--bin" "tansu" "--no-default-features" "--features" "libsql")
+
 test: test-workspace test-doc
 
 test-workspace:
@@ -381,3 +383,7 @@ customer-topic-create *args: (topic-create "customer" "--partitions=1"  "--confi
 customer-topic-generator *args: (generator "customer" args)
 
 customer-duckdb-delta: (duckdb "\"select * from delta_scan('s3://lake/tansu.customer');\"")
+
+profile-null:
+    cargo build --profile profiling
+    RUST_LOG=warn samply record ./target/profiling/tansu --storage-engine=null://sink

tansu-service/src/api.rs

@@ -19,7 +19,6 @@ use tansu_sans_io::{
     ApiKey, ApiVersionsRequest, ApiVersionsResponse, Body, ErrorCode, Frame, Header,
     RootMessageMeta, api_versions_response::ApiVersion,
 };
-use tracing::debug;
 
 use crate::Error;
 
@@ -167,8 +166,6 @@ where
     type Error = E;
 
     async fn serve(&self, ctx: Context<State>, req: Frame) -> Result<Self::Response, Self::Error> {
-        debug!(?req);
-
         let api_key = req.api_key()?;
 
         if let Some(service) = self.routes.get(&api_key) {

tansu-service/src/frame.rs

@@ -159,7 +159,6 @@ where
     type Error = S::Error;
 
     async fn serve(&self, ctx: Context<State>, req: Frame) -> Result<Self::Response, Self::Error> {
-        debug!(?req);
         let correlation_id = req.correlation_id()?;
 
         let req = Q::try_from(req.body).map_err(Into::into)?;
@@ -216,7 +215,7 @@ where
     type Error = S::Error;
 
     async fn serve(&self, ctx: Context<State>, req: Bytes) -> Result<Self::Response, Self::Error> {
-        let req = Frame::request_from_bytes(req).inspect(|req| debug!(?req))?;
+        let req = Frame::request_from_bytes(req)?;
         let api_key = req.api_key()?;
         let api_version = req.api_version()?;
         let correlation_id = req.correlation_id()?;

tansu-service/src/stream.rs

@@ -235,6 +235,14 @@ pub struct TcpBytesService<S, State> {
     _state: PhantomData<State>,
 }
 
+impl<S, State> TcpBytesService<S, State> {
+    fn elapsed_millis(&self, start: SystemTime) -> u64 {
+        start
+            .elapsed()
+            .map_or(0, |duration| duration.as_millis() as u64)
+    }
+}
+
 impl<S, State> Service<TcpContext, TcpStream> for TcpBytesService<S, State>
 where
     S: Service<State, Bytes, Response = Bytes>,
@@ -298,7 +306,6 @@ where
                     .read_exact(&mut request[4..])
                     .await
                     .inspect_err(|err| error!(?err))?;
-                debug!(?request);
 
                 REQUEST_SIZE.record(request.len() as u64, &attributes);
 
@@ -313,15 +320,11 @@ where
                     .inspect(|response| {
                         RESPONSE_SIZE.record(response.len() as u64, &attributes);
 
-                        REQUEST_DURATION.record(
-                            request_start
-                                .elapsed()
-                                .map_or(0, |duration| duration.as_millis() as u64),
-                            &attributes,
-                        );
-                    })?;
+                        let elapsed_millis = self.elapsed_millis(request_start);
+                        debug!(elapsed_millis);
 
-                debug!(response = ?&response[..]);
+                        REQUEST_DURATION.record(elapsed_millis, &attributes);
+                    })?;
 
                 req.write_all(&response)
                     .await

tansu-storage/src/lib.rs

@@ -1700,13 +1700,23 @@ impl Builder<i32, String, Url, Url> {
                 message: self.storage.to_string(),
             }),
 
+            "null" => Ok(StorageContainer::Null(null::Engine::new(
+                self.cluster_id.clone(),
+                self.node_id,
+                self.advertised_listener.clone(),
+            ))),
+
             #[cfg(not(any(
                 feature = "dynostore",
                 feature = "libsql",
                 feature = "postgres",
                 feature = "turso"
             )))]
-            _storage => Ok(StorageContainer::Null(null::Engine)),
+            _storage => Ok(StorageContainer::Null(null::Engine::new(
+                self.cluster_id.clone(),
+                self.node_id,
+                self.advertised_listener.clone(),
+            ))),
 
             #[cfg(any(
                 feature = "dynostore",
@@ -1923,7 +1933,7 @@ impl Storage for StorageContainer {
         })
     }
 
-    #[instrument(skip(self), ret)]
+    #[instrument(skip_all, ret)]
     async fn produce(
         &self,
         transaction_id: Option<&str>,

tansu-storage/src/lite.rs

@@ -221,7 +221,7 @@ impl managed::Manager for ConnectionManager {
     type Type = Connection;
     type Error = Error;
 
-    #[instrument(ret)]
+    #[instrument(skip_all, ret)]
     async fn create(&self) -> Result<Self::Type, Self::Error> {
         let start = SystemTime::now();
 
@@ -239,6 +239,11 @@ impl managed::Manager for ConnectionManager {
         }
 
         {
+            _ = connection
+                .execute("PRAGMA synchronous = normal", ())
+                .await
+                .inspect(|rows| debug!(rows))?;
+
             let mut rows = connection.query("PRAGMA synchronous", ()).await?;
 
             if let Some(row) = rows.next().await.inspect_err(|err| error!(?err))? {
@@ -272,13 +277,11 @@ impl managed::Manager for ConnectionManager {
             .inspect(|_| CONNECT_DURATION.record(elapsed_millis(start), &[]))
     }
 
-    #[instrument(ret)]
     async fn recycle(
         &self,
-        obj: &mut Self::Type,
-        metrics: &managed::Metrics,
+        _obj: &mut Self::Type,
+        _metrics: &managed::Metrics,
     ) -> managed::RecycleResult<Self::Error> {
-        debug!(?obj, ?metrics);
         Ok(())
     }
 }
@@ -543,13 +546,11 @@ impl Delegate {
 
     async fn idempotent_message_check(
         &self,
-        transaction_id: Option<&str>,
+        _transaction_id: Option<&str>,
         topition: &Topition,
         deflated: &deflated::Batch,
         connection: &Connection,
     ) -> Result<()> {
-        debug!(transaction_id, ?deflated);
-
         let mut rows = connection
             .query(
                 &sql_lookup("producer_epoch_current_for_producer.sql")?,
@@ -663,8 +664,6 @@ impl Delegate {
         deflated: deflated::Batch,
         tx: &Transaction,
     ) -> Result<i64> {
-        debug!(cluster = ?self.cluster, ?transaction_id, ?topition, ?deflated);
-
         let start = SystemTime::now();
 
         let topic = topition.topic();
@@ -731,7 +730,7 @@ impl Delegate {
             let key = record.key.as_deref();
             let value = record.value.as_deref();
 
-            debug!(?delta, ?record, ?offset);
+            debug!(?delta, ?offset);
 
             _ = self
                 .prepare_execute(
@@ -1349,7 +1348,7 @@ impl Storage for Engine {
             })
     }
 
-    #[instrument(ret)]
+    #[instrument(skip_all, ret)]
     async fn produce(
         &self,
         transaction_id: Option<&str>,
@@ -2118,8 +2117,6 @@ impl Storage for Delegate {
     ) -> Result<i64> {
         let start = SystemTime::now();
 
-        debug!(cluster = self.cluster, transaction_id, ?topition, ?deflated);
-
         let tx = self.transaction().await.inspect(|_| {
             debug!(after_produce_transaction = elapsed_millis(start));
         })?;