fix compose.yaml image link

Cargo.lock

@@ -83,8 +83,10 @@ dependencies = [
  "bon",
  "bytes",
  "byteyarn",
+ "clap",
  "env_logger",
  "log",
+ "shell-words",
  "thiserror",
  "tokio",
 ]
@@ -177,6 +179,46 @@ version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9555578bc9e57714c812a1f84e4fc5b4d21fcb063490c624de019f7464c91268"
 
+[[package]]
+name = "clap"
+version = "4.5.40"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "40b6887a1d8685cebccf115538db5c0efe625ccac9696ad45c409d96566e910f"
+dependencies = [
+ "clap_builder",
+ "clap_derive",
+]
+
+[[package]]
+name = "clap_builder"
+version = "4.5.40"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e0c66c08ce9f0c698cbce5c0279d0bb6ac936d8674174fe48f736533b964f59e"
+dependencies = [
+ "anstream",
+ "anstyle",
+ "clap_lex",
+ "strsim",
+]
+
+[[package]]
+name = "clap_derive"
+version = "4.5.40"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2c7947ae4cc3d851207c1adb5b5e260ff0cca11446b1d6d1423788e442257ce"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "clap_lex"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b94f61472cee1439c0b966b47e3aca9ae07e45d070759512cd390ea2bebc6675"
+
 [[package]]
 name = "colorchoice"
 version = "1.0.4"
@@ -253,6 +295,12 @@ version = "0.31.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f"
 
+[[package]]
+name = "heck"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
+
 [[package]]
 name = "ident_case"
 version = "1.0.1"
@@ -500,6 +548,12 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "shell-words"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "24188a676b6ae68c3b2cb3a01be17fbf7240ce009799bb56d5b1409051e78fde"
+
 [[package]]
 name = "signal-hook-registry"
 version = "1.4.5"

Cargo.toml

@@ -1,13 +1,25 @@
 [package]
 name = "archive"
-version = "0.1.0"
+version = "0.1.1"
 edition = "2024"
+authors = ["409"]
+default-run = "archive-server"
 
 [dependencies]
 bon = "3.6.4"
 bytes = "1.10.1"
 byteyarn = "0.5.1"
+clap = { version = "4.5.40", features = ["derive"] }
 env_logger = "0.11.8"
 log = "0.4.27"
+shell-words = "1.1.0"
 thiserror = "2.0.12"
 tokio = { version = "1.45.1", features = ["full"] }
+
+[[bin]]
+name = "archive-server"
+path = "src/bin/server.rs"
+
+[[bin]]
+name = "archive-cli"
+path = "src/bin/cli.rs"

Dockerfile

@@ -0,0 +1,17 @@
+FROM rust:bullseye AS builder
+WORKDIR /usr/src/archive
+
+COPY Cargo.toml .
+RUN mkdir -p src/bin && echo "fn main() {}" > src/bin/server.rs
+RUN cargo build --release --bin archive-server
+
+COPY src src
+RUN touch src/main.rs
+RUN cargo build --release --bin archive-server
+
+RUN strip target/release/archive-server
+
+FROM debian:12.11
+RUN apt-get update
+COPY --from=builder /usr/src/archive/target/release/archive-server /usr/local/bin/archive-server
+ENTRYPOINT ["archive-server"]

README.md

@@ -0,0 +1,34 @@
+![Logo](logo.webp)
+
+# ArcHIVE
+
+A lightweight in-memory database written in Rust
+
+## Usage
+
+### CLI
+
+Download a binary or compile it yourself using `cargo build --release --bin archive-cli`
+
+To compile the CLI and install it globally run `cargo install --path .`
+
+### Server
+
+#### Docker
+
+The repository contains a simple `compose.yaml`
+
+#### Binary
+
+Alternatively you can run the server binary directly
+
+Download a binary or compile it yourself using `cargo build --release --bin archive-server`
+
+#### Environment variables
+
+This is a list of the server's environment variables and their default values:
+
+- SERVER_HOST=0.0.0.0
+- SERVER_PORT=6171
+- MAX_CONNECTIONS=256
+- LOG_LEVEL=info

compose.yaml

@@ -0,0 +1,12 @@
+services:
+  archive-server:
+    container_name: 'archive-server'
+    image: 'git.409dev.online/409/archive:latest'
+    build: '.'
+    ports:
+      - '6171:6171/tcp'
+    environment:
+      - 'SERVER_HOST=0.0.0.0'
+      - 'SERVER_PORT=6171'
+      - 'MAX_CONNECTIONS=256'
+      - 'LOG_LEVEL=info'

src/bin/cli.rs

@@ -0,0 +1,137 @@
+use std::io::{Write as _, stdin};
+
+use archive::{Result, client::Client};
+use clap::Parser;
+
+#[derive(Debug, Parser)]
+#[command(version, about, long_about = None)]
+struct Args {
+    #[arg(long, default_value = "127.0.0.1")]
+    host: String,
+    #[arg(short, long, default_value_t = 6171)]
+    port: u16,
+}
+
+#[derive(Debug, Parser)]
+#[command(help_template = "{subcommands}")]
+enum Commands {
+    Get {
+        key: String,
+    },
+    Set {
+        key: String,
+        value: String,
+        expiration: Option<u64>,
+    },
+    Delete {
+        key: String,
+    },
+    Has {
+        key: String,
+    },
+    Ttl {
+        key: String,
+    },
+    Expire {
+        key: String,
+        seconds: u64,
+    },
+    Persist {
+        key: String,
+    },
+
+    #[command(name = "mget")]
+    MGet {
+        #[arg(num_args = 1..)]
+        keys: Vec<String>,
+    },
+
+    #[command(aliases = &["exit", "q"])]
+    Quit,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let args = Args::parse();
+
+    let mut client = Client::new(&format!("{}:{}", args.host, args.port)).await?;
+
+    print_welcome();
+
+    let stdin = stdin();
+    let mut input = String::new();
+
+    loop {
+        input.clear();
+        print!("> ");
+        std::io::stdout().flush()?;
+        stdin.read_line(&mut input)?;
+
+        let Ok(mut words) = shell_words::split(&input) else {
+            continue;
+        };
+
+        words.insert(0, "".into());
+
+        let command = match Commands::try_parse_from(words.clone()) {
+            Ok(command) => command,
+            Err(e) => {
+                println!("{}", e.render());
+
+                continue;
+            }
+        };
+
+        match command {
+            Commands::Get { key } => {
+                let value = client.get(&key).await?;
+                println!("{value:?}");
+            }
+            Commands::Set {
+                key,
+                value,
+                expiration,
+            } => {
+                client.set(&key, value.as_bytes(), expiration).await?;
+                println!("1");
+            }
+            Commands::Delete { key } => {
+                let value = client.delete(&key).await?;
+                println!("{value:?}");
+            }
+            Commands::Has { key } => {
+                let value = client.has(&key).await?;
+                println!("{value:?}");
+            }
+            Commands::Ttl { key } => {
+                let value = client.ttl(&key).await?;
+                println!("{value:?}");
+            }
+            Commands::Expire { key, seconds } => {
+                let value = client.expire(&key, seconds).await?;
+                println!("{value:?}");
+            }
+            Commands::Persist { key } => {
+                let value = client.persist(&key).await?;
+                println!("{value:?}");
+            }
+
+            Commands::MGet { keys } => {
+                let value = client.m_get(keys).await?;
+
+                println!("{value:?}");
+            }
+
+            Commands::Quit => break,
+        }
+    }
+
+    println!("{input}");
+
+    Ok(())
+}
+
+fn print_welcome() {
+    let version = env!("CARGO_PKG_VERSION");
+    println!("archive-cli {version}");
+}

src/bin/server.rs

@@ -1,27 +1,15 @@
-use config::ServerConfig;
+use archive::Result;
-use errors::AppError;
+use archive::config::ServerConfig;
-use server::Server;
+use archive::server::Server;
+
 use tokio::{signal::ctrl_c, sync::oneshot};
 
-#[cfg(test)]
-pub mod tests;
-
-pub mod client;
-pub mod commands;
-pub mod config;
-pub mod connection;
-pub mod database;
-pub mod errors;
-pub mod handler;
-pub mod server;
-
-pub type Result<T> = std::result::Result<T, AppError>;
-
 #[tokio::main]
 async fn main() -> Result<()> {
     env_logger::builder()
         .format_target(false)
         .filter_level(log::LevelFilter::Info)
+        .parse_env("LOG_LEVEL")
         .parse_default_env()
         .init();
 

src/buffer.rs

@@ -0,0 +1,164 @@
+use std::num::TryFromIntError;
+
+use bytes::{Buf, BufMut, Bytes};
+
+use crate::{Result, errors::AppError};
+
+pub trait ArchiveBuf<B: Buf> {
+    fn try_get_string(&mut self) -> Result<String>;
+    fn try_get_short_string(&mut self) -> Result<String>;
+    fn try_get_bytes(&mut self) -> Result<Bytes>;
+    fn try_get_bool(&mut self) -> Result<bool>;
+    fn try_get_option<T, F, E>(&mut self, f: F) -> Result<Option<T>>
+    where
+        T: Sized,
+        F: FnOnce(&mut B) -> std::result::Result<T, E>,
+        AppError: From<E>;
+    fn try_get_vec<T, F, E>(&mut self, f: F) -> Result<Vec<T>>
+    where
+        T: Sized,
+        F: FnMut(&mut B) -> std::result::Result<T, E>,
+        AppError: From<E>;
+}
+
+pub trait ArchiveBufMut<B: Buf> {
+    fn put_string(&mut self, s: &str) -> Result<()>;
+    fn put_short_string(&mut self, s: &str) -> Result<()>;
+    fn put_bytes_with_length<T: AsRef<[u8]>>(&mut self, bytes: T);
+    fn put_option<T, F>(&mut self, value: Option<T>, f: F)
+    where
+        T: Sized,
+        F: FnOnce(&mut B, T);
+    fn try_put_vec<T, F, V, E>(&mut self, value: V, f: F) -> Result<()>
+    where
+        T: Sized,
+        F: FnMut(&T, &mut B) -> std::result::Result<(), E>,
+        V: AsRef<[T]>,
+        AppError: From<E> + From<TryFromIntError>;
+}
+
+impl<B: Buf> ArchiveBuf<B> for B {
+    fn try_get_string(&mut self) -> Result<String> {
+        let len = self.try_get_u32()? as usize;
+
+        if self.remaining() < len {
+            return Err(AppError::IncompleteBuffer);
+        }
+
+        Ok(String::from_utf8(self.copy_to_bytes(len).to_vec())?)
+    }
+
+    fn try_get_short_string(&mut self) -> Result<String> {
+        let len = self.try_get_u16()? as usize;
+
+        if self.remaining() < len {
+            return Err(AppError::IncompleteBuffer);
+        }
+
+        Ok(String::from_utf8(self.copy_to_bytes(len).to_vec())?)
+    }
+
+    fn try_get_bytes(&mut self) -> Result<Bytes> {
+        let len = self.try_get_u32()? as usize;
+
+        if self.remaining() < len {
+            return Err(AppError::IncompleteBuffer);
+        }
+
+        let data = self.copy_to_bytes(len);
+
+        Ok(data)
+    }
+
+    fn try_get_bool(&mut self) -> Result<bool> {
+        Ok(match self.try_get_u8()? {
+            1 => true,
+            0 => false,
+            _ => return Err(AppError::UnexpectedData),
+        })
+    }
+
+    fn try_get_option<T, F, E>(&mut self, f: F) -> Result<Option<T>>
+    where
+        T: Sized,
+        F: FnOnce(&mut B) -> std::result::Result<T, E>,
+        AppError: From<E>,
+    {
+        if !self.try_get_bool()? {
+            return Ok(None);
+        }
+
+        Ok(Some(f(self)?))
+    }
+
+    fn try_get_vec<T, F, E>(&mut self, mut f: F) -> Result<Vec<T>>
+    where
+        T: Sized,
+        F: FnMut(&mut B) -> std::result::Result<T, E>,
+        AppError: From<E>,
+    {
+        let len = self.try_get_u16()?;
+
+        let mut vec = Vec::with_capacity(len.into());
+
+        for _ in 0..len {
+            vec.push(f(self)?);
+        }
+
+        Ok(vec)
+    }
+}
+
+impl<B: Buf + BufMut> ArchiveBufMut<B> for B {
+    fn put_string(&mut self, s: &str) -> Result<()> {
+        self.put_u32(s.len().try_into()?);
+        self.put_slice(s.as_bytes());
+
+        Ok(())
+    }
+
+    fn put_short_string(&mut self, s: &str) -> Result<()> {
+        self.put_u16(s.len().try_into()?);
+        self.put_slice(s.as_bytes());
+
+        Ok(())
+    }
+
+    fn put_bytes_with_length<T: AsRef<[u8]>>(&mut self, bytes: T) {
+        let bytes = bytes.as_ref();
+
+        self.put_u32(bytes.len() as u32);
+        self.put_slice(bytes);
+    }
+
+    fn put_option<T, F>(&mut self, value: Option<T>, f: F)
+    where
+        T: Sized,
+        F: FnOnce(&mut B, T),
+    {
+        let Some(value) = value else {
+            self.put_u8(0);
+            return;
+        };
+
+        self.put_u8(1);
+        f(self, value);
+    }
+
+    fn try_put_vec<T, F, V, E>(&mut self, vec: V, mut f: F) -> Result<()>
+    where
+        T: Sized,
+        F: FnMut(&T, &mut B) -> std::result::Result<(), E>,
+        V: AsRef<[T]>,
+        AppError: From<E> + From<TryFromIntError>,
+    {
+        let vec = vec.as_ref();
+        self.put_u16(vec.len().try_into()?);
+
+        for element in vec {
+            f(element, self)?;
+        }
+
+        Ok(())
+    }
+}

src/client.rs

@@ -1,7 +1,16 @@
-use bytes::{Buf, BufMut as _, Bytes, BytesMut};
+use bytes::{Buf, Bytes, BytesMut};
 use tokio::net::{TcpStream, ToSocketAddrs};
 
-use crate::{Result, connection::Connection, errors::AppError};
+use crate::{
+    Result,
+    buffer::ArchiveBuf,
+    commands::{
+        delete::Delete, expire::Expire, get::Get, has::Has, m_get::MGet, m_set::MSet,
+        persist::Persist, set::Set, ttl::Ttl,
+    },
+    connection::Connection,
+    errors::AppError,
+};
 
 pub struct Client {
     connection: Connection,
@@ -19,35 +28,16 @@ impl Client {
     pub async fn get(&mut self, key: &str) -> Result<Option<Bytes>> {
         let mut bytes = BytesMut::new();
 
-        bytes.put_u16(3);
+        let cmd = Get::new(key.to_owned());
-        bytes.put_slice(b"get");
+        cmd.put(&mut bytes)?;
-
-        let key_length: u16 = key
-            .len()
-            .try_into()
-            .map_err(|_| AppError::KeyLength(key.len()))?;
-
-        bytes.put_u16(key_length);
-        bytes.put_slice(key.as_bytes());
 
         self.connection.write(bytes.into()).await?;
 
         let mut r = self.get_response().await?;
-        let response = match r.try_get_u8()? {
-            0 => None,
-            1 => {
-                let len = r.try_get_u32()? as usize;
 
-                if r.remaining() < len {
+        let value = r.try_get_option(ArchiveBuf::try_get_bytes)?;
-                    return Err(AppError::InvalidCommandResponse);
-                }
 
-                Some(r.copy_to_bytes(len))
+        Ok(value)
-            }
-            _ => return Err(AppError::InvalidCommandResponse),
-        };
-
-        Ok(response)
     }
 
     pub async fn set(
@@ -56,120 +46,139 @@ impl Client {
         data: &[u8],
         expiration_secs: Option<u64>,
     ) -> Result<()> {
-        let mut bytes = BytesMut::new();
+        let mut buf = BytesMut::new();
 
-        bytes.put_u16(3);
+        let cmd = Set::new(key.to_owned(), data.into(), expiration_secs);
-        bytes.put_slice(b"set");
+        cmd.put(&mut buf)?;
 
-        let key_length: u16 = key
+        self.connection.write(buf.into()).await?;
-            .len()
-            .try_into()
-            .map_err(|_| AppError::KeyLength(key.len()))?;
-
-        bytes.put_u16(key_length);
-        bytes.put_slice(key.as_bytes());
-
-        bytes.put_u32(data.len() as u32);
-        bytes.put_slice(data);
-
-        match expiration_secs {
-            Some(seconds) => {
-                bytes.put_u8(1);
-                bytes.put_u64(seconds);
-            }
-            None => {
-                bytes.put_u8(0);
-            }
-        }
-
-        self.connection.write(bytes.into()).await?;
 
         let mut r = self.get_response().await?;
 
-        match r.try_get_u8()? {
+        if !r.try_get_bool()? {
-            1 => return Ok(()),
+            return Err(AppError::InvalidCommandResponse);
-            _ => return Err(AppError::InvalidCommandResponse),
         }
+
+        Ok(())
     }
 
     pub async fn delete(&mut self, key: &str) -> Result<Option<Bytes>> {
         let mut bytes = BytesMut::new();
 
-        bytes.put_u16(6);
+        let cmd = Delete::new(key.to_owned());
-        bytes.put_slice(b"delete");
+        cmd.put(&mut bytes)?;
-
-        let key_length: u16 = key
-            .len()
-            .try_into()
-            .map_err(|_| AppError::KeyLength(key.len()))?;
-
-        bytes.put_u16(key_length);
-        bytes.put_slice(key.as_bytes());
 
         self.connection.write(bytes.into()).await?;
 
         let mut r = self.get_response().await?;
 
-        let response = match r.try_get_u8()? {
+        let value = r.try_get_option(ArchiveBuf::try_get_bytes)?;
-            1 => {
-                let len = r.try_get_u32()?;
-                let bytes = r.copy_to_bytes(len as usize);
 
-                Some(bytes)
+        Ok(value)
-            }
-            0 => None,
-            _ => return Err(AppError::InvalidCommandResponse),
-        };
-
-        Ok(response)
     }
 
     pub async fn has(&mut self, key: &str) -> Result<bool> {
         let mut bytes = BytesMut::new();
-        bytes.put_u16(3);
-        bytes.put_slice(b"has");
 
-        let key_length: u16 = key
+        let cmd = Has::new(key.to_owned());
-            .len()
+        cmd.put(&mut bytes)?;
-            .try_into()
-            .map_err(|_| AppError::KeyLength(key.len()))?;
-
-        bytes.put_u16(key_length);
-        bytes.put_slice(key.as_bytes());
 
         self.connection.write(bytes.into()).await?;
 
         let mut r = self.get_response().await?;
 
-        Ok(r.try_get_u8()? == 1)
+        let has = r.try_get_bool()?;
+
+        Ok(has)
     }
 
     pub async fn ttl(&mut self, key: &str) -> Result<Option<u64>> {
         let mut bytes = BytesMut::new();
-        bytes.put_u16(3);
-        bytes.put_slice(b"ttl");
 
-        let key_length: u16 = key
+        let cmd = Ttl::new(key.to_owned());
-            .len()
+        cmd.put(&mut bytes)?;
-            .try_into()
-            .map_err(|_| AppError::KeyLength(key.len()))?;
-
-        bytes.put_u16(key_length);
-        bytes.put_slice(key.as_bytes());
 
         self.connection.write(bytes.into()).await?;
 
         let mut r = self.get_response().await?;
 
-        let ttl = match r.try_get_u8()? {
+        let ttl = r.try_get_option(Bytes::try_get_u64)?;
-            1 => Some(r.try_get_u64()?),
-            0 => None,
-            _ => return Err(AppError::InvalidCommandResponse),
-        };
 
         Ok(ttl)
     }
 
+    pub async fn expire(&mut self, key: &str, seconds: u64) -> Result<bool> {
+        let mut bytes = BytesMut::new();
+
+        let cmd = Expire::new(key.to_owned(), seconds);
+        cmd.put(&mut bytes)?;
+
+        self.connection.write(bytes.into()).await?;
+
+        let mut r = self.get_response().await?;
+
+        Ok(r.try_get_bool()?)
+    }
+
+    pub async fn persist(&mut self, key: &str) -> Result<bool> {
+        let mut bytes = BytesMut::new();
+
+        let cmd = Persist::new(key.to_owned());
+        cmd.put(&mut bytes)?;
+
+        self.connection.write(bytes.into()).await?;
+
+        let mut r = self.get_response().await?;
+
+        Ok(r.try_get_bool()?)
+    }
+
+    pub async fn m_set(
+        &mut self,
+        keys: Vec<&str>,
+        data: Vec<&[u8]>,
+        expirations: Vec<Option<u64>>,
+    ) -> Result<()> {
+        let mut bytes = BytesMut::new();
+
+        let len = keys.len().min(data.len()).min(expirations.len());
+
+        let mut sets = Vec::with_capacity(len);
+
+        for i in 0..len {
+            sets.push(Set::new(keys[i].to_owned(), data[i].into(), expirations[i]));
+        }
+
+        let cmd = MSet::new(sets);
+        cmd.put(&mut bytes)?;
+
+        self.connection.write(bytes.into()).await?;
+
+        let mut r = self.get_response().await?;
+
+        if !r.try_get_bool()? {
+            return Err(AppError::InvalidCommandResponse);
+        }
+
+        Ok(())
+    }
+
+    pub async fn m_get(&mut self, keys: Vec<String>) -> Result<Vec<Option<Bytes>>> {
+        let mut bytes = BytesMut::new();
+
+        let gets: Vec<Get> = keys.into_iter().map(Get::new).collect();
+        let cmd = MGet::new(gets);
+        cmd.put(&mut bytes)?;
+
+        self.connection.write(bytes.into()).await?;
+
+        let mut r = self.get_response().await?;
+
+        let values = r.try_get_vec(|b| b.try_get_option(ArchiveBuf::try_get_bytes))?;
+
+        Ok(values)
+    }
+
     async fn get_response(&mut self) -> Result<Bytes> {
         self.connection
             .read_bytes()

src/commands/delete.rs

@@ -1,8 +1,13 @@
 use std::io::Cursor;
 
-use bytes::{Buf as _, BufMut as _, BytesMut};
+use bytes::BytesMut;
 
-use crate::{Result, connection::Connection, database::Database, errors::AppError};
+use crate::{
+    Result,
+    buffer::{ArchiveBuf as _, ArchiveBufMut},
+    connection::Connection,
+    database::Database,
+};
 
 #[derive(Debug, Clone)]
 pub struct Delete {
@@ -10,33 +15,33 @@ pub struct Delete {
 }
 
 impl Delete {
+    pub fn new(key: String) -> Self {
+        Self { key }
+    }
+
     pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
         let value = db.delete(&self.key).await;
 
         let mut buf = BytesMut::new();
-        match value {
+        buf.put_option(value, ArchiveBufMut::put_bytes_with_length);
-            Some(v) => {
-                buf.put_u8(1);
-                buf.put_u32(v.len() as u32);
-                buf.put_slice(&v);
-            }
-            None => buf.put_u8(0),
-        }
 
         connection.write(buf.into()).await?;
 
         Ok(())
     }
 
-    pub fn parse(bytes: &mut Cursor<&[u8]>) -> Result<Self> {
+    pub fn parse(buf: &mut Cursor<&[u8]>) -> Result<Self> {
-        let key_length = bytes.try_get_u16()? as usize;
+        let key = buf.try_get_short_string()?;
-
-        if bytes.remaining() < key_length {
-            return Err(AppError::IncompleteCommandBuffer);
-        }
-
-        let key = String::from_utf8(bytes.copy_to_bytes(key_length).to_vec())?;
 
         Ok(Self { key })
     }
+
+    pub fn put(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string("delete")?;
+        self.put_without_cmd_name(buf)
+    }
+
+    pub fn put_without_cmd_name(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string(&self.key)
+    }
 }

src/commands/expire.rs

@@ -0,0 +1,53 @@
+use std::io::Cursor;
+
+use bytes::{Buf as _, BufMut as _, Bytes, BytesMut};
+
+use crate::{
+    Result,
+    buffer::{ArchiveBuf as _, ArchiveBufMut as _},
+    connection::Connection,
+    database::Database,
+};
+
+#[derive(Debug, Clone)]
+pub struct Expire {
+    key: String,
+    seconds: u64,
+}
+
+impl Expire {
+    pub fn new(key: String, seconds: u64) -> Self {
+        Self { key, seconds }
+    }
+
+    pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
+        let success = db.expire(&self.key, self.seconds).await?;
+
+        connection
+            .write(Bytes::from_static(if success { &[1] } else { &[0] }))
+            .await?;
+
+        Ok(())
+    }
+
+    pub fn parse(buf: &mut Cursor<&[u8]>) -> Result<Self> {
+        let key = buf.try_get_short_string()?;
+        let seconds = buf.try_get_u64()?;
+
+        Ok(Self { key, seconds })
+    }
+
+    pub fn put(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string("expire")?;
+
+        self.put_without_cmd_name(buf)
+    }
+
+    pub fn put_without_cmd_name(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string(&self.key)?;
+
+        buf.put_u64(self.seconds);
+
+        Ok(())
+    }
+}

src/commands/get.rs

@@ -1,44 +1,47 @@
 use std::io::Cursor;
 
-use bytes::{Buf as _, BufMut as _, BytesMut};
+use bytes::BytesMut;
 
-use crate::{Result, connection::Connection, database::Database, errors::AppError};
+use crate::{
+    Result,
+    buffer::{ArchiveBuf as _, ArchiveBufMut},
+    connection::Connection,
+    database::Database,
+};
 
 #[derive(Debug, Clone)]
 pub struct Get {
-    key: String,
+    pub(super) key: String,
 }
 
 impl Get {
+    pub fn new(key: String) -> Self {
+        Self { key }
+    }
+
     pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
         let value = db.get(&self.key).await;
 
         let mut buf = BytesMut::new();
-        match value {
+        buf.put_option(value, ArchiveBufMut::put_bytes_with_length);
-            Some(v) => {
-                buf.put_u8(1);
-                buf.put_u32(v.len() as u32);
-                buf.put_slice(&v);
-            }
-            None => {
-                buf.put_u8(0);
-            }
-        }
 
         connection.write(buf.into()).await?;
 
         Ok(())
     }
 
-    pub fn parse(bytes: &mut Cursor<&[u8]>) -> Result<Self> {
+    pub fn parse(buf: &mut Cursor<&[u8]>) -> Result<Self> {
-        let key_length = bytes.try_get_u16()? as usize;
+        let key = buf.try_get_short_string()?;
-
-        if bytes.remaining() < key_length {
-            return Err(AppError::IncompleteCommandBuffer);
-        }
-
-        let key = String::from_utf8(bytes.copy_to_bytes(key_length).to_vec())?;
 
         Ok(Self { key })
     }
+
+    pub fn put(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string("get")?;
+        self.put_without_cmd_name(buf)
+    }
+
+    pub fn put_without_cmd_name(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string(&self.key)
+    }
 }

src/commands/has.rs

@@ -1,8 +1,13 @@
 use std::io::Cursor;
 
-use bytes::{Buf as _, Bytes};
+use bytes::{Bytes, BytesMut};
 
-use crate::{Result, connection::Connection, database::Database, errors::AppError};
+use crate::{
+    Result,
+    buffer::{ArchiveBuf as _, ArchiveBufMut as _},
+    connection::Connection,
+    database::Database,
+};
 
 #[derive(Debug, Clone)]
 pub struct Has {
@@ -10,6 +15,10 @@ pub struct Has {
 }
 
 impl Has {
+    pub fn new(key: String) -> Self {
+        Self { key }
+    }
+
     pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
         let value = db.has(&self.key).await;
 
@@ -20,15 +29,18 @@ impl Has {
         Ok(())
     }
 
-    pub fn parse(bytes: &mut Cursor<&[u8]>) -> Result<Self> {
+    pub fn parse(buf: &mut Cursor<&[u8]>) -> Result<Self> {
-        let key_length = bytes.try_get_u16()? as usize;
+        let key = buf.try_get_short_string()?;
-
-        if bytes.remaining() < key_length {
-            return Err(AppError::IncompleteCommandBuffer);
-        }
-
-        let key = String::from_utf8(bytes.copy_to_bytes(key_length).to_vec())?;
 
         Ok(Self { key })
     }
+
+    pub fn put(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string("has")?;
+        self.put_without_cmd_name(buf)
+    }
+
+    pub fn put_without_cmd_name(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string(&self.key)
+    }
 }

src/commands/m_get.rs

@@ -0,0 +1,60 @@
+use std::io::Cursor;
+
+use crate::{
+    Result,
+    buffer::{ArchiveBuf as _, ArchiveBufMut},
+    connection::Connection,
+    database::Database,
+    errors::AppError,
+};
+use bytes::BytesMut;
+
+use super::get::Get;
+
+#[derive(Debug, Clone)]
+pub struct MGet {
+    gets: Vec<Get>,
+}
+
+impl MGet {
+    pub fn new(gets: Vec<Get>) -> Self {
+        Self { gets }
+    }
+
+    pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
+        let mut values = Vec::with_capacity(self.gets.len());
+
+        for get in self.gets {
+            values.push(db.get(&get.key).await);
+        }
+
+        let mut buf = BytesMut::new();
+
+        buf.try_put_vec(values, |data, buf| {
+            buf.put_option(data.as_deref(), ArchiveBufMut::put_bytes_with_length);
+
+            Ok::<(), AppError>(())
+        })?;
+
+        connection.write(buf.into()).await?;
+
+        Ok(())
+    }
+
+    pub fn parse(buf: &mut Cursor<&[u8]>) -> Result<Self> {
+        let gets = buf.try_get_vec(Get::parse)?;
+
+        Ok(Self { gets })
+    }
+
+    pub fn put(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string("mget")?;
+        self.put_without_cmd_name(buf)
+    }
+
+    pub fn put_without_cmd_name(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.try_put_vec(&self.gets, Get::put_without_cmd_name)?;
+
+        Ok(())
+    }
+}

src/commands/m_set.rs

@@ -0,0 +1,49 @@
+use std::io::Cursor;
+
+use crate::{
+    Result,
+    buffer::{ArchiveBuf as _, ArchiveBufMut as _},
+    connection::Connection,
+    database::Database,
+};
+use bytes::{Bytes, BytesMut};
+
+use super::set::Set;
+
+#[derive(Debug, Clone)]
+pub struct MSet {
+    sets: Vec<Set>,
+}
+
+impl MSet {
+    pub fn new(sets: Vec<Set>) -> Self {
+        Self { sets }
+    }
+
+    pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
+        for set in self.sets {
+            db.set(set.key, set.data, set.expiration).await?;
+        }
+
+        connection.write(Bytes::from_static(&[1])).await?;
+
+        Ok(())
+    }
+
+    pub fn parse(buf: &mut Cursor<&[u8]>) -> Result<Self> {
+        let sets = buf.try_get_vec(Set::parse)?;
+
+        Ok(Self { sets })
+    }
+
+    pub fn put(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string("mset")?;
+        self.put_without_cmd_name(buf)
+    }
+
+    pub fn put_without_cmd_name(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.try_put_vec(&self.sets, Set::put_without_cmd_name)?;
+
+        Ok(())
+    }
+}

src/commands/mod.rs

@@ -1,19 +1,29 @@
-mod delete;
+pub mod delete;
-mod get;
+pub mod expire;
-mod has;
+pub mod get;
-mod set;
+pub mod has;
-mod ttl;
+pub mod m_get;
+pub mod m_set;
+pub mod persist;
+pub mod set;
+pub mod ttl;
 
 use std::io::Cursor;
 
-use bytes::{Buf, BytesMut};
+use bytes::BytesMut;
 use delete::Delete;
+use expire::Expire;
 use get::Get;
 use has::Has;
+use m_get::MGet;
+use m_set::MSet;
+use persist::Persist;
 use set::Set;
 use ttl::Ttl;
 
-use crate::{Result, connection::Connection, database::Database, errors::AppError};
+use crate::{
+    Result, buffer::ArchiveBuf as _, connection::Connection, database::Database, errors::AppError,
+};
 
 #[derive(Debug)]
 pub enum Command {
@@ -22,6 +32,11 @@ pub enum Command {
     Delete(Delete),
     Has(Has),
     Ttl(Ttl),
+    Expire(Expire),
+    Persist(Persist),
+
+    MSet(MSet),
+    MGet(MGet),
 }
 
 impl Command {
@@ -32,21 +47,20 @@ impl Command {
             Command::Delete(delete) => delete.execute(db, connection).await,
             Command::Has(has) => has.execute(db, connection).await,
             Command::Ttl(ttl) => ttl.execute(db, connection).await,
+            Command::Expire(expire) => expire.execute(db, connection).await,
+            Command::Persist(persist) => persist.execute(db, connection).await,
+
+            Command::MSet(m_set) => m_set.execute(db, connection).await,
+            Command::MGet(m_get) => m_get.execute(db, connection).await,
         }
     }
 
     pub fn parse(bytes: &BytesMut) -> Result<(Self, u64)> {
-        let mut buffer = Cursor::new(&bytes[..]);
+        let mut buf = Cursor::new(&bytes[..]);
 
-        let name_length = buffer.try_get_u16()? as usize;
+        let name = buf.try_get_short_string()?;
 
-        if buffer.remaining() < name_length {
+        Self::parse_inner(name, &mut buf)
-            return Err(AppError::IncompleteCommandBuffer);
-        }
-
-        let name = String::from_utf8(buffer.copy_to_bytes(name_length).to_vec())?;
-
-        Self::parse_inner(name, &mut buffer)
     }
 
     fn parse_inner(command_name: String, bytes: &mut Cursor<&[u8]>) -> Result<(Self, u64)> {
@@ -56,6 +70,10 @@ impl Command {
             "delete" => Self::Delete(Delete::parse(bytes)?),
             "has" => Self::Has(Has::parse(bytes)?),
             "ttl" => Self::Ttl(Ttl::parse(bytes)?),
+            "expire" => Self::Expire(Expire::parse(bytes)?),
+            "persist" => Self::Persist(Persist::parse(bytes)?),
+            "mset" => Self::MSet(MSet::parse(bytes)?),
+            "mget" => Self::MGet(MGet::parse(bytes)?),
             _ => return Err(AppError::UnknownCommand(command_name)),
         };
 

src/commands/persist.rs

@@ -0,0 +1,46 @@
+use std::io::Cursor;
+
+use bytes::{Bytes, BytesMut};
+
+use crate::{
+    Result,
+    buffer::{ArchiveBuf as _, ArchiveBufMut as _},
+    connection::Connection,
+    database::Database,
+};
+
+#[derive(Debug, Clone)]
+pub struct Persist {
+    key: String,
+}
+
+impl Persist {
+    pub fn new(key: String) -> Self {
+        Self { key }
+    }
+
+    pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
+        let value = db.persist(&self.key).await?;
+
+        connection
+            .write(Bytes::from_static(if value { &[1] } else { &[0] }))
+            .await?;
+
+        Ok(())
+    }
+
+    pub fn parse(buf: &mut Cursor<&[u8]>) -> Result<Self> {
+        let key = buf.try_get_short_string()?;
+
+        Ok(Self { key })
+    }
+
+    pub fn put(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string("persist")?;
+        self.put_without_cmd_name(buf)
+    }
+
+    pub fn put_without_cmd_name(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string(&self.key)
+    }
+}

src/commands/set.rs

@@ -1,56 +1,64 @@
-use std::{io::Cursor, time::Duration};
+use std::io::Cursor;
-
-use bytes::{Buf as _, Bytes};
-use tokio::time::Instant;
 
 use crate::{
     Result,
+    buffer::{ArchiveBuf as _, ArchiveBufMut as _},
     connection::Connection,
-    database::{Database, Value},
+    database::Database,
-    errors::AppError,
 };
+use bytes::{Buf as _, BufMut as _, Bytes, BytesMut};
 
 #[derive(Debug, Clone)]
 pub struct Set {
-    key: String,
+    pub(super) key: String,
-    value: Value,
+    pub(super) data: Box<[u8]>,
+    pub(super) expiration: Option<u64>,
 }
 
 impl Set {
+    pub fn new(key: String, data: Box<[u8]>, expiration: Option<u64>) -> Self {
+        Self {
+            key,
+            data,
+            expiration,
+        }
+    }
+
     pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
-        db.set(self.key, self.value).await?;
+        db.set(self.key, self.data, self.expiration).await?;
 
         connection.write(Bytes::from_static(&[1])).await?;
 
         Ok(())
     }
 
-    pub fn parse(bytes: &mut Cursor<&[u8]>) -> Result<Self> {
+    pub fn parse(buf: &mut Cursor<&[u8]>) -> Result<Self> {
-        let key_length = bytes.try_get_u16()? as usize;
+        let key = buf.try_get_short_string()?;
 
-        if bytes.remaining() < key_length {
+        let data = buf.try_get_bytes()?;
-            return Err(AppError::IncompleteCommandBuffer);
-        }
 
-        let key = String::from_utf8(bytes.copy_to_bytes(key_length).to_vec())?;
+        let expiration = buf.try_get_option(Cursor::try_get_u64)?;
-
-        let value_length = bytes.try_get_u32()? as usize;
-
-        if bytes.remaining() < value_length {
-            return Err(AppError::IncompleteCommandBuffer);
-        }
-
-        let data = bytes.copy_to_bytes(value_length);
-
-        let expiration: Option<Instant> = match bytes.try_get_u8()? {
-            1 => Some(Instant::now() + Duration::from_secs(bytes.try_get_u64()?)),
-            0 => None,
-            _ => return Err(AppError::UnexpectedCommandData),
-        };
 
         Ok(Self {
             key,
-            value: Value::new(data, expiration),
+            data: (*data).into(),
+            expiration,
         })
     }
+
+    pub fn put(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string("set")?;
+
+        self.put_without_cmd_name(buf)?;
+
+        Ok(())
+    }
+
+    pub fn put_without_cmd_name(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string(&self.key)?;
+        buf.put_bytes_with_length(&self.data);
+        buf.put_option(self.expiration, BytesMut::put_u64);
+
+        Ok(())
+    }
 }

src/commands/ttl.rs

@@ -1,8 +1,13 @@
 use std::io::Cursor;
 
-use bytes::{Buf as _, BufMut, Bytes, BytesMut};
+use bytes::{BufMut, BytesMut};
 
-use crate::{Result, connection::Connection, database::Database, errors::AppError};
+use crate::{
+    Result,
+    buffer::{ArchiveBuf as _, ArchiveBufMut as _},
+    connection::Connection,
+    database::Database,
+};
 
 #[derive(Debug, Clone)]
 pub struct Ttl {
@@ -10,33 +15,36 @@ pub struct Ttl {
 }
 
 impl Ttl {
+    pub fn new(key: String) -> Self {
+        Self { key }
+    }
+
     pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
         let ttl = db.ttl(&self.key).await;
 
-        let Some(ttl) = ttl else {
-            connection.write(Bytes::from_static(&[0])).await?;
-            return Ok(());
-        };
-
         let mut buf = BytesMut::new();
-
+        buf.put_option(ttl, BytesMut::put_u64);
-        buf.put_u8(1);
-        buf.put_u64(ttl);
 
         connection.write(buf.into()).await?;
 
         Ok(())
     }
 
-    pub fn parse(bytes: &mut Cursor<&[u8]>) -> Result<Self> {
+    pub fn parse(buf: &mut Cursor<&[u8]>) -> Result<Self> {
-        let key_length = bytes.try_get_u16()? as usize;
+        let key = buf.try_get_short_string()?;
-
-        if bytes.remaining() < key_length {
-            return Err(AppError::IncompleteCommandBuffer);
-        }
-
-        let key = String::from_utf8(bytes.copy_to_bytes(key_length).to_vec())?;
 
         Ok(Self { key })
     }
+
+    pub fn put(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string("ttl")?;
+
+        self.put_without_cmd_name(buf)?;
+
+        Ok(())
+    }
+
+    pub fn put_without_cmd_name(&self, buf: &mut BytesMut) -> Result<()> {
+        buf.put_short_string(&self.key)
+    }
 }

src/database.rs

@@ -1,9 +1,10 @@
 use std::{
     collections::{BTreeMap, BTreeSet},
     sync::Arc,
+    time::Duration,
 };
 
-use bytes::{BufMut, Bytes, BytesMut};
+use bytes::{BufMut, BytesMut};
 use byteyarn::Yarn;
 use tokio::{
     sync::{Mutex, Notify},
@@ -32,11 +33,8 @@ pub struct Value {
 }
 
 impl Value {
-    pub fn new(data: Bytes, expiration: Option<Instant>) -> Self {
+    pub fn new(data: Box<[u8]>, expiration: Option<Instant>) -> Self {
-        Self {
+        Self { data, expiration }
-            data: data.into_iter().collect(),
-            expiration,
-        }
     }
 
     pub fn from_string(data: String, expiration: Option<Instant>) -> Self {
@@ -68,13 +66,15 @@ impl Database {
         state.entries.get(key).map(|v| v.data.clone())
     }
 
-    pub async fn set(&self, key: String, value: Value) -> Result<()> {
+    pub async fn set(&self, key: String, data: Box<[u8]>, expiration: Option<u64>) -> Result<()> {
         let mut state = self.state.lock().await;
 
-        let expiration = value.expiration.clone();
+        let expiration = expiration.map(|seconds| Instant::now() + Duration::from_secs(seconds));
 
         let key = Yarn::from(key.clone());
 
+        let value = Value::new(data, expiration);
+
         let previous = state.entries.insert(key.clone(), value);
 
         let mut notify = false;
@@ -138,6 +138,73 @@ impl Database {
             .flatten()
     }
 
+    pub async fn expire(&self, key: &str, seconds: u64) -> Result<bool> {
+        let mut state = self.state.lock().await;
+
+        let key = Yarn::copy(key);
+
+        let expiration = Instant::now() + Duration::from_secs(seconds);
+
+        let notify =
+            state
+                .expirations
+                .iter()
+                .next()
+                .is_none_or(|&(instant, ref next_expiration_key)| {
+                    next_expiration_key == &key || instant > expiration
+                });
+
+        let Some(value) = state.entries.get_mut(&key) else {
+            return Ok(false);
+        };
+
+        let previous_expiration = value.expiration.take();
+        value.expiration = Some(expiration);
+
+        if let Some(previous_expiration) = previous_expiration {
+            state
+                .expirations
+                .remove(&(previous_expiration, key.clone()));
+        };
+
+        state.expirations.insert((expiration, key));
+
+        if notify {
+            self.notify.notify_one();
+        }
+
+        Ok(true)
+    }
+
+    pub async fn persist(&self, key: &str) -> Result<bool> {
+        let mut state = self.state.lock().await;
+
+        let key = Yarn::copy(key);
+
+        let notify = state
+            .expirations
+            .iter()
+            .next()
+            .is_some_and(|&(_, ref next_expiration_key)| next_expiration_key == &key);
+
+        let Some(value) = state.entries.get_mut(&key) else {
+            return Ok(false);
+        };
+
+        match value.expiration.take() {
+            Some(expiration) => {
+                state.expirations.remove(&(expiration, key));
+            }
+            None => return Ok(false),
+        }
+
+        if notify {
+            self.notify.notify_one();
+        }
+
+        Ok(true)
+    }
+
     pub async fn shutdown(&mut self) {
         self.state.lock().await.shutdown = true;
         self.notify.notify_one();

src/errors.rs

@@ -6,8 +6,8 @@ pub enum AppError {
     Io(#[from] std::io::Error),
     #[error("A TryGetError occurred")]
     TryGet(#[from] bytes::TryGetError),
-    #[error("The buffer is missing data for a complete command")]
+    #[error("The buffer is missing data")]
-    IncompleteCommandBuffer,
+    IncompleteBuffer,
     #[error("A Utf8Error occurred")]
     FromUtf8(#[from] std::string::FromUtf8Error),
     #[error("The command {0} was not recognized")]
@@ -18,6 +18,8 @@ pub enum AppError {
     NoResponse,
     #[error("Expected a different response for the executed command")]
     InvalidCommandResponse,
-    #[error("The binary command data is not structured correctly")]
+    #[error("The binary data is not structured correctly")]
-    UnexpectedCommandData,
+    UnexpectedData,
+    #[error("Failed to convert integer")]
+    TryFromInt(#[from] std::num::TryFromIntError),
 }

src/lib.rs

@@ -0,0 +1,16 @@
+use errors::AppError;
+
+#[cfg(test)]
+pub mod tests;
+
+pub mod buffer;
+pub mod client;
+pub mod commands;
+pub mod config;
+pub mod connection;
+pub mod database;
+pub mod errors;
+pub mod handler;
+pub mod server;
+
+pub type Result<T> = std::result::Result<T, AppError>;

src/server.rs

@@ -43,25 +43,27 @@ impl Server {
     pub async fn run(&mut self, mut shutdown: oneshot::Receiver<()>) -> Result<()> {
         let shutdown = &mut shutdown;
 
+        let result = tokio::select! {
+            biased;
+            v = self._run() => v,
+            _ = &mut *shutdown => Ok(()),
+        };
+
+        log::info!("Shutting down");
+        self.db.shutdown().await;
+        let _ = (&mut self.expiration_manager_handle).await;
+
+        result
+    }
+
+    pub async fn _run(&mut self) -> Result<()> {
         loop {
             let permit = Arc::clone(&self.connection_limit)
                 .acquire_owned()
                 .await
                 .unwrap();
 
-            let Some(socket) = ({
+            let socket = self.listener.accept().await?.0;
-                tokio::select! {
-                    socket = self.listener.accept() => Some(socket?.0),
-                    _ = &mut *shutdown => None,
-                }
-            }) else {
-                log::info!("Shutting down");
-
-                self.db.shutdown().await;
-                let _ = (&mut self.expiration_manager_handle).await;
-
-                return Ok(());
-            };
 
             let addr = socket.peer_addr()?;
 
@@ -71,7 +73,7 @@ impl Server {
             tokio::spawn(async move {
                 log::debug!("Spawned a new connection handler: {addr}");
                 if let Err(e) = handler.run().await {
-                    log::debug!("Handler::run error: {e:?}");
+                    log::error!("Handler::run: {e:?}");
                 }
                 log::debug!("Connection handler ended: {addr}");
 

src/tests.rs

@@ -15,7 +15,7 @@ async fn expiration() -> Result<(), Box<dyn std::error::Error>> {
     let mut client = client("127.0.0.1:6171").await?;
 
     client
-        .set("test-key", "test-value".as_bytes(), Some(3))
+        .set("test-key", b"test-value", Some(3))
         .await
         .unwrap();
 
@@ -27,9 +27,70 @@ async fn expiration() -> Result<(), Box<dyn std::error::Error>> {
     assert!(client.has("test-key").await.unwrap());
     assert_eq!(client.ttl("test-key").await.unwrap(), Some(1));
 
+    assert!(client.expire("test-key", 2).await?);
+
     tokio::time::sleep(Duration::from_secs(2)).await;
     assert!(!client.has("test-key").await.unwrap());
 
+    assert!(!client.expire("test-key", 10).await?);
+
+    client.set("test-key", b"test-value", Some(2)).await?;
+
+    assert_eq!(client.ttl("test-key").await?, Some(1));
+    assert!(client.persist("test-key").await?);
+
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    assert_eq!(client.get("test-key").await?, Some("test-value".into()));
+
+    shutdown_tx.send(()).unwrap();
+
+    server_handle.await??;
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn m_set_m_get() -> Result<(), Box<dyn std::error::Error>> {
+    let config = ServerConfig::builder()
+        .host("127.0.0.1".into())
+        .port(6172)
+        .build();
+    let mut server = Server::new(&config).await?;
+
+    let (shutdown_tx, shutdown_rx) = oneshot::channel();
+    let server_handle = tokio::spawn(async move { server.run(shutdown_rx).await });
+
+    let mut client = client("127.0.0.1:6172").await?;
+
+    client
+        .m_set(
+            vec!["key-0", "key-1", "key-2"],
+            vec![b"value-0", b"value-1", b"value-2"],
+            vec![None, Some(2), None],
+        )
+        .await?;
+
+    assert_eq!(
+        client
+            .m_get(vec!["key-0".into(), "key-1".into(), "key-2".into()])
+            .await?,
+        vec![
+            Some("value-0".into()),
+            Some("value-1".into()),
+            Some("value-2".into())
+        ]
+    );
+
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    assert_eq!(
+        client
+            .m_get(vec!["key-0".into(), "key-1".into(), "key-2".into()])
+            .await?,
+        vec![Some("value-0".into()), None, Some("value-2".into())]
+    );
+
     shutdown_tx.send(()).unwrap();
 
     server_handle.await??;