added disk quota

pkg/fs/fs.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
+	"golang.org/x/exp/maps"
 )
 
 type FS struct {
@@ -29,12 +30,13 @@ type FS struct {
 	mdf     *os.File
 	jenc    *json.Encoder
 	dc      *DirCache
-	mm      map[string]*Metadata
-	prem    map[string]func()
+	q       *Quota
+	md      MetadataMap
+	pm      map[string]func()
 	cancel  func()
 }
 
-func NewFS(src, dst, metadata string, log logr.Logger) (fs *FS, err error) {
+func NewFS(max int64, src, dst, metadata string, log logr.Logger) (fs *FS, err error) {
 	if !filepath.IsAbs(src) {
 		return nil, errors.New("src path is not absolute")
 	}
@@ -58,15 +60,20 @@ func NewFS(src, dst, metadata string, log logr.Logger) (fs *FS, err error) {
 		dst:     dst,
 		mdf:     mdf,
 		dc:      NewDirCache(),
-		mm:      make(map[string]*Metadata),
-		prem:    make(map[string]func()),
+		md:      make(MetadataMap),
+		pm:      make(map[string]func()),
 		jenc:    json.NewEncoder(mdf),
 	}
-	err = json.NewDecoder(mdf).Decode(&fs.mm)
+	fs.q, err = NewQuota(max, fs, fs.log.WithName("quota"))
+	if err != nil {
+		return
+	}
+	err = json.NewDecoder(mdf).Decode(&fs.md)
 	if err == io.EOF {
 		err = nil
 	}
 	fs.initMetadata()
+	fs.q.Init()
 	ctx, cancel := context.WithCancel(context.Background())
 	fs.cancel = cancel
 	go fs.flusher(ctx)
@@ -75,26 +82,25 @@ func NewFS(src, dst, metadata string, log logr.Logger) (fs *FS, err error) {
 
 func (fs *FS) initMetadata() {
 	log := fs.log
-	for k, v := range fs.mm {
+	maps.DeleteFunc(fs.md, func(k string, v *Metadata) bool {
 		_, err := fs.statDst(k)
 		if errors.Is(err, os.ErrNotExist) {
 			log.V(2).Info("removing not existing file from metadata", "file", k)
-			delete(fs.mm, k)
-			continue
+			return true
 		}
 		if len(v.Chunks) == 0 {
 			log.V(2).Info("removing empty file", "file", k)
 			err := fs.RemoveDst(k)
 			if err != nil {
 				log.Error(err, "error removing empty file", "file", k)
-				continue
+				return false
 			}
-			delete(fs.mm, k)
-			continue
+			return true
 		}
 		v.fs = fs
 		v.name = k
-	}
+		return false
+	})
 }
 
 func (fs *FS) Stat(name string) (fi stdfs.FileInfo, err error) {
@@ -126,7 +132,7 @@ func (fs *FS) statDst(name string) (fi stdfs.FileInfo, err error) {
 }
 
 func (fs *FS) CancelPreload(name string) {
-	if cancel, ok := fs.prem[name]; ok {
+	if cancel, ok := fs.pm[name]; ok {
 		cancel()
 	}
 }
@@ -146,9 +152,9 @@ func (fs *FS) Preload(name string) {
 		return
 	}
 	ctx, cancel := context.WithCancel(context.Background())
-	fs.prem[name] = cancel
+	fs.pm[name] = cancel
 	unlock := func() {
-		delete(fs.prem, name)
+		delete(fs.pm, name)
 		f.md.UnlockPreload()
 	}
 	go f.Preload(ctx, unlock)
@@ -199,15 +205,17 @@ func (fs *FS) Open(name string) (f http.File, err error) {
 			offline = true
 			log.V(2).Info("dir offline mode", "path", dp)
 		}
-		return &Dir{f: sf, dc: fs.dc}, err
+		f = &Dir{f: sf, dc: fs.dc}
+		return
 	}
 	md := fs.metadata(name, sfi.Size())
-	return &File{
+	f = &File{
 		log:     log,
 		f:       sf,
 		md:      md,
 		offline: offline,
-	}, nil
+	}
+	return
 }
 
 func (fs *FS) openCacheFile(name string, size int64) (df *os.File, err error) {
@@ -248,19 +256,29 @@ func (fs *FS) openCacheFile(name string, size int64) (df *os.File, err error) {
 
 func (fs *FS) metadata(name string, size int64) (md *Metadata) {
 	fs.mu.Lock()
+	fs.log.Info("fs.mu.Lock()")
 	defer fs.mu.Unlock()
-	if md, ok := fs.mm[name]; ok {
+	defer fs.log.Info("fs.mu.Unlock()")
+	if md, ok := fs.md[name]; ok {
+		if size > 0 && md.Size != size {
+			log := fs.log.WithValues("file", name)
+			log.V(2).Info("source file changed: deleting cache file", "src", size, "dst", md.Size)
+			err := md.Delete()
+			if err != nil {
+				log.Error(err, "error deleting cache file")
+			}
+		}
 		return md
 	}
 	md = &Metadata{Size: size, fs: fs, name: name}
-	fs.mm[name] = md
+	fs.md[name] = md
 	return
 }
 
 func (fs *FS) CacheStatus(name string) int {
 	fs.mu.RLock()
 	defer fs.mu.RUnlock()
-	if md, ok := fs.mm[name]; ok {
+	if md, ok := fs.md[name]; ok {
 		return int(math.Round(float64(md.Chunks.Size()) / float64(md.Size) * 100))
 	}
 	return -1
@@ -281,12 +299,11 @@ func (fs *FS) flushMetadata() {
 	log := fs.log
 	fs.mu.Lock()
 	defer fs.mu.Unlock()
-	mm := make(map[string]*Metadata)
-	for k, v := range fs.mm {
-		if v.f != nil || len(v.Chunks) > 0 {
-			mm[k] = v
-		}
-	}
+	log.V(2).Info("flushing metadata to disk")
+	md := maps.Clone(fs.md)
+	maps.DeleteFunc(md, func(_ string, v *Metadata) bool {
+		return len(v.Chunks) == 0
+	})
 	_, err := fs.mdf.Seek(0, io.SeekStart)
 	if err != nil {
 		log.Error(err, "failure seeking metadata file")
@@ -297,7 +314,7 @@ func (fs *FS) flushMetadata() {
 		log.Error(err, "failure truncating metadata file")
 		return
 	}
-	err = fs.jenc.Encode(mm)
+	err = fs.jenc.Encode(md)
 	if err != nil {
 		log.Error(err, "failure flushing metadata file")
 	}
@@ -305,7 +322,10 @@ func (fs *FS) flushMetadata() {
 
 func (fs *FS) Close() {
 	fs.cancel()
-	for _, md := range fs.mm {
+	for _, cancel := range fs.pm {
+		cancel()
+	}
+	for _, md := range fs.md {
 		md.f.Sync()
 		md.f.Close()
 	}