final 0.5.10 fixes

Author: shortishly

justfile

@@ -387,26 +387,18 @@ customer-topic-generator *args: (generator "customer" args)
 
 customer-duckdb-delta: (duckdb "\"select * from delta_scan('s3://lake/tansu.customer');\"")
 
-broker-memory profile="profiling":
-    cargo build --profile {{ profile }} --bin tansu
-    ./target/{{ replace(profile, "dev", "debug") }}/tansu broker --storage-engine=memory:// 2>&1 | tee broker.log
+broker-memory profile="profiling": (build profile "dynostore") (tansu-broker profile "--storage-engine=memory://")
 
-broker-null profile="profiling":
-    cargo build --profile {{ profile }} --bin tansu
-    ./target/{{ replace(profile, "dev", "debug") }}/tansu --storage-engine=null://sink 2>&1 | tee broker.log
+broker-null profile="profiling": (build profile "default") (tansu-broker profile "--storage-engine=null://")
 
 clean-tansu-db:
     rm -f tansu.db*
 
-broker-sqlite profile="profiling": clean-tansu-db
-    cargo build --profile {{ profile }} --features libsql --bin tansu
-    ./target/{{ replace(profile, "dev", "debug") }}/tansu broker --storage-engine=sqlite://tansu.db 2>&1 | tee broker.log
+broker-sqlite profile="profiling": clean-tansu-db (build profile "libsql") (tansu-broker profile "--storage-engine=sqlite://tansu.db")
 
 broker-s3 profile="profiling": (build profile "dynostore") docker-compose-down minio-up minio-ready-local minio-local-alias minio-tansu-bucket (tansu-broker profile "--storage-engine=s3://tansu/")
 
-broker-postgres profile="profiling":
-    cargo build --profile {{ profile }} --features postgres --bin tansu
-    ./target/{{ replace(profile, "dev", "debug") }}/tansu broker --storage-engine=postgres://pmorgan@localhost 2>&1 | tee broker.log
+broker-postgres profile="profiling": (build profile "postgres") docker-compose-down db-up (tansu-broker profile "--storage-engine=postgres://pmorgan@localhost")
 
 samply-null profile="profiling":
     cargo build --profile {{ profile }} --bin tansu

tansu-sans-io/src/record/deflated.rs

@@ -1,4 +1,4 @@
-// Copyright ⓒ 2024-2025 Peter Morgan <peter.james.morgan@gmail.com>
+// Copyright ⓒ 2024-2026 Peter Morgan <peter.james.morgan@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -177,6 +177,71 @@ impl TryFrom<Vec<u8>> for Batch {
     }
 }
 
+impl TryFrom<Bytes> for Batch {
+    type Error = Error;
+    fn try_from(mut encoded: Bytes) -> result::Result<Self, Self::Error> {
+        let base_offset = encoded.try_get_i64()?;
+        let batch_length = encoded.try_get_i32()?;
+
+        let partition_leader_epoch = encoded.try_get_i32()?;
+        let magic = encoded.try_get_i8()?;
+        let crc = encoded.try_get_u32()?;
+
+        let crc_data_size = usize::try_from(batch_length).map(|batch_length| {
+            batch_length
+                - size_of_val(&partition_leader_epoch)
+                - size_of_val(&magic)
+                - size_of_val(&crc)
+        })?;
+
+        let crc_data = &encoded[..crc_data_size];
+
+        let computed = {
+            let mut digest = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32Iscsi);
+            digest.update(crc_data);
+
+            digest.finalize() as u32
+        };
+
+        if computed != crc {
+            error!(crc, computed);
+        }
+
+        let attributes = encoded.try_get_i16()?;
+        let last_offset_delta = encoded.try_get_i32()?;
+        let base_timestamp = encoded.try_get_i64()?;
+        let max_timestamp = encoded.try_get_i64()?;
+        let producer_id = encoded.try_get_i64()?;
+        let producer_epoch = encoded.try_get_i16()?;
+        let base_sequence = encoded.try_get_i32()?;
+        let record_count = encoded.try_get_u32()?;
+
+        let record_data_size =
+            usize::try_from(batch_length).map(|batch_length| batch_length - FIXED_BATCH_LENGTH)?;
+
+        let record_data = encoded.slice(..record_data_size);
+
+        let batch = Batch {
+            base_offset,
+            batch_length,
+            partition_leader_epoch,
+            magic,
+            crc,
+            attributes,
+            last_offset_delta,
+            base_timestamp,
+            max_timestamp,
+            producer_id,
+            producer_epoch,
+            base_sequence,
+            record_count,
+            record_data,
+        };
+
+        Ok(batch)
+    }
+}
+
 impl TryFrom<&[u8]> for Batch {
     type Error = Error;
 

tansu-storage/src/dynostore.rs

@@ -1,4 +1,4 @@
-// Copyright ⓒ 2024-2025 Peter Morgan <peter.james.morgan@gmail.com>
+// Copyright ⓒ 2024-2026 Peter Morgan <peter.james.morgan@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -48,7 +48,6 @@ use tansu_sans_io::{
         AddPartitionsToTxnPartitionResult, AddPartitionsToTxnTopicResult,
     },
     create_topics_request::{CreatableTopic, CreatableTopicConfig},
-    de::BatchDecoder,
     delete_groups_response::DeletableGroupResult,
     delete_records_request::DeleteRecordsTopic,
     delete_records_response::DeleteRecordsTopicResult,
@@ -402,8 +401,7 @@ impl DynoStore {
     }
 
     fn decode(&self, encoded: Bytes) -> Result<deflated::Batch> {
-        let decoder = BatchDecoder::new(encoded);
-        deflated::Batch::deserialize(decoder).map_err(Into::into)
+        deflated::Batch::try_from(encoded).map_err(Into::into)
     }
 
     async fn get<V>(&self, location: &Path) -> Result<(V, Version)>
@@ -555,6 +553,7 @@ impl Storage for DynoStore {
         }
     }
 
+    #[instrument(skip_all, fields(topic = %topic.name))]
     async fn create_topic(&self, topic: CreatableTopic, _validate_only: bool) -> Result<Uuid> {
         match self
             .meta
@@ -564,6 +563,8 @@ impl Storage for DynoStore {
                 }
 
                 let id = Uuid::now_v7();
+                debug!(%id);
+
                 let td = TopicMetadata {
                     id,
                     topic: topic.clone(),

tansu-storage/src/dynostore/metadata.rs

@@ -1,4 +1,4 @@
-// Copyright ⓒ 2024-2025 Peter Morgan <peter.james.morgan@gmail.com>
+// Copyright ⓒ 2024-2026 Peter Morgan <peter.james.morgan@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -17,7 +17,7 @@ use std::{
     ops::DerefMut,
     slice::from_ref,
     sync::{Arc, LazyLock, Mutex},
-    time::{Duration, SystemTime},
+    time::Duration,
 };
 
 use async_trait::async_trait;
@@ -37,7 +37,17 @@ use super::METER;
 #[derive(Clone, Debug, Eq, PartialEq)]
 struct CacheEntry {
     version: UpdateVersion,
-    tagged_at: SystemTime,
+}
+
+impl From<GetOptions> for CacheEntry {
+    fn from(value: GetOptions) -> Self {
+        Self {
+            version: UpdateVersion {
+                e_tag: value.if_none_match,
+                version: None,
+            },
+        }
+    }
 }
 
 impl From<&PutResult> for CacheEntry {
@@ -49,7 +59,6 @@ impl From<&PutResult> for CacheEntry {
 
         Self {
             version: UpdateVersion { e_tag, version },
-            tagged_at: SystemTime::now(),
         }
     }
 }
@@ -63,7 +72,6 @@ impl From<&GetResult> for CacheEntry {
 
         Self {
             version: UpdateVersion { e_tag, version },
-            tagged_at: SystemTime::now(),
         }
     }
 }
@@ -142,14 +150,13 @@ impl<O> ObjectStore for Cache<O>
 where
     O: ObjectStore,
 {
+    #[instrument(skip_all, fields(location = %location))]
     async fn put_opts(
         &self,
         location: &Path,
         payload: PutPayload,
         opts: PutOptions,
     ) -> Result<PutResult, object_store::Error> {
-        debug!(%location, ?opts);
-
         let method = KeyValue::new("method", "put_opts");
 
         REQUESTS.add(1, from_ref(&method));
@@ -174,6 +181,7 @@ where
             })
     }
 
+    #[instrument(skip_all, fields(location = %location))]
     async fn put_multipart_opts(
         &self,
         location: &Path,
@@ -201,7 +209,7 @@ where
             })
     }
 
-    #[instrument(skip_all, fields(%location), ret)]
+    #[instrument(skip_all, fields(%location, if_none_match = options.if_none_match), ret)]
     async fn get_opts(
         &self,
         location: &Path,
@@ -260,14 +268,14 @@ where
         }
 
         self.object_store
-            .get_opts(location, options)
+            .get_opts(location, options.clone())
             .await
             .inspect(|get_result| {
                 let e_tag = get_result.meta.e_tag.clone();
                 let version = get_result.meta.version.clone();
 
                 if let Ok(mut guard) = self.entries.lock() {
-                    debug!(%location, e_tag, version);
+                    debug!(e_tag, version);
 
                     let replacement = CacheEntry::from(get_result);
 
@@ -284,22 +292,48 @@ where
                         Some(_) => "replace",
                     };
 
+                    debug!(outcome);
+
                     ENTRIES.add(1, &[method.clone(), KeyValue::new("outcome", outcome)]);
                 }
             })
             .inspect_err(|error| {
                 debug!(%location, ?error);
 
-                ERRORS.add(
-                    1,
-                    &[
-                        method,
-                        KeyValue::new("error", object_store_error_name(error)),
-                    ],
-                );
+                if matches!(error, object_store::Error::NotModified { .. }) {
+                    if let Ok(mut guard) = self.entries.lock() {
+                        let replacement = CacheEntry::from(options);
+
+                        let outcome = match guard
+                            .deref_mut()
+                            .insert_evict(location.to_owned(), replacement.clone(), true)
+                            .ok()
+                            .flatten()
+                        {
+                            None => "add",
+
+                            Some(existing) if existing == replacement => "existing",
+
+                            Some(_) => "replace",
+                        };
+
+                        debug!(outcome);
+
+                        ENTRIES.add(1, &[method.clone(), KeyValue::new("outcome", outcome)]);
+                    }
+                } else {
+                    ERRORS.add(
+                        1,
+                        &[
+                            method,
+                            KeyValue::new("error", object_store_error_name(error)),
+                        ],
+                    );
+                }
             })
     }
 
+    #[instrument(skip_all, fields(location = %location))]
     async fn delete(&self, location: &Path) -> Result<(), object_store::Error> {
         debug!(%location);
 
@@ -328,6 +362,7 @@ where
             })
     }
 
+    #[instrument(skip_all, fields(prefix))]
     fn list(
         &self,
         prefix: Option<&Path>,
@@ -337,6 +372,7 @@ where
         self.object_store.list(prefix)
     }
 
+    #[instrument(skip_all, fields(prefix))]
     async fn list_with_delimiter(
         &self,
         prefix: Option<&Path>,
@@ -359,6 +395,7 @@ where
             })
     }
 
+    #[instrument(skip_all, fields(from = %from, to = %to))]
     async fn copy(&self, from: &Path, to: &Path) -> Result<(), object_store::Error> {
         debug!(%from, %to);
         REQUESTS.add(1, &[KeyValue::new("method", "copy")]);
@@ -375,6 +412,7 @@ where
         })
     }
 
+    #[instrument(skip_all, fields(from = %from, to = %to))]
     async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<(), object_store::Error> {
         debug!(%from, %to);
         REQUESTS.add(1, &[KeyValue::new("method", "copy_if_not_exists")]);