Finish implementation of the KV

3 files changed, 984 insertions, 0 deletions

diff --git a/plugins/kv/boltdb/config.go b/plugins/kv/boltdb/config.go
new file mode 100644
index 00000000..6b116611
--- /dev/null
+++ b/plugins/kv/boltdb/config.go
@@ -0,0 +1,23 @@
+package boltdb
+
+type Config struct {
+	// Dir is a directory to store the DB files
+	Dir string
+	// File is boltDB file. No need to create it by your own,
+	// boltdb driver is able to create the file, or read existing
+	File string
+	// Bucket to store data in boltDB
+	Bucket string
+
+	Permissions int
+	TTL         int
+}
+
+func (s *Config) InitDefaults() error {
+	s.Dir = "."          // current dir
+	s.Bucket = "rr"      // default bucket name
+	s.File = "rr.db"     // default file name
+	s.Permissions = 0777 // free for all
+	s.TTL = 60           // 60 seconds is default TTL
+	return nil
+}
diff --git a/plugins/kv/boltdb/plugin.go b/plugins/kv/boltdb/plugin.go
new file mode 100644
index 00000000..e5eda0c2
--- /dev/null
+++ b/plugins/kv/boltdb/plugin.go
@@ -0,0 +1,476 @@
+package boltdb
+
+import (
+	"bytes"
+	"context"
+	"encoding/gob"
+	"os"
+	"path"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/spiral/errors"
+	"github.com/spiral/roadrunner/v2/plugins/config"
+	"github.com/spiral/roadrunner/v2/plugins/kv"
+	"github.com/spiral/roadrunner/v2/plugins/logger"
+	bolt "go.etcd.io/bbolt"
+)
+
+const PluginName = "boltdb"
+
+// BoltDB K/V storage.
+type Plugin struct {
+	// db instance
+	DB *bolt.DB
+	// name should be UTF-8
+	bucket []byte
+
+	// config for RR integration
+	cfg *Config
+
+	// logger
+	log logger.Logger
+
+	// gc contains key which are contain timeouts
+	gc *sync.Map
+	// default timeout for cache cleanup is 1 minute
+	timeout time.Duration
+
+	// stop is used to stop keys GC and close boltdb connection
+	stop chan struct{}
+}
+
+// NewBoltClient instantiate new BOLTDB client
+// The parameters are:
+// path string 			 -- path to database file (can be placed anywhere), if file is not exist, it will be created
+// perm os.FileMode 	 -- file permissions, for example 0777
+// options *bolt.Options -- boltDB options, such as timeouts, noGrows options and other
+// bucket string 		 -- name of the bucket to use, should be UTF-8
+func NewBoltClient(path string, perm os.FileMode, options *bolt.Options, bucket string, ttl time.Duration) (kv.Storage, error) {
+	const op = errors.Op("newBoltClient")
+	db, err := bolt.Open(path, perm, options)
+	if err != nil {
+		return nil, errors.E(op, err)
+	}
+
+	// bucket should be SET
+	if bucket == "" {
+		return nil, errors.E(op, errors.Str("bucket should be set"))
+	}
+
+	// create bucket if it does not exist
+	// tx.Commit invokes via the db.Update
+	err = db.Update(func(tx *bolt.Tx) error {
+		_, err = tx.CreateBucketIfNotExists([]byte(bucket))
+		if err != nil {
+			return errors.E(op, err)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, errors.E(op, err)
+	}
+
+	// if TTL is not set, make it default
+	if ttl == 0 {
+		ttl = time.Minute
+	}
+
+	s := &Plugin{
+		DB:      db,
+		bucket:  []byte(bucket),
+		stop:    make(chan struct{}),
+		timeout: ttl,
+		gc:      &sync.Map{},
+	}
+
+	// start the TTL gc
+	go s.gcPhase()
+
+	return s, nil
+}
+
+func (s *Plugin) Init(log logger.Logger, cfg config.Configurer) error {
+	const op = errors.Op("boltdb plugin init")
+	s.cfg = &Config{}
+
+	err := cfg.UnmarshalKey(PluginName, &s.cfg)
+	if err != nil {
+		return errors.E(op, errors.Disabled, err)
+	}
+
+	s.log = log
+
+	return nil
+}
+
+func (s *Plugin) Serve() chan error {
+	const op = errors.Op("boltdb serve")
+	errCh := make(chan error, 1)
+
+	db, err := bolt.Open(path.Join(s.cfg.Dir, s.cfg.File), os.FileMode(s.cfg.Permissions), nil)
+	if err != nil {
+		errCh <- errors.E(op, err)
+		return errCh
+	}
+
+	// create bucket if it does not exist
+	// tx.Commit invokes via the db.Update
+	err = db.Update(func(tx *bolt.Tx) error {
+		const upOp = errors.Op("boltdb Update")
+		_, err = tx.CreateBucketIfNotExists([]byte(s.cfg.Bucket))
+		if err != nil {
+			return errors.E(op, upOp)
+		}
+		return nil
+	})
+	if err != nil {
+		errCh <- err
+		return errCh
+	}
+
+	s.DB = db
+	s.bucket = []byte(s.cfg.Bucket)
+	s.stop = make(chan struct{})
+	s.timeout = time.Duration(s.cfg.TTL) * time.Second
+	s.gc = &sync.Map{}
+
+	// start the TTL gc
+	go s.gcPhase()
+
+	return errCh
+}
+
+func (s Plugin) Stop() error {
+	const op = errors.Op("boltdb stop")
+	err := s.Close()
+	if err != nil {
+		return errors.E(op, err)
+	}
+	return nil
+}
+
+func (s Plugin) Has(ctx context.Context, keys ...string) (map[string]bool, error) {
+	const op = errors.Op("boltdb Has")
+	if keys == nil {
+		return nil, errors.E(op, errors.NoKeys)
+	}
+
+	m := make(map[string]bool, len(keys))
+
+	// this is readable transaction
+	err := s.DB.View(func(tx *bolt.Tx) error {
+		// Get retrieves the value for a key in the bucket.
+		// Returns a nil value if the key does not exist or if the key is a nested bucket.
+		// The returned value is only valid for the life of the transaction.
+		for _, key := range keys {
+			keyTrimmed := strings.TrimSpace(key)
+			if keyTrimmed == "" {
+				return errors.E(op, errors.EmptyKey)
+			}
+			b := tx.Bucket(s.bucket)
+			if b == nil {
+				return errors.E(op, errors.NoSuchBucket)
+			}
+			exist := b.Get([]byte(key))
+			if exist != nil {
+				m[key] = true
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return m, nil
+}
+
+// Get retrieves the value for a key in the bucket.
+// Returns a nil value if the key does not exist or if the key is a nested bucket.
+// The returned value is only valid for the life of the transaction.
+func (s Plugin) Get(ctx context.Context, key string) ([]byte, error) {
+	const op = errors.Op("boltdb Get")
+	// to get cases like "  "
+	keyTrimmed := strings.TrimSpace(key)
+	if keyTrimmed == "" {
+		return nil, errors.E(op, errors.EmptyKey)
+	}
+
+	var val []byte
+	err := s.DB.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucket)
+		if b == nil {
+			return errors.E(op, errors.NoSuchBucket)
+		}
+		val = b.Get([]byte(key))
+
+		// try to decode values
+		if val != nil {
+			buf := bytes.NewReader(val)
+			decoder := gob.NewDecoder(buf)
+
+			i := kv.Item{}
+			err := decoder.Decode(&i)
+			if err != nil {
+				// unsafe (w/o runes) convert
+				return errors.E(op, err)
+			}
+
+			// set the value
+			val = []byte(i.Value)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, errors.E(op, err)
+	}
+
+	return val, nil
+}
+
+func (s Plugin) MGet(ctx context.Context, keys ...string) (map[string]interface{}, error) {
+	const op = errors.Op("boltdb MGet")
+	// defence
+	if keys == nil {
+		return nil, errors.E(op, errors.NoKeys)
+	}
+
+	// should not be empty keys
+	for _, key := range keys {
+		keyTrimmed := strings.TrimSpace(key)
+		if keyTrimmed == "" {
+			return nil, errors.E(op, errors.EmptyKey)
+		}
+	}
+
+	m := make(map[string]interface{}, len(keys))
+
+	err := s.DB.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucket)
+		if b == nil {
+			return errors.E(op, errors.NoSuchBucket)
+		}
+
+		for _, key := range keys {
+			value := b.Get([]byte(key))
+			if value != nil {
+				m[key] = value
+			}
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, errors.E(op, err)
+	}
+
+	return m, nil
+}
+
+// Set puts the K/V to the bolt
+func (s Plugin) Set(ctx context.Context, items ...kv.Item) error {
+	const op = errors.Op("boltdb Set")
+	if items == nil {
+		return errors.E(op, errors.NoKeys)
+	}
+
+	// start writable transaction
+	tx, err := s.DB.Begin(true)
+	if err != nil {
+		return errors.E(op, err)
+	}
+	defer func() {
+		err = tx.Commit()
+		if err != nil {
+			errRb := tx.Rollback()
+			if errRb != nil {
+				s.log.Error("during the commit, Rollback error occurred", "commit error", err, "rollback error", errRb)
+			}
+		}
+	}()
+
+	b := tx.Bucket(s.bucket)
+	// use access by index to avoid copying
+	for i := range items {
+		// performance note: pass a prepared bytes slice with initial cap
+		// we can't move buf and gob out of loop, because we need to clear both from data
+		// but gob will contain (w/o re-init) the past data
+		buf := bytes.Buffer{}
+		encoder := gob.NewEncoder(&buf)
+		if errors.Is(errors.EmptyItem, err) {
+			return errors.E(op, errors.EmptyItem)
+		}
+
+		err = encoder.Encode(&items[i])
+		if err != nil {
+			return errors.E(op, err)
+		}
+		// buf.Bytes will copy the underlying slice. Take a look in case of performance problems
+		err = b.Put([]byte(items[i].Key), buf.Bytes())
+		if err != nil {
+			return errors.E(op, err)
+		}
+
+		// if there are no errors, and TTL > 0,  we put the key with timeout to the hashmap, for future check
+		// we do not need mutex here, since we use sync.Map
+		if items[i].TTL != "" {
+			// check correctness of provided TTL
+			_, err := time.Parse(time.RFC3339, items[i].TTL)
+			if err != nil {
+				return errors.E(op, err)
+			}
+			s.gc.Store(items[i].Key, items[i].TTL)
+		}
+
+		buf.Reset()
+	}
+
+	return nil
+}
+
+// Delete all keys from DB
+func (s Plugin) Delete(ctx context.Context, keys ...string) error {
+	const op = errors.Op("boltdb Delete")
+	if keys == nil {
+		return errors.E(op, errors.NoKeys)
+	}
+
+	// should not be empty keys
+	for _, key := range keys {
+		keyTrimmed := strings.TrimSpace(key)
+		if keyTrimmed == "" {
+			return errors.E(op, errors.EmptyKey)
+		}
+	}
+
+	// start writable transaction
+	tx, err := s.DB.Begin(true)
+	if err != nil {
+		return errors.E(op, err)
+	}
+
+	defer func() {
+		err = tx.Commit()
+		if err != nil {
+			errRb := tx.Rollback()
+			if errRb != nil {
+				s.log.Error("during the commit, Rollback error occurred", "commit error", err, "rollback error", errRb)
+			}
+		}
+	}()
+
+	b := tx.Bucket(s.bucket)
+	if b == nil {
+		return errors.E(op, errors.NoSuchBucket)
+	}
+
+	for _, key := range keys {
+		err = b.Delete([]byte(key))
+		if err != nil {
+			return errors.E(op, err)
+		}
+	}
+
+	return nil
+}
+
+// MExpire sets the expiration time to the key
+// If key already has the expiration time, it will be overwritten
+func (s Plugin) MExpire(ctx context.Context, items ...kv.Item) error {
+	const op = errors.Op("boltdb MExpire")
+	for i := range items {
+		if items[i].TTL == "" || strings.TrimSpace(items[i].Key) == "" {
+			return errors.E(op, errors.Str("should set timeout and at least one key"))
+		}
+
+		// verify provided TTL
+		_, err := time.Parse(time.RFC3339, items[i].TTL)
+		if err != nil {
+			return errors.E(op, err)
+		}
+
+		s.gc.Store(items[i].Key, items[i].TTL)
+	}
+	return nil
+}
+
+func (s Plugin) TTL(ctx context.Context, keys ...string) (map[string]interface{}, error) {
+	const op = errors.Op("boltdb TTL")
+	if keys == nil {
+		return nil, errors.E(op, errors.NoKeys)
+	}
+
+	// should not be empty keys
+	for i := range keys {
+		keyTrimmed := strings.TrimSpace(keys[i])
+		if keyTrimmed == "" {
+			return nil, errors.E(op, errors.EmptyKey)
+		}
+	}
+
+	m := make(map[string]interface{}, len(keys))
+
+	for i := range keys {
+		if item, ok := s.gc.Load(keys[i]); ok {
+			// a little bit dangerous operation, but user can't store value other that kv.Item.TTL --> int64
+			m[keys[i]] = item.(string)
+		}
+	}
+	return m, nil
+}
+
+// Close the DB connection
+func (s Plugin) Close() error {
+	// stop the keys GC
+	s.stop <- struct{}{}
+	return s.DB.Close()
+}
+
+// ========================= PRIVATE =================================
+
+func (s Plugin) gcPhase() {
+	t := time.NewTicker(s.timeout)
+	defer t.Stop()
+	for {
+		select {
+		case <-t.C:
+			// calculate current time before loop started to be fair
+			now := time.Now()
+			s.gc.Range(func(key, value interface{}) bool {
+				const op = errors.Op("gcPhase")
+				k := key.(string)
+				v, err := time.Parse(time.RFC3339, value.(string))
+				if err != nil {
+					return false
+				}
+
+				if now.After(v) {
+					// time expired
+					s.gc.Delete(k)
+					err := s.DB.Update(func(tx *bolt.Tx) error {
+						b := tx.Bucket(s.bucket)
+						if b == nil {
+							return errors.E(op, errors.NoSuchBucket)
+						}
+						err := b.Delete([]byte(k))
+						if err != nil {
+							return errors.E(op, err)
+						}
+						return nil
+					})
+					if err != nil {
+						s.log.Error("error during the gc phase of update", "error", err)
+						// todo this error is ignored, it means, that timer still be active
+						// to prevent this, we need to invoke t.Stop()
+						return false
+					}
+				}
+				return true
+			})
+		case <-s.stop:
+			return
+		}
+	}
+}
diff --git a/plugins/kv/boltdb/plugin_unit_test.go b/plugins/kv/boltdb/plugin_unit_test.go
new file mode 100644
index 00000000..a12c830d
--- /dev/null
+++ b/plugins/kv/boltdb/plugin_unit_test.go
@@ -0,0 +1,485 @@
+package boltdb
+
+import (
+	"context"
+	"os"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/spiral/roadrunner/v2/plugins/kv"
+	"github.com/stretchr/testify/assert"
+)
+
+func initStorage() kv.Storage {
+	storage, err := NewBoltClient("rr.db", 0777, nil, "rr", time.Second)
+	if err != nil {
+		panic(err)
+	}
+	return storage
+}
+
+func cleanup(t *testing.T, path string) {
+	err := os.RemoveAll(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestStorage_Has(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+
+	v, err := s.Has(context.Background(), "key")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+}
+
+func TestStorage_Has_Set_Has(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+}
+
+func TestConcurrentReadWriteTransactions(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+
+	wg := &sync.WaitGroup{}
+	wg.Add(3)
+
+	m := &sync.RWMutex{}
+	// concurrently set the keys
+	go func(s kv.Storage) {
+		defer wg.Done()
+		for i := 0; i <= 1000; i++ {
+			m.Lock()
+			// set is writable transaction
+			// it should stop readable
+			assert.NoError(t, s.Set(ctx, kv.Item{
+				Key:   "key" + strconv.Itoa(i),
+				Value: "hello world" + strconv.Itoa(i),
+				TTL:   "",
+			}, kv.Item{
+				Key:   "key2" + strconv.Itoa(i),
+				Value: "hello world" + strconv.Itoa(i),
+				TTL:   "",
+			}))
+			m.Unlock()
+		}
+	}(s)
+
+	// should be no errors
+	go func(s kv.Storage) {
+		defer wg.Done()
+		for i := 0; i <= 1000; i++ {
+			m.RLock()
+			v, err = s.Has(ctx, "key")
+			assert.NoError(t, err)
+			// no such key
+			assert.True(t, v["key"])
+			m.RUnlock()
+		}
+	}(s)
+
+	// should be no errors
+	go func(s kv.Storage) {
+		defer wg.Done()
+		for i := 0; i <= 1000; i++ {
+			m.Lock()
+			err = s.Delete(ctx, "key"+strconv.Itoa(i))
+			assert.NoError(t, err)
+			m.Unlock()
+		}
+	}(s)
+
+	wg.Wait()
+}
+
+func TestStorage_Has_Set_MGet(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+
+	res, err := s.MGet(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.Len(t, res, 2)
+}
+
+func TestStorage_Has_Set_Get(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world2",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+
+	res, err := s.Get(ctx, "key")
+	assert.NoError(t, err)
+
+	if string(res) != "hello world" {
+		t.Fatal("wrong value by key")
+	}
+}
+
+func TestStorage_Set_Del_Get(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+
+	ctx := context.Background()
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	// no such key
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	// no such key
+	assert.True(t, v["key"])
+	assert.True(t, v["key2"])
+
+	// check that keys are present
+	res, err := s.MGet(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.Len(t, res, 2)
+
+	assert.NoError(t, s.Delete(ctx, "key", "key2"))
+	// check that keys are not present
+	res, err = s.MGet(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.Len(t, res, 0)
+}
+
+func TestStorage_Set_GetM(t *testing.T) {
+	s := initStorage()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+	ctx := context.Background()
+
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}, kv.Item{
+		Key:   "key2",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	res, err := s.MGet(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.Len(t, res, 2)
+}
+
+func TestNilAndWrongArgs(t *testing.T) {
+	s := initStorage()
+	ctx := context.Background()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+
+	// check
+	v, err := s.Has(ctx, "key")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+
+	_, err = s.Has(ctx, "")
+	assert.Error(t, err)
+
+	_, err = s.Get(ctx, "")
+	assert.Error(t, err)
+
+	_, err = s.Get(ctx, " ")
+	assert.Error(t, err)
+
+	_, err = s.Get(ctx, "                 ")
+	assert.Error(t, err)
+
+	_, err = s.MGet(ctx, "key", "key2", "")
+	assert.Error(t, err)
+
+	_, err = s.MGet(ctx, "key", "key2", "   ")
+	assert.Error(t, err)
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	}))
+
+	assert.Error(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "asdf",
+	}))
+
+	_, err = s.Has(ctx, "key")
+	assert.NoError(t, err)
+
+	assert.Error(t, s.Set(ctx, kv.Item{}))
+
+	err = s.Delete(ctx, "")
+	assert.Error(t, err)
+
+	err = s.Delete(ctx, "key", "")
+	assert.Error(t, err)
+
+	err = s.Delete(ctx, "key", "     ")
+	assert.Error(t, err)
+
+	err = s.Delete(ctx, "key")
+	assert.NoError(t, err)
+}
+
+func TestStorage_MExpire_TTL(t *testing.T) {
+	s := initStorage()
+	ctx := context.Background()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+
+	// ensure that storage is clean
+	v, err := s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+	assert.False(t, v["key2"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	},
+		kv.Item{
+			Key:   "key2",
+			Value: "hello world",
+			TTL:   "",
+		}))
+	// set timeout to 5 sec
+	nowPlusFive := time.Now().Add(time.Second * 5).Format(time.RFC3339)
+
+	i1 := kv.Item{
+		Key:   "key",
+		Value: "",
+		TTL:   nowPlusFive,
+	}
+	i2 := kv.Item{
+		Key:   "key2",
+		Value: "",
+		TTL:   nowPlusFive,
+	}
+	assert.NoError(t, s.MExpire(ctx, i1, i2))
+
+	time.Sleep(time.Second * 6)
+
+	// ensure that storage is clean
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+	assert.False(t, v["key2"])
+}
+
+func TestStorage_SetExpire_TTL(t *testing.T) {
+	s := initStorage()
+	ctx := context.Background()
+	defer func() {
+		if err := s.Close(); err != nil {
+			panic(err)
+		}
+		cleanup(t, "rr.db")
+	}()
+
+	// ensure that storage is clean
+	v, err := s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+	assert.False(t, v["key2"])
+
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "hello world",
+		TTL:   "",
+	},
+		kv.Item{
+			Key:   "key2",
+			Value: "hello world",
+			TTL:   "",
+		}))
+
+	nowPlusFive := time.Now().Add(time.Second * 5).Format(time.RFC3339)
+
+	// set timeout to 5 sec
+	assert.NoError(t, s.Set(ctx, kv.Item{
+		Key:   "key",
+		Value: "value",
+		TTL:   nowPlusFive,
+	},
+		kv.Item{
+			Key:   "key2",
+			Value: "value",
+			TTL:   nowPlusFive,
+		}))
+
+	time.Sleep(time.Second * 2)
+	m, err := s.TTL(ctx, "key", "key2")
+	assert.NoError(t, err)
+
+	// remove a precision 4.02342342 -> 4
+	keyTTL, err := strconv.Atoi(m["key"].(string)[0:1])
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// remove a precision 4.02342342 -> 4
+	key2TTL, err := strconv.Atoi(m["key"].(string)[0:1])
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.True(t, keyTTL < 5)
+	assert.True(t, key2TTL < 5)
+
+	time.Sleep(time.Second * 4)
+
+	// ensure that storage is clean
+	v, err = s.Has(ctx, "key", "key2")
+	assert.NoError(t, err)
+	assert.False(t, v["key"])
+	assert.False(t, v["key2"])
+}