cache: purge file data on notification

backend/cache/cache.go

@@ -499,17 +499,12 @@ func (f *Fs) httpExpireRemote(in rc.Params) (out rc.Params, err error) {
 		return out, nil
 	}
 	// expire the entry
-	co.CacheTs = time.Now().Add(f.fileAge * -1)
-	err = f.cache.AddObject(co)
+	err = f.cache.ExpireObject(co, withData)
 	if err != nil {
 		return out, errors.WithMessage(err, "error expiring file")
 	}
 	// notify vfs too
 	f.notifyChangeUpstream(co.Remote(), fs.EntryObject)
-	if withData {
-		// safe to ignore as the file might not have been open
-		_ = os.RemoveAll(path.Join(f.cache.dataPath, co.abs()))
-	}
 
 	out["status"] = "ok"
 	out["message"] = fmt.Sprintf("cached file cleared: %v", remote)
@@ -537,8 +532,7 @@ func (f *Fs) receiveChangeNotify(forgetPath string, entryType fs.EntryType) {
 		err := f.cache.GetObject(co)
 		if err == nil {
 			// expire the entry
-			co.CacheTs = time.Now().Add(f.fileAge * -1)
-			err = f.cache.AddObject(co)
+			err = f.cache.ExpireObject(co, true)
 			if err != nil {
 				fs.Debugf(forgetPath, "notify: error expiring '%v': %v", co, err)
 			} else {

backend/cache/handle.go

@@ -255,7 +255,7 @@ func (r *Handle) getChunk(chunkStart int64) ([]byte, error) {
 
 	// first chunk will be aligned with the start
 	if offset > 0 {
-		if offset >= int64(len(data)) {
+		if offset > int64(len(data)) {
 			fs.Errorf(r, "unexpected conditions during reading. current position: %v, current chunk position: %v, current chunk size: %v, offset: %v, chunk size: %v, file size: %v",
 				r.offset, chunkStart, len(data), offset, r.cacheFs().chunkSize, r.cachedObject.Size())
 			return nil, io.ErrUnexpectedEOF
@@ -282,8 +282,10 @@ func (r *Handle) Read(p []byte) (n int, err error) {
 	}
 	currentOffset := r.offset
 	buf, err = r.getChunk(currentOffset)
-	if err != nil && len(buf) == 0 {
+	if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 		fs.Errorf(r, "(%v/%v) error (%v) response", currentOffset, r.cachedObject.Size(), err)
+	}
+	if len(buf) == 0 && err != io.ErrUnexpectedEOF {
 		return 0, io.EOF
 	}
 	readSize := copy(p, buf)

backend/cache/storage_persistent.go

@@ -400,6 +400,16 @@ func (b *Persistent) RemoveObject(fp string) error {
 	})
 }
 
+// ExpireObject will flush an Object and all its data if desired
+func (b *Persistent) ExpireObject(co *Object, withData bool) error {
+	co.CacheTs = time.Now().Add(co.CacheFs.fileAge * -1)
+	err := b.AddObject(co)
+	if withData {
+		_ = os.RemoveAll(path.Join(b.dataPath, co.abs()))
+	}
+	return err
+}
+
 // HasEntry confirms the existence of a single entry (dir or object)
 func (b *Persistent) HasEntry(remote string) bool {
 	dir, name := path.Split(remote)