Merge pull request #142 from MichaelEischer/atomic-upload

Atomic file upload and directory sync

1 files changed, 48 insertions, 9 deletions

diff --git a/repo/repo.go b/repo/repo.go
index 70e7693..01fe1d8 100644
--- a/repo/repo.go
+++ b/repo/repo.go
@@ -12,6 +12,7 @@ import (
 	"os"
 	"path/filepath"
 	"regexp"
+	"runtime"
 	"strings"
 	"time"
 
@@ -542,7 +543,18 @@ func (h *Handler) saveBlob(w http.ResponseWriter, r *http.Request) {
 	}
 	path := h.getObjectPath(objectType, objectID)
 
-	tf, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_EXCL, h.opt.FileMode)
+	_, err := os.Stat(path)
+	if err == nil {
+		httpDefaultError(w, http.StatusForbidden)
+		return
+	}
+	if !os.IsNotExist(err) {
+		h.internalServerError(w, err)
+		return
+	}
+
+	tmpFn := objectID + ".rest-server-temp"
+	tf, err := ioutil.TempFile(filepath.Dir(path), tmpFn)
 	if os.IsNotExist(err) {
 		// the error is caused by a missing directory, create it and retry
 		mkdirErr := os.MkdirAll(filepath.Dir(path), h.opt.DirMode)
@@ -550,13 +562,9 @@ func (h *Handler) saveBlob(w http.ResponseWriter, r *http.Request) {
 			log.Print(mkdirErr)
 		} else {
 			// try again
-			tf, err = os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_EXCL, h.opt.FileMode)
+			tf, err = ioutil.TempFile(filepath.Dir(path), tmpFn)
 		}
 	}
-	if os.IsExist(err) {
-		httpDefaultError(w, http.StatusForbidden)
-		return
-	}
 	if err != nil {
 		h.internalServerError(w, err)
 		return
@@ -590,7 +598,7 @@ func (h *Handler) saveBlob(w http.ResponseWriter, r *http.Request) {
 
 	if err != nil {
 		_ = tf.Close()
-		_ = os.Remove(path)
+		_ = os.Remove(tf.Name())
 		h.incrementRepoSpaceUsage(-written)
 		if h.opt.Debug {
 			log.Print(err)
@@ -601,22 +609,53 @@ func (h *Handler) saveBlob(w http.ResponseWriter, r *http.Request) {
 
 	if err := tf.Sync(); err != nil {
 		_ = tf.Close()
-		_ = os.Remove(path)
+		_ = os.Remove(tf.Name())
 		h.incrementRepoSpaceUsage(-written)
 		h.internalServerError(w, err)
 		return
 	}
 
 	if err := tf.Close(); err != nil {
-		_ = os.Remove(path)
+		_ = os.Remove(tf.Name())
 		h.incrementRepoSpaceUsage(-written)
 		h.internalServerError(w, err)
 		return
 	}
 
+	if err := os.Rename(tf.Name(), path); err != nil {
+		_ = os.Remove(tf.Name())
+		h.incrementRepoSpaceUsage(-written)
+		h.internalServerError(w, err)
+		return
+	}
+
+	if err := syncDir(filepath.Dir(path)); err != nil {
+		// Don't call os.Remove(path) as this is prone to race conditions with parallel upload retries
+		h.internalServerError(w, err)
+		return
+	}
+
 	h.sendMetric(objectType, BlobWrite, uint64(written))
 }
 
+func syncDir(dirname string) error {
+	if runtime.GOOS == "windows" {
+		// syncing a directory is not possible on windows
+		return nil
+	}
+
+	dir, err := os.Open(dirname)
+	if err != nil {
+		return err
+	}
+	err = dir.Sync()
+	if err != nil {
+		_ = dir.Close()
+		return err
+	}
+	return dir.Close()
+}
+
 // deleteBlob deletes a blob from the repository.
 func (h *Handler) deleteBlob(w http.ResponseWriter, r *http.Request) {
 	if h.opt.Debug {