Implement new server storage

BREAKING_CHANGES for API and SDK but comes with a built-in data migration for existing Sled DB

bench/src/benchmarks/benchmark.rs

@@ -14,6 +14,7 @@ use iggy::clients::client::{IggyClient, IggyClientBackgroundConfig};
 use iggy::compression::compression_algorithm::CompressionAlgorithm;
 use iggy::error::IggyError;
 use iggy::utils::expiry::IggyExpiry;
+use iggy::utils::topic_size::MaxTopicSize;
 use integration::test_server::{login_root, ClientFactory};
 use std::{pin::Pin, sync::Arc};
 use tracing::info;
@@ -90,7 +91,7 @@ pub trait Benchmarkable {
                         None,
                         None,
                         IggyExpiry::NeverExpire,
-                        None,
+                        MaxTopicSize::ServerDefault,
                     )
                     .await?;
             }

cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "cli"
-version = "0.20.1"
+version = "0.3.0"
 edition = "2021"
 authors = ["bartosz.ciesla@gmail.com"]
 repository = "https://github.com/iggy-rs/iggy"

cli/src/args/topic.rs

@@ -2,8 +2,8 @@ use crate::args::common::ListMode;
 use clap::{Args, Subcommand};
 use iggy::compression::compression_algorithm::CompressionAlgorithm;
 use iggy::identifier::Identifier;
-use iggy::utils::byte_size::IggyByteSize;
 use iggy::utils::expiry::IggyExpiry;
+use iggy::utils::topic_size::MaxTopicSize;
 
 #[derive(Debug, Clone, Subcommand)]
 pub(crate) enum TopicAction {
@@ -102,7 +102,7 @@ pub(crate) struct TopicCreateArgs {
     /// ("unlimited" or skipping parameter disables max topic size functionality in topic)
     /// Can't be lower than segment size in the config.
     #[arg(short, long, default_value = "unlimited", verbatim_doc_comment)]
-    pub(crate) max_topic_size: IggyByteSize,
+    pub(crate) max_topic_size: MaxTopicSize,
     /// Replication factor for the topic
     #[arg(short, long, default_value = "1")]
     pub(crate) replication_factor: u8,
@@ -149,7 +149,7 @@ pub(crate) struct TopicUpdateArgs {
     /// ("unlimited" or skipping parameter causes removal of max topic size parameter in topic)
     /// Can't be lower than segment size in the config.
     #[arg(short, long, default_value = "unlimited", verbatim_doc_comment)]
-    pub(crate) max_topic_size: IggyByteSize,
+    pub(crate) max_topic_size: MaxTopicSize,
     #[arg(short, long, default_value = "1")]
     /// New replication factor for the topic
     pub(crate) replication_factor: u8,

configs/server.json

@@ -34,7 +34,6 @@
         "iggy.rs"
       ],
       "access_token_expiry": "1 h",
-      "refresh_token_expiry": "1 d",
       "clock_skew": "5 s",
       "not_before": "0 s",
       "encoding_secret": "top_secret$iggy.rs$_jwt_HS256_key#!",
@@ -94,9 +93,6 @@
   },
   "system": {
     "path": "local_data",
-    "database": {
-      "path": "database"
-    },
     "backup": {
       "path": "backup",
       "compatibility": {
@@ -118,7 +114,7 @@
       "size": "4 GB"
     },
     "retention_policy": {
-      "message_expiry": "disabled",
+      "message_expiry": "none",
       "max_topic_size": "10 GB"
     },
     "encryption": {

configs/server.toml

@@ -66,9 +66,6 @@ valid_audiences = ["iggy.rs"]
 # Expiry time for access tokens.
 access_token_expiry = "1 h"
 
-# Expiry time for refresh tokens.
-refresh_token_expiry = "1 d"
-
 # Tolerance for timing discrepancies during token validation.
 clock_skew = "5 s"
 
@@ -233,11 +230,11 @@ path = "backup"
 # Subpath of the backup directory where converted segment data is stored after compatibility conversion.
 path = "compatibility"
 
-# Database configuration.
-[system.database]
-# Path for storing database files.
-# Specifies the directory where database files are stored, relative to `system.path`.
-path = "database"
+# Legacy database configuration - used only for the migration purposes.
+#[system.database]
+## Path for storing database files.
+## Specifies the directory where database files are stored, relative to `system.path`.
+#path = "database"
 
 # Runtime configuration.
 [system.runtime]
@@ -275,10 +272,10 @@ size = "4 GB"
 # Data retention policy configuration.
 [system.retention_policy]
 # Configures the message time-based expiry setting.
-# "disabled" means messages are kept indefinitely.
+# "none" means messages are kept indefinitely.
 # A time value in human-readable format determines the lifespan of messages.
 # Example: `message_expiry = "2 days 4 hours 15 minutes"` means messages will expire after that duration.
-message_expiry = "disabled"
+message_expiry = "none"
 
 # Configures the topic size-based expiry setting.
 # "unlimited" or "0" means topics are kept indefinitely.

examples/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "iggy_examples"
-version = "0.0.3"
+version = "0.0.4"
 edition = "2021"
 
 [[example]]

examples/src/getting-started/producer/main.rs

@@ -5,6 +5,7 @@ use iggy::compression::compression_algorithm::CompressionAlgorithm;
 use iggy::messages::send_messages::{Message, Partitioning};
 use iggy::users::defaults::*;
 use iggy::utils::expiry::IggyExpiry;
+use iggy::utils::topic_size::MaxTopicSize;
 use std::env;
 use std::error::Error;
 use std::str::FromStr;
@@ -53,7 +54,7 @@ async fn init_system(client: &IggyClient) {
             None,
             Some(TOPIC_ID),
             IggyExpiry::NeverExpire,
-            None,
+            MaxTopicSize::ServerDefault,
         )
         .await
     {

examples/src/shared/messages.rs

@@ -1,3 +1,4 @@
+use iggy::utils::timestamp::IggyTimestamp;
 use serde::{Deserialize, Serialize};
 use std::fmt::{self, Debug};
 
@@ -41,7 +42,7 @@ pub struct OrderCreated {
     pub price: f64,
     pub quantity: f64,
     pub side: String,
-    pub timestamp: u64,
+    pub timestamp: IggyTimestamp,
 }
 
 impl Debug for OrderCreated {
@@ -52,7 +53,7 @@ impl Debug for OrderCreated {
             .field("price", &format!("{:.2}", self.price))
             .field("quantity", &format!("{:.2}", self.quantity))
             .field("side", &self.side)
-            .field("timestamp", &self.timestamp)
+            .field("timestamp", &self.timestamp.to_micros())
             .finish()
     }
 }
@@ -61,26 +62,36 @@ impl Debug for OrderCreated {
 pub struct OrderConfirmed {
     pub order_id: u64,
     pub price: f64,
-    pub timestamp: u64,
+    pub timestamp: IggyTimestamp,
 }
 
 impl Debug for OrderConfirmed {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("OrderConfirmed")
             .field("order_id", &self.order_id)
             .field("price", &format!("{:.2}", self.price))
-            .field("timestamp", &self.timestamp)
+            .field("timestamp", &self.timestamp.to_micros())
             .finish()
     }
 }
 
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Deserialize, Serialize)]
 pub struct OrderRejected {
     pub order_id: u64,
-    pub timestamp: u64,
+    pub timestamp: IggyTimestamp,
     pub reason: String,
 }
 
+impl Debug for OrderRejected {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("OrderRejected")
+            .field("order_id", &self.order_id)
+            .field("timestamp", &self.timestamp.to_micros())
+            .field("reason", &self.reason)
+            .finish()
+    }
+}
+
 impl SerializableMessage for OrderCreated {
     fn get_message_type(&self) -> &str {
         ORDER_CREATED_TYPE

examples/src/shared/messages_generator.rs

@@ -1,5 +1,5 @@
 use crate::shared::messages::{OrderConfirmed, OrderCreated, OrderRejected, SerializableMessage};
-use crate::shared::utils;
+use iggy::utils::timestamp::IggyTimestamp;
 use rand::rngs::ThreadRng;
 use rand::Rng;
 
@@ -32,7 +32,7 @@ impl MessagesGenerator {
         self.order_id += 1;
         Box::new(OrderCreated {
             order_id: self.order_id,
-            timestamp: utils::timestamp(),
+            timestamp: IggyTimestamp::now(),
             currency_pair: CURRENCY_PAIRS[self.rng.gen_range(0..CURRENCY_PAIRS.len())].to_string(),
             price: self.rng.gen_range(10.0..=1000.0),
             quantity: self.rng.gen_range(0.1..=1.0),
@@ -47,15 +47,15 @@ impl MessagesGenerator {
     fn generate_order_confirmed(&mut self) -> Box<dyn SerializableMessage> {
         Box::new(OrderConfirmed {
             order_id: self.order_id,
-            timestamp: utils::timestamp(),
+            timestamp: IggyTimestamp::now(),
             price: self.rng.gen_range(10.0..=1000.0),
         })
     }
 
     fn generate_order_rejected(&mut self) -> Box<dyn SerializableMessage> {
         Box::new(OrderRejected {
             order_id: self.order_id,
-            timestamp: utils::timestamp(),
+            timestamp: IggyTimestamp::now(),
             reason: match self.rng.gen_range(0..=1) {
                 0 => "cancelled_by_user",
                 _ => "other",

examples/src/shared/mod.rs

@@ -2,4 +2,3 @@ pub mod args;
 pub mod messages;
 pub mod messages_generator;
 pub mod system;
-pub mod utils;

examples/src/shared/system.rs

@@ -8,6 +8,7 @@ use iggy::messages::poll_messages::PollingStrategy;
 use iggy::models::messages::PolledMessage;
 use iggy::users::defaults::*;
 use iggy::utils::expiry::IggyExpiry;
+use iggy::utils::topic_size::MaxTopicSize;
 use tracing::info;
 
 type MessageHandler = dyn Fn(&PolledMessage) -> Result<(), Box<dyn std::error::Error>>;
@@ -74,7 +75,7 @@ pub async fn init_by_producer(args: &Args, client: &dyn Client) -> Result<(), Ig
             None,
             Some(args.topic_id),
             IggyExpiry::NeverExpire,
-            None,
+            MaxTopicSize::ServerDefault,
         )
         .await?;
     Ok(())

integration/Cargo.toml

@@ -17,7 +17,6 @@ predicates = "3.1.0"
 regex = "1.10.4"
 serial_test = "3.1.1"
 server = { path = "../server" }
-sled = "0.34.7"
 tempfile = "3.10.1"
 tokio = { version = "1.38.0", features = ["full"] }
 tracing-subscriber = "0.3.18"

integration/tests/cli/consumer_group/test_consumer_group_create_command.rs

@@ -6,6 +6,7 @@ use assert_cmd::assert::Assert;
 use async_trait::async_trait;
 use iggy::client::Client;
 use iggy::utils::expiry::IggyExpiry;
+use iggy::utils::topic_size::MaxTopicSize;
 use predicates::str::diff;
 use serial_test::parallel;
 
@@ -83,7 +84,7 @@ impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
                 None,
                 Some(self.topic_id),
                 IggyExpiry::NeverExpire,
-                None,
+                MaxTopicSize::ServerDefault,
             )
             .await;
         assert!(topic.is_ok());