Merge pull request #361 from superseriousbusiness/media_refactorv0.2.0

Refactor media handler to allow async media resolution

7 files changed, 404 insertions, 56 deletions

diff --git a/vendor/codeberg.org/gruf/go-store/storage/block.go b/vendor/codeberg.org/gruf/go-store/storage/block.go
index 9a8c4dc7d..c50faa10b 100644
--- a/vendor/codeberg.org/gruf/go-store/storage/block.go
+++ b/vendor/codeberg.org/gruf/go-store/storage/block.go
@@ -87,6 +87,7 @@ type BlockStorage struct {
 	config    BlockConfig      // cfg is the supplied configuration for this store
 	hashPool  sync.Pool        // hashPool is this store's hashEncoder pool
 	bufpool   pools.BufferPool // bufpool is this store's bytes.Buffer pool
+	lock      *Lock            // lock is the opened lockfile for this storage instance
 
 	// NOTE:
 	// BlockStorage does not need to lock each of the underlying block files
@@ -138,6 +139,12 @@ func OpenBlock(path string, cfg *BlockConfig) (*BlockStorage, error) {
 		return nil, errPathIsFile
 	}
 
+	// Open and acquire storage lock for path
+	lock, err := OpenLock(pb.Join(path, LockFile))
+	if err != nil {
+		return nil, err
+	}
+
 	// Figure out the largest size for bufpool slices
 	bufSz := encodedHashLen
 	if bufSz < config.BlockSize {
@@ -159,19 +166,29 @@ func OpenBlock(path string, cfg *BlockConfig) (*BlockStorage, error) {
 			},
 		},
 		bufpool: pools.NewBufferPool(bufSz),
+		lock:    lock,
 	}, nil
 }
 
 // Clean implements storage.Clean()
 func (st *BlockStorage) Clean() error {
-	nodes := map[string]*node{}
+	// Track open
+	st.lock.Add()
+	defer st.lock.Done()
+
+	// Check if open
+	if st.lock.Closed() {
+		return ErrClosed
+	}
 
 	// Acquire path builder
 	pb := util.GetPathBuilder()
 	defer util.PutPathBuilder(pb)
 
-	// Walk nodes dir for entries
+	nodes := map[string]*node{}
 	onceErr := errors.OnceError{}
+
+	// Walk nodes dir for entries
 	err := util.WalkDir(pb, st.nodePath, func(npath string, fsentry fs.DirEntry) {
 		// Only deal with regular files
 		if !fsentry.Type().IsRegular() {
@@ -293,6 +310,7 @@ func (st *BlockStorage) ReadBytes(key string) ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer rc.Close()
 
 	// Read all bytes and return
 	return io.ReadAll(rc)
@@ -306,9 +324,19 @@ func (st *BlockStorage) ReadStream(key string) (io.ReadCloser, error) {
 		return nil, err
 	}
 
+	// Track open
+	st.lock.Add()
+
+	// Check if open
+	if st.lock.Closed() {
+		st.lock.Done()
+		return nil, ErrClosed
+	}
+
 	// Attempt to open RO file
 	file, err := open(npath, defaultFileROFlags)
 	if err != nil {
+		st.lock.Done()
 		return nil, err
 	}
 	defer file.Close()
@@ -328,14 +356,16 @@ func (st *BlockStorage) ReadStream(key string) (io.ReadCloser, error) {
 		nil,
 	)
 	if err != nil {
+		st.lock.Done()
 		return nil, err
 	}
 
-	// Return new block reader
-	return util.NopReadCloser(&blockReader{
+	// Prepare block reader and return
+	rc := util.NopReadCloser(&blockReader{
 		storage: st,
 		node:    &node,
-	}), nil
+	}) // we wrap the blockreader to decr lockfile waitgroup
+	return util.ReadCloserWithCallback(rc, st.lock.Done), nil
 }
 
 func (st *BlockStorage) readBlock(key string) ([]byte, error) {
@@ -373,6 +403,15 @@ func (st *BlockStorage) WriteStream(key string, r io.Reader) error {
 		return err
 	}
 
+	// Track open
+	st.lock.Add()
+	defer st.lock.Done()
+
+	// Check if open
+	if st.lock.Closed() {
+		return ErrClosed
+	}
+
 	// Check if this exists
 	ok, err := stat(key)
 	if err != nil {
@@ -443,11 +482,16 @@ loop:
 			continue loop
 		}
 
-		// Write in separate goroutine
+		// Check if reached EOF
+		atEOF := (n < buf.Len())
+
 		wg.Add(1)
 		go func() {
-			// Defer buffer release + signal done
+			// Perform writes in goroutine
+
 			defer func() {
+				// Defer release +
+				// signal we're done
 				st.bufpool.Put(buf)
 				wg.Done()
 			}()
@@ -460,8 +504,8 @@ loop:
 			}
 		}()
 
-		// We reached EOF
-		if n < buf.Len() {
+		// Break at end
+		if atEOF {
 			break loop
 		}
 	}
@@ -552,6 +596,15 @@ func (st *BlockStorage) Stat(key string) (bool, error) {
 		return false, err
 	}
 
+	// Track open
+	st.lock.Add()
+	defer st.lock.Done()
+
+	// Check if open
+	if st.lock.Closed() {
+		return false, ErrClosed
+	}
+
 	// Check for file on disk
 	return stat(kpath)
 }
@@ -564,12 +617,35 @@ func (st *BlockStorage) Remove(key string) error {
 		return err
 	}
 
+	// Track open
+	st.lock.Add()
+	defer st.lock.Done()
+
+	// Check if open
+	if st.lock.Closed() {
+		return ErrClosed
+	}
+
 	// Attempt to remove file
 	return os.Remove(kpath)
 }
 
+// Close implements Storage.Close()
+func (st *BlockStorage) Close() error {
+	return st.lock.Close()
+}
+
 // WalkKeys implements Storage.WalkKeys()
 func (st *BlockStorage) WalkKeys(opts WalkKeysOptions) error {
+	// Track open
+	st.lock.Add()
+	defer st.lock.Done()
+
+	// Check if open
+	if st.lock.Closed() {
+		return ErrClosed
+	}
+
 	// Acquire path builder
 	pb := util.GetPathBuilder()
 	defer util.PutPathBuilder(pb)
@@ -610,7 +686,7 @@ func (st *BlockStorage) blockPathForKey(hash string) string {
 }
 
 // hashSeparator is the separating byte between block hashes
-const hashSeparator = byte(':')
+const hashSeparator = byte('\n')
 
 // node represents the contents of a node file in storage
 type node struct {
@@ -773,24 +849,27 @@ func (r *blockReader) Read(b []byte) (int, error) {
 	}
 }
 
+var (
+	// base64Encoding is our base64 encoding object.
+	base64Encoding = hashenc.Base64()
+
+	// encodedHashLen is the once-calculated encoded hash-sum length
+	encodedHashLen = base64Encoding.EncodedLen(
+		sha256.New().Size(),
+	)
+)
+
 // hashEncoder is a HashEncoder with built-in encode buffer
 type hashEncoder struct {
 	henc hashenc.HashEncoder
 	ebuf []byte
 }
 
-// encodedHashLen is the once-calculated encoded hash-sum length
-var encodedHashLen = hashenc.Base64().EncodedLen(
-	sha256.New().Size(),
-)
-
 // newHashEncoder returns a new hashEncoder instance
 func newHashEncoder() *hashEncoder {
-	hash := sha256.New()
-	enc := hashenc.Base64()
 	return &hashEncoder{
-		henc: hashenc.New(hash, enc),
-		ebuf: make([]byte, enc.EncodedLen(hash.Size())),
+		henc: hashenc.New(sha256.New(), base64Encoding),
+		ebuf: make([]byte, encodedHashLen),
 	}
 }
 
diff --git a/vendor/codeberg.org/gruf/go-store/storage/disk.go b/vendor/codeberg.org/gruf/go-store/storage/disk.go
index 060d56688..287042886 100644
--- a/vendor/codeberg.org/gruf/go-store/storage/disk.go
+++ b/vendor/codeberg.org/gruf/go-store/storage/disk.go
@@ -5,6 +5,8 @@ import (
 	"io/fs"
 	"os"
 	"path"
+	_path "path"
+	"strings"
 	"syscall"
 
 	"codeberg.org/gruf/go-bytes"
@@ -31,6 +33,11 @@ type DiskConfig struct {
 	// Overwrite allows overwriting values of stored keys in the storage
 	Overwrite bool
 
+	// LockFile allows specifying the filesystem path to use for the lockfile,
+	// providing only a filename it will store the lockfile within provided store
+	// path and nest the store under `path/store` to prevent access to lockfile
+	LockFile string
+
 	// Compression is the Compressor to use when reading / writing files, default is no compression
 	Compression Compressor
 }
@@ -57,11 +64,17 @@ func getDiskConfig(cfg *DiskConfig) DiskConfig {
 		cfg.WriteBufSize = DefaultDiskConfig.WriteBufSize
 	}
 
+	// Assume empty lockfile path == use default
+	if len(cfg.LockFile) < 1 {
+		cfg.LockFile = LockFile
+	}
+
 	// Return owned config copy
 	return DiskConfig{
 		Transform:    cfg.Transform,
 		WriteBufSize: cfg.WriteBufSize,
 		Overwrite:    cfg.Overwrite,
+		LockFile:     cfg.LockFile,
 		Compression:  cfg.Compression,
 	}
 }
@@ -71,23 +84,35 @@ type DiskStorage struct {
 	path   string           // path is the root path of this store
 	bufp   pools.BufferPool // bufp is the buffer pool for this DiskStorage
 	config DiskConfig       // cfg is the supplied configuration for this store
+	lock   *Lock            // lock is the opened lockfile for this storage instance
 }
 
 // OpenFile opens a DiskStorage instance for given folder path and configuration
 func OpenFile(path string, cfg *DiskConfig) (*DiskStorage, error) {
+	// Get checked config
+	config := getDiskConfig(cfg)
+
 	// Acquire path builder
 	pb := util.GetPathBuilder()
 	defer util.PutPathBuilder(pb)
 
-	// Clean provided path, ensure ends in '/' (should
-	// be dir, this helps with file path trimming later)
-	path = pb.Clean(path) + "/"
+	// Clean provided store path, ensure
+	// ends in '/' to help later path trimming
+	storePath := pb.Clean(path) + "/"
 
-	// Get checked config
-	config := getDiskConfig(cfg)
+	// Clean provided lockfile path
+	lockfile := pb.Clean(config.LockFile)
+
+	// Check if lockfile is an *actual* path or just filename
+	if lockDir, _ := _path.Split(lockfile); len(lockDir) < 1 {
+		// Lockfile is a filename, store must be nested under
+		// $storePath/store to prevent access to the lockfile
+		storePath += "store/"
+		lockfile = pb.Join(path, lockfile)
+	}
 
 	// Attempt to open dir path
-	file, err := os.OpenFile(path, defaultFileROFlags, defaultDirPerms)
+	file, err := os.OpenFile(storePath, defaultFileROFlags, defaultDirPerms)
 	if err != nil {
 		// If not a not-exist error, return
 		if !os.IsNotExist(err) {
@@ -95,13 +120,13 @@ func OpenFile(path string, cfg *DiskConfig) (*DiskStorage, error) {
 		}
 
 		// Attempt to make store path dirs
-		err = os.MkdirAll(path, defaultDirPerms)
+		err = os.MkdirAll(storePath, defaultDirPerms)
 		if err != nil {
 			return nil, err
 		}
 
 		// Reopen dir now it's been created
-		file, err = os.OpenFile(path, defaultFileROFlags, defaultDirPerms)
+		file, err = os.OpenFile(storePath, defaultFileROFlags, defaultDirPerms)
 		if err != nil {
 			return nil, err
 		}
@@ -116,16 +141,28 @@ func OpenFile(path string, cfg *DiskConfig) (*DiskStorage, error) {
 		return nil, errPathIsFile
 	}
 
+	// Open and acquire storage lock for path
+	lock, err := OpenLock(lockfile)
+	if err != nil {
+		return nil, err
+	}
+
 	// Return new DiskStorage
 	return &DiskStorage{
-		path:   path,
+		path:   storePath,
 		bufp:   pools.NewBufferPool(config.WriteBufSize),
 		config: config,
+		lock:   lock,
 	}, nil
 }
 
 // Clean implements Storage.Clean()
 func (st *DiskStorage) Clean() error {
+	st.lock.Add()
+	defer st.lock.Done()
+	if st.lock.Closed() {
+		return ErrClosed
+	}
 	return util.CleanDirs(st.path)
 }
 
@@ -150,9 +187,18 @@ func (st *DiskStorage) ReadStream(key string) (io.ReadCloser, error) {
 		return nil, err
 	}
 
+	// Track open
+	st.lock.Add()
+
+	// Check if open
+	if st.lock.Closed() {
+		return nil, ErrClosed
+	}
+
 	// Attempt to open file (replace ENOENT with our own)
 	file, err := open(kpath, defaultFileROFlags)
 	if err != nil {
+		st.lock.Done()
 		return nil, errSwapNotFound(err)
 	}
 
@@ -160,12 +206,14 @@ func (st *DiskStorage) ReadStream(key string) (io.ReadCloser, error) {
 	cFile, err := st.config.Compression.Reader(file)
 	if err != nil {
 		file.Close() // close this here, ignore error
+		st.lock.Done()
 		return nil, err
 	}
 
 	// Wrap compressor to ensure file close
 	return util.ReadCloserWithCallback(cFile, func() {
 		file.Close()
+		st.lock.Done()
 	}), nil
 }
 
@@ -182,6 +230,15 @@ func (st *DiskStorage) WriteStream(key string, r io.Reader) error {
 		return err
 	}
 
+	// Track open
+	st.lock.Add()
+	defer st.lock.Done()
+
+	// Check if open
+	if st.lock.Closed() {
+		return ErrClosed
+	}
+
 	// Ensure dirs leading up to file exist
 	err = os.MkdirAll(path.Dir(kpath), defaultDirPerms)
 	if err != nil {
@@ -232,6 +289,15 @@ func (st *DiskStorage) Stat(key string) (bool, error) {
 		return false, err
 	}
 
+	// Track open
+	st.lock.Add()
+	defer st.lock.Done()
+
+	// Check if open
+	if st.lock.Closed() {
+		return false, ErrClosed
+	}
+
 	// Check for file on disk
 	return stat(kpath)
 }
@@ -244,20 +310,44 @@ func (st *DiskStorage) Remove(key string) error {
 		return err
 	}
 
+	// Track open
+	st.lock.Add()
+	defer st.lock.Done()
+
+	// Check if open
+	if st.lock.Closed() {
+		return ErrClosed
+	}
+
 	// Attempt to remove file
 	return os.Remove(kpath)
 }
 
+// Close implements Storage.Close()
+func (st *DiskStorage) Close() error {
+	return st.lock.Close()
+}
+
 // WalkKeys implements Storage.WalkKeys()
 func (st *DiskStorage) WalkKeys(opts WalkKeysOptions) error {
+	// Track open
+	st.lock.Add()
+	defer st.lock.Done()
+
+	// Check if open
+	if st.lock.Closed() {
+		return ErrClosed
+	}
+
 	// Acquire path builder
 	pb := util.GetPathBuilder()
 	defer util.PutPathBuilder(pb)
 
 	// Walk dir for entries
 	return util.WalkDir(pb, st.path, func(kpath string, fsentry fs.DirEntry) {
-		// Only deal with regular files
 		if fsentry.Type().IsRegular() {
+			// Only deal with regular files
+
 			// Get full item path (without root)
 			kpath = pb.Join(kpath, fsentry.Name())[len(st.path):]
 
@@ -269,21 +359,39 @@ func (st *DiskStorage) WalkKeys(opts WalkKeysOptions) error {
 
 // filepath checks and returns a formatted filepath for given key
 func (st *DiskStorage) filepath(key string) (string, error) {
+	// Calculate transformed key path
+	key = st.config.Transform.KeyToPath(key)
+
 	// Acquire path builder
 	pb := util.GetPathBuilder()
 	defer util.PutPathBuilder(pb)
 
-	// Calculate transformed key path
-	key = st.config.Transform.KeyToPath(key)
-
 	// Generated joined root path
 	pb.AppendString(st.path)
 	pb.AppendString(key)
 
 	// Check for dir traversal outside of root
-	if util.IsDirTraversal(st.path, pb.StringPtr()) {
+	if isDirTraversal(st.path, pb.StringPtr()) {
 		return "", ErrInvalidKey
 	}
 
 	return pb.String(), nil
 }
+
+// isDirTraversal will check if rootPlusPath is a dir traversal outside of root,
+// assuming that both are cleaned and that rootPlusPath is path.Join(root, somePath)
+func isDirTraversal(root, rootPlusPath string) bool {
+	switch {
+	// Root is $PWD, check for traversal out of
+	case root == ".":
+		return strings.HasPrefix(rootPlusPath, "../")
+
+	// The path MUST be prefixed by root
+	case !strings.HasPrefix(rootPlusPath, root):
+		return true
+
+	// In all other cases, check not equal
+	default:
+		return len(root) == len(rootPlusPath)
+	}
+}
diff --git a/vendor/codeberg.org/gruf/go-store/storage/errors.go b/vendor/codeberg.org/gruf/go-store/storage/errors.go
index 016593596..ad2b742e6 100644
--- a/vendor/codeberg.org/gruf/go-store/storage/errors.go
+++ b/vendor/codeberg.org/gruf/go-store/storage/errors.go
@@ -19,6 +19,9 @@ func (e errorString) Extend(s string, a ...interface{}) errorString {
 }
 
 var (
+	// ErrClosed is returned on operations on a closed storage
+	ErrClosed = errorString("store/storage: closed")
+
 	// ErrNotFound is the error returned when a key cannot be found in storage
 	ErrNotFound = errorString("store/storage: key not found")
 
@@ -39,6 +42,9 @@ var (
 
 	// errCorruptNodes is returned when nodes with missing blocks are found during a BlockStorage clean
 	errCorruptNodes = errorString("store/storage: corrupted nodes")
+
+	// ErrAlreadyLocked is returned on fail opening a storage lockfile
+	ErrAlreadyLocked = errorString("store/storage: storage lock already open")
 )
 
 // errSwapNoop performs no error swaps
@@ -61,3 +67,11 @@ func errSwapExist(err error) error {
 	}
 	return err
 }
+
+// errSwapUnavailable swaps syscall.EAGAIN for ErrAlreadyLocked
+func errSwapUnavailable(err error) error {
+	if err == syscall.EAGAIN {
+		return ErrAlreadyLocked
+	}
+	return err
+}
diff --git a/vendor/codeberg.org/gruf/go-store/storage/fs.go b/vendor/codeberg.org/gruf/go-store/storage/fs.go
index 444cee4b0..b1c3560d2 100644
--- a/vendor/codeberg.org/gruf/go-store/storage/fs.go
+++ b/vendor/codeberg.org/gruf/go-store/storage/fs.go
@@ -8,11 +8,14 @@ import (
 )
 
 const (
-	defaultDirPerms      = 0755
-	defaultFilePerms     = 0644
+	// default file permission bits
+	defaultDirPerms  = 0755
+	defaultFilePerms = 0644
+
+	// default file open flags
 	defaultFileROFlags   = syscall.O_RDONLY
 	defaultFileRWFlags   = syscall.O_CREAT | syscall.O_RDWR
-	defaultFileLockFlags = syscall.O_RDONLY | syscall.O_EXCL | syscall.O_CREAT
+	defaultFileLockFlags = syscall.O_RDONLY | syscall.O_CREAT
 )
 
 // NOTE:
@@ -39,7 +42,7 @@ func stat(path string) (bool, error) {
 		return syscall.Stat(path, &stat)
 	})
 	if err != nil {
-		if err == syscall.ENOENT {
+		if err == syscall.ENOENT { //nolint
 			err = nil
 		}
 		return false, err
diff --git a/vendor/codeberg.org/gruf/go-store/storage/lock.go b/vendor/codeberg.org/gruf/go-store/storage/lock.go
index 3d794cda9..8a6c4c5e8 100644
--- a/vendor/codeberg.org/gruf/go-store/storage/lock.go
+++ b/vendor/codeberg.org/gruf/go-store/storage/lock.go
@@ -1,34 +1,76 @@
 package storage
 
 import (
-	"os"
+	"sync"
+	"sync/atomic"
 	"syscall"
 
 	"codeberg.org/gruf/go-store/util"
 )
 
-type lockableFile struct {
-	*os.File
+// LockFile is our standard lockfile name.
+const LockFile = "store.lock"
+
+// Lock represents a filesystem lock to ensure only one storage instance open per path.
+type Lock struct {
+	fd int
+	wg sync.WaitGroup
+	st uint32
 }
 
-func openLock(path string) (*lockableFile, error) {
-	file, err := open(path, defaultFileLockFlags)
+// OpenLock opens a lockfile at path.
+func OpenLock(path string) (*Lock, error) {
+	var fd int
+
+	// Open the file descriptor at path
+	err := util.RetryOnEINTR(func() (err error) {
+		fd, err = syscall.Open(path, defaultFileLockFlags, defaultFilePerms)
+		return
+	})
 	if err != nil {
 		return nil, err
 	}
-	return &lockableFile{file}, nil
+
+	// Get a flock on the file descriptor
+	err = util.RetryOnEINTR(func() error {
+		return syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB)
+	})
+	if err != nil {
+		return nil, errSwapUnavailable(err)
+	}
+
+	return &Lock{fd: fd}, nil
 }
 
-func (f *lockableFile) lock() error {
-	return f.flock(syscall.LOCK_EX | syscall.LOCK_NB)
+// Add will add '1' to the underlying sync.WaitGroup.
+func (f *Lock) Add() {
+	f.wg.Add(1)
 }
 
-func (f *lockableFile) unlock() error {
-	return f.flock(syscall.LOCK_UN | syscall.LOCK_NB)
+// Done will decrememnt '1' from the underlying sync.WaitGroup.
+func (f *Lock) Done() {
+	f.wg.Done()
 }
 
-func (f *lockableFile) flock(how int) error {
-	return util.RetryOnEINTR(func() error {
-		return syscall.Flock(int(f.Fd()), how)
-	})
+// Close will attempt to close the lockfile and file descriptor.
+func (f *Lock) Close() error {
+	var err error
+	if atomic.CompareAndSwapUint32(&f.st, 0, 1) {
+		// Wait until done
+		f.wg.Wait()
+
+		// Ensure gets closed
+		defer syscall.Close(f.fd)
+
+		// Call funlock on the file descriptor
+		err = util.RetryOnEINTR(func() error {
+			return syscall.Flock(f.fd, syscall.LOCK_UN|syscall.LOCK_NB)
+		})
+	}
+	return err
+}
+
+// Closed will return whether this lockfile has been closed (and unlocked).
+func (f *Lock) Closed() bool {
+	return (atomic.LoadUint32(&f.st) == 1)
 }
diff --git a/vendor/codeberg.org/gruf/go-store/storage/memory.go b/vendor/codeberg.org/gruf/go-store/storage/memory.go
index be60fa464..2dab562d6 100644
--- a/vendor/codeberg.org/gruf/go-store/storage/memory.go
+++ b/vendor/codeberg.org/gruf/go-store/storage/memory.go
@@ -2,6 +2,7 @@ package storage
 
 import (
 	"io"
+	"sync"
 
 	"codeberg.org/gruf/go-bytes"
 	"codeberg.org/gruf/go-store/util"
@@ -10,36 +11,76 @@ import (
 // MemoryStorage is a storage implementation that simply stores key-value
 // pairs in a Go map in-memory. The map is protected by a mutex.
 type MemoryStorage struct {
+	ow bool // overwrites
 	fs map[string][]byte
+	mu sync.Mutex
+	st uint32
 }
 
 // OpenMemory opens a new MemoryStorage instance with internal map of 'size'.
-func OpenMemory(size int) *MemoryStorage {
+func OpenMemory(size int, overwrites bool) *MemoryStorage {
 	return &MemoryStorage{
 		fs: make(map[string][]byte, size),
+		mu: sync.Mutex{},
+		ow: overwrites,
 	}
 }
 
 // Clean implements Storage.Clean().
 func (st *MemoryStorage) Clean() error {
+	st.mu.Lock()
+	defer st.mu.Unlock()
+	if st.st == 1 {
+		return ErrClosed
+	}
 	return nil
 }
 
 // ReadBytes implements Storage.ReadBytes().
 func (st *MemoryStorage) ReadBytes(key string) ([]byte, error) {
+	// Lock storage
+	st.mu.Lock()
+
+	// Check store open
+	if st.st == 1 {
+		st.mu.Unlock()
+		return nil, ErrClosed
+	}
+
+	// Check for key
 	b, ok := st.fs[key]
+	st.mu.Unlock()
+
+	// Return early if not exist
 	if !ok {
 		return nil, ErrNotFound
 	}
+
+	// Create return copy
 	return bytes.Copy(b), nil
 }
 
 // ReadStream implements Storage.ReadStream().
 func (st *MemoryStorage) ReadStream(key string) (io.ReadCloser, error) {
+	// Lock storage
+	st.mu.Lock()
+
+	// Check store open
+	if st.st == 1 {
+		st.mu.Unlock()
+		return nil, ErrClosed
+	}
+
+	// Check for key
 	b, ok := st.fs[key]
+	st.mu.Unlock()
+
+	// Return early if not exist
 	if !ok {
 		return nil, ErrNotFound
 	}
+
+	// Create io.ReadCloser from 'b' copy
 	b = bytes.Copy(b)
 	r := bytes.NewReader(b)
 	return util.NopReadCloser(r), nil
@@ -47,43 +88,101 @@ func (st *MemoryStorage) ReadStream(key string) (io.ReadCloser, error) {
 
 // WriteBytes implements Storage.WriteBytes().
 func (st *MemoryStorage) WriteBytes(key string, b []byte) error {
+	// Lock storage
+	st.mu.Lock()
+	defer st.mu.Unlock()
+
+	// Check store open
+	if st.st == 1 {
+		return ErrClosed
+	}
+
 	_, ok := st.fs[key]
-	if ok {
+
+	// Check for already exist
+	if ok && !st.ow {
 		return ErrAlreadyExists
 	}
+
+	// Write + unlock
 	st.fs[key] = bytes.Copy(b)
 	return nil
 }
 
 // WriteStream implements Storage.WriteStream().
 func (st *MemoryStorage) WriteStream(key string, r io.Reader) error {
+	// Read all from reader
 	b, err := io.ReadAll(r)
 	if err != nil {
 		return err
 	}
+
+	// Write to storage
 	return st.WriteBytes(key, b)
 }
 
 // Stat implements Storage.Stat().
 func (st *MemoryStorage) Stat(key string) (bool, error) {
+	// Lock storage
+	st.mu.Lock()
+	defer st.mu.Unlock()
+
+	// Check store open
+	if st.st == 1 {
+		return false, ErrClosed
+	}
+
+	// Check for key
 	_, ok := st.fs[key]
 	return ok, nil
 }
 
 // Remove implements Storage.Remove().
 func (st *MemoryStorage) Remove(key string) error {
+	// Lock storage
+	st.mu.Lock()
+	defer st.mu.Unlock()
+
+	// Check store open
+	if st.st == 1 {
+		return ErrClosed
+	}
+
+	// Check for key
 	_, ok := st.fs[key]
 	if !ok {
 		return ErrNotFound
 	}
+
+	// Remove from store
 	delete(st.fs, key)
+
+	return nil
+}
+
+// Close implements Storage.Close().
+func (st *MemoryStorage) Close() error {
+	st.mu.Lock()
+	st.st = 1
+	st.mu.Unlock()
 	return nil
 }
 
 // WalkKeys implements Storage.WalkKeys().
 func (st *MemoryStorage) WalkKeys(opts WalkKeysOptions) error {
+	// Lock storage
+	st.mu.Lock()
+	defer st.mu.Unlock()
+
+	// Check store open
+	if st.st == 1 {
+		return ErrClosed
+	}
+
+	// Walk store keys
 	for key := range st.fs {
 		opts.WalkFn(entry(key))
 	}
+
 	return nil
 }
diff --git a/vendor/codeberg.org/gruf/go-store/storage/storage.go b/vendor/codeberg.org/gruf/go-store/storage/storage.go
index b160267a4..346aff097 100644
--- a/vendor/codeberg.org/gruf/go-store/storage/storage.go
+++ b/vendor/codeberg.org/gruf/go-store/storage/storage.go
@@ -19,9 +19,6 @@ func (e entry) Key() string {
 
 // Storage defines a means of storing and accessing key value pairs
 type Storage interface {
-	// Clean removes unused values and unclutters the storage (e.g. removing empty folders)
-	Clean() error
-
 	// ReadBytes returns the byte value for key in storage
 	ReadBytes(key string) ([]byte, error)
 
@@ -40,6 +37,12 @@ type Storage interface {
 	// Remove attempts to remove the supplied key-value pair from storage
 	Remove(key string) error
 
+	// Close will close the storage, releasing any file locks
+	Close() error
+
+	// Clean removes unused values and unclutters the storage (e.g. removing empty folders)
+	Clean() error
+
 	// WalkKeys walks the keys in the storage
 	WalkKeys(opts WalkKeysOptions) error
 }