write upload session info atomically

Signed-off-by: Jörn Friedrich Dreyer <[email protected]>

changelog/unreleased/write-upload-sessions-atomically.md

@@ -0,0 +1,5 @@
+Bugfix: Write upload session info atomically
+
+We now use a lock and atomic write on upload session metadata to prevent empty reads. A virus scan event might cause the file to be truncated and then a finished event might try to read the file, just getting an empty string.
+
+https://github.com/cs3org/reva/pull/4850

pkg/storage/utils/decomposedfs/upload/session.go

@@ -26,6 +26,8 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/google/renameio/v2"
+	"github.com/rogpeppe/go-internal/lockedfile"
 	tusd "github.com/tus/tusd/v2/pkg/handler"
 
 	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
@@ -74,7 +76,13 @@ func (s *OcisSession) executantUser() *userpb.User {
 
 // Purge deletes the upload session metadata and written binary data
 func (s *OcisSession) Purge(ctx context.Context) error {
-	if err := os.Remove(sessionPath(s.store.root, s.info.ID)); err != nil {
+	sessionPath := sessionPath(s.store.root, s.info.ID)
+	f, err := lockedfile.OpenFile(sessionPath+".lock", os.O_RDWR|os.O_TRUNC|os.O_CREATE, 0600)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	if err := os.Remove(sessionPath); err != nil {
 		return err
 	}
 	if err := os.Remove(s.binPath()); err != nil {
@@ -93,10 +101,12 @@ func (s *OcisSession) TouchBin() error {
 }
 
 // Persist writes the upload session metadata to disk
+// events can update the scan outcome and the finished event might read an empty file because of race conditions
+// so we need to lock the file while writing and use atomic writes
 func (s *OcisSession) Persist(ctx context.Context) error {
-	uploadPath := sessionPath(s.store.root, s.info.ID)
+	sessionPath := sessionPath(s.store.root, s.info.ID)
 	// create folder structure (if needed)
-	if err := os.MkdirAll(filepath.Dir(uploadPath), 0700); err != nil {
+	if err := os.MkdirAll(filepath.Dir(sessionPath), 0700); err != nil {
 		return err
 	}
 
@@ -105,8 +115,12 @@ func (s *OcisSession) Persist(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-
-	return os.WriteFile(uploadPath, d, 0600)
+	f, err := lockedfile.OpenFile(sessionPath+".lock", os.O_RDWR|os.O_TRUNC|os.O_CREATE, 0600)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	return renameio.WriteFile(sessionPath, d, 0600)
 }
 
 // ToFileInfo returns tus compatible FileInfo so the tus handler can access the upload offset

pkg/storage/utils/decomposedfs/upload/store.go

@@ -119,7 +119,7 @@ func (store OcisStore) List(ctx context.Context) ([]*OcisSession, error) {
 
 // Get returns the upload session for the given upload id
 func (store OcisStore) Get(ctx context.Context, id string) (*OcisSession, error) {
-	sessionPath := filepath.Join(store.root, "uploads", id+".info")
+	sessionPath := sessionPath(store.root, id)
 	match := _idRegexp.FindStringSubmatch(sessionPath)
 	if match == nil || len(match) < 2 {
 		return nil, fmt.Errorf("invalid upload path")
@@ -129,6 +129,15 @@ func (store OcisStore) Get(ctx context.Context, id string) (*OcisSession, error)
 		store: store,
 		info:  tusd.FileInfo{},
 	}
+	lock, err := lockedfile.Open(sessionPath + ".lock")
+	if err != nil {
+		if errors.Is(err, iofs.ErrNotExist) {
+			// Interpret os.ErrNotExist as 404 Not Found
+			err = tusd.ErrNotFound
+		}
+		return nil, err
+	}
+	defer lock.Close()
 	data, err := os.ReadFile(sessionPath)
 	if err != nil {
 		if errors.Is(err, iofs.ErrNotExist) {
@@ -137,6 +146,8 @@ func (store OcisStore) Get(ctx context.Context, id string) (*OcisSession, error)
 		}
 		return nil, err
 	}
+	lock.Close() // release lock asap
+
 	if err := json.Unmarshal(data, &session.info); err != nil {
 		return nil, err
 	}