fix(multipart): resume interrupted uploads without file hashes

internal/multipart/session.go

@@ -79,6 +79,7 @@ type Session struct {
 	Mimetype  string
 	Modified  time.Time
 	Hashes    map[*utils.HashType]string
+	Identity  string // opaque client fingerprint, see qualifiesResume
 	Creator   *model.User
 
 	mu       sync.Mutex
@@ -160,6 +161,7 @@ type InitReq struct {
 	Mimetype  string
 	Modified  time.Time
 	Hashes    map[*utils.HashType]string
+	Identity  string
 }
 
 // Manager owns all live sessions. Sessions are in-memory only (aligned with
@@ -193,11 +195,61 @@ func sessionKey(userID uint, path string, size int64) string {
 	return fmt.Sprintf("%d|%s|%d", userID, path, size)
 }
 
+// qualifiesResume reports whether the client proved it is re-uploading the
+// file a live session already holds data for. Path+size alone is NOT enough —
+// buffered chunks of a different same-sized file would be silently mixed into
+// the result. Note this guards against accidents, not against an adversary:
+// sessions are per-user (see sessionKey and Manager.get), so the worst a
+// client can do is splice its own upload, which it could equally do by
+// uploading the spliced file directly.
+func qualifiesResume(s *Session, req InitReq) bool {
+	// unlike the file hashes, the identity fingerprint is sent regardless of
+	// whether rapid upload was requested. It is opaque and may cover only part
+	// of the content (the web client hashes the first and last MiB), so a hash
+	// both sides sent must still agree
+	if s.Identity != "" || req.Identity != "" {
+		return s.Identity != "" && s.Identity == req.Identity && !hashesConflict(s.Hashes, req.Hashes)
+	}
+	return hashesQualifyResume(s.Hashes, req.Hashes)
+}
+
+// resumes reports whether an init for req picks up s, in state st, instead of
+// superseding it.
+func resumes(s *Session, st State, req InitReq) bool {
+	return st == StateFailedRetriable || (st == StateReceiving && qualifiesResume(s, req))
+}
+
+// ResumesSameFile reports whether Init(req) would resume a session that req
+// proves to be the same file by identity or hashes. That upload owns what sits
+// at the destination (its own partial output on in-place drivers, or a file it
+// was allowed to replace), so "file exists" must not stop the resume. Picking
+// up a failed attempt is no proof: any init for the same path and size does.
+func (m *Manager) ResumesSameFile(req InitReq) bool {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	s, ok := m.byID[m.byKey[sessionKey(req.User.ID, req.Path, req.Size)]]
+	if !ok {
+		return false
+	}
+	s.mu.Lock()
+	st := s.state
+	s.mu.Unlock()
+	return resumes(s, st, req) && qualifiesResume(s, req)
+}
+
+// hashesConflict reports whether a hash type sent on both inits differs.
+func hashesConflict(old, new map[*utils.HashType]string) bool {
+	for t, ov := range old {
+		if nv, ok := new[t]; ok && nv != ov {
+			return true
+		}
+	}
+	return false
+}
+
 // hashesQualifyResume reports whether two hash sets prove the client is
 // re-uploading the same file: they must share at least one hash type and
-// agree on every shared one. Path+size alone is NOT enough to resume into a
-// receiving session — buffered chunks of a different same-sized file would be
-// silently mixed into the result.
+// agree on every shared one.
 func hashesQualifyResume(old, new map[*utils.HashType]string) bool {
 	shared := false
 	for t, ov := range old {
@@ -317,9 +369,8 @@ func (m *Manager) Init(req InitReq) (SessionSnapshot, bool, error) {
 			// failed_retriable resumes unconditionally: nothing of the old
 			// attempt survives except CRCs, and the re-fill CRC check catches
 			// a changed file. A receiving session still holds data, so it only
-			// resumes when hashes prove it is the same file.
-			if st == StateFailedRetriable ||
-				(st == StateReceiving && hashesQualifyResume(s.Hashes, req.Hashes)) {
+			// resumes when the client proves it is the same file.
+			if resumes(s, st, req) {
 				m.mu.Unlock()
 				s.touch()
 				return s.Snapshot(), true, nil
@@ -344,6 +395,7 @@ func (m *Manager) Init(req InitReq) (SessionSnapshot, bool, error) {
 		Mimetype:  req.Mimetype,
 		Modified:  req.Modified,
 		Hashes:    req.Hashes,
+		Identity:  req.Identity,
 		Creator:   req.User,
 		state:     StateReceiving,
 	}

internal/multipart/session_test.go

@@ -667,3 +667,170 @@ func TestInitRejectsBadSize(t *testing.T) {
 		t.Fatal("Init with size 0: expected error")
 	}
 }
+
+func withIdentity(req InitReq, identity string) InitReq {
+	req.Identity = identity
+	return req
+}
+
+func withMD5(req InitReq, md5 string) InitReq {
+	if md5 != "" {
+		req.Hashes = map[*utils.HashType]string{utils.MD5: md5}
+	}
+	return req
+}
+
+// The identity fingerprint is what lets a reload resume: it is sent whether or
+// not the user asked for a rapid upload, unlike the file hashes.
+func TestResumeReceivingByIdentity(t *testing.T) {
+	cases := []struct {
+		name          string
+		first, second string
+		// MD5 sent with the first and the second init; empty sends none
+		firstMD5, secondMD5 string
+		wantSameResume      bool
+	}{
+		{name: "matching identity resumes", first: "fp-a", second: "fp-a", wantSameResume: true},
+		{name: "different identity starts fresh", first: "fp-a", second: "fp-b"},
+		{name: "identity dropped starts fresh", first: "fp-a", second: ""},
+		{name: "identity added starts fresh", first: "", second: "fp-b"},
+		{name: "no identity falls back to hashes", firstMD5: "same-file", secondMD5: "same-file", wantSameResume: true},
+		// the identity covers only part of the content: hashes both sides
+		// sent must still agree, or a file changed in the middle is spliced
+		{name: "matching identity with conflicting hashes starts fresh", first: "fp-a", second: "fp-a", firstMD5: "file-v1", secondMD5: "file-v2"},
+		{name: "matching identity with matching hashes resumes", first: "fp-a", second: "fp-a", firstMD5: "same-file", secondMD5: "same-file", wantSameResume: true},
+		{name: "no identity and no hashes never resumes", first: "", second: ""},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			m := setupSessionTest(t)
+			const chunkSize = 1024
+			totalSize := int64(4 * chunkSize)
+			data := genData(totalSize)
+			user := testUser()
+
+			started := make(chan struct{})
+			var once sync.Once // a superseding Init starts a second pipeline
+			stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error {
+				defer fs.Close()
+				once.Do(func() { close(started) })
+				<-ctx.Done()
+				return ctx.Err()
+			})
+
+			base := initReq(user, totalSize, chunkSize)
+			snap, _, err := m.Init(withMD5(withIdentity(base, tc.first), tc.firstMD5))
+			if err != nil {
+				t.Fatalf("Init: %v", err)
+			}
+			<-started
+			sendChunk(t, m, user, snap.ID, data, 0, chunkSize)
+
+			again, resumed, err := m.Init(withMD5(withIdentity(base, tc.second), tc.secondMD5))
+			if err != nil {
+				t.Fatalf("re-Init: %v", err)
+			}
+			if tc.wantSameResume {
+				if !resumed || again.ID != snap.ID {
+					t.Fatalf("re-Init = (id=%s, resumed=%v), want the same session resumed", again.ID, resumed)
+				}
+				if len(again.Received) == 0 {
+					t.Fatal("resumed session reports nothing received, so the client re-uploads everything")
+				}
+			} else if resumed || again.ID == snap.ID {
+				t.Fatalf("re-Init = (id=%s, resumed=%v), want a fresh session", again.ID, resumed)
+			}
+		})
+	}
+}
+
+// An interrupted upload must be able to resume even though an in-place driver
+// has already written its partial output to the destination, but only an init
+// that proves it is the same file may skip the "file exists" check: a failed
+// attempt is picked up by any init for the same path and size.
+func TestResumesSameFile(t *testing.T) {
+	const md5A = "0cc175b9c0f1b6a831c399e269772661"
+	cases := []struct {
+		name          string
+		failed        bool   // the first attempt fails after one chunk
+		firstIdentity string // sent with the first init
+		firstMD5      string
+		againIdentity string // sent with the second init
+		againMD5      string
+		want          bool
+	}{
+		{name: "receiving, same identity", firstIdentity: "fp-a", againIdentity: "fp-a", want: true},
+		{name: "receiving, different identity", firstIdentity: "fp-a", againIdentity: "fp-b"},
+		{name: "receiving, identity dropped", firstIdentity: "fp-a"},
+		{name: "receiving, same hash", firstMD5: md5A, againMD5: md5A, want: true},
+		{name: "failed, same identity", failed: true, firstIdentity: "fp-a", againIdentity: "fp-a", want: true},
+		{name: "failed, different identity", failed: true, firstIdentity: "fp-a", againIdentity: "fp-b"},
+		{name: "failed, nothing proves the file", failed: true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			m := setupSessionTest(t)
+			const chunkSize = 1024
+			totalSize := int64(4 * chunkSize)
+			data := genData(totalSize)
+			user := testUser()
+
+			started := make(chan struct{})
+			var once sync.Once
+			stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error {
+				defer fs.Close()
+				once.Do(func() { close(started) })
+				if tc.failed {
+					if _, err := io.ReadFull(fs, make([]byte, chunkSize)); err != nil {
+						return err
+					}
+					return errors.New("transient storage hiccup")
+				}
+				<-ctx.Done()
+				return ctx.Err()
+			})
+
+			first := withMD5(withIdentity(initReq(user, totalSize, chunkSize), tc.firstIdentity), tc.firstMD5)
+			snap, _, err := m.Init(first)
+			if err != nil {
+				t.Fatalf("Init: %v", err)
+			}
+			<-started
+			sendChunk(t, m, user, snap.ID, data, 0, chunkSize)
+			if tc.failed {
+				waitState(t, m, user, snap.ID, StateFailedRetriable)
+			}
+
+			again := withMD5(withIdentity(initReq(user, totalSize, chunkSize), tc.againIdentity), tc.againMD5)
+			if got := m.ResumesSameFile(again); got != tc.want {
+				t.Fatalf("ResumesSameFile = %v, want %v", got, tc.want)
+			}
+		})
+	}
+	t.Run("no session", func(t *testing.T) {
+		m := setupSessionTest(t)
+		if m.ResumesSameFile(withIdentity(initReq(testUser(), 4096, 1024), "fp-a")) {
+			t.Fatal("ResumesSameFile = true without any session")
+		}
+	})
+	t.Run("failed permanently", func(t *testing.T) {
+		m := setupSessionTest(t)
+		user := testUser()
+		stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error {
+			defer fs.Close()
+			return fmt.Errorf("denied: %w", errs.PermissionDenied)
+		})
+
+		req := withIdentity(initReq(user, 4096, 1024), "fp-a")
+		snap, _, err := m.Init(req)
+		if err != nil {
+			t.Fatalf("Init: %v", err)
+		}
+		waitState(t, m, user, snap.ID, StateFailedPermanent)
+		// Init supersedes such a session with a fresh upload, which must
+		// still meet "file exists"
+		if m.ResumesSameFile(req) {
+			t.Fatal("ResumesSameFile = true for a session Init would supersede")
+		}
+	})
+}

server/handles/multipart.go

@@ -19,7 +19,11 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
-const multipartMinChunkSize = int64(1) << 20 // 1MB
+const (
+	multipartMinChunkSize = int64(1) << 20 // 1MB
+	// X-File-Identity is stored per session, so bound what a client can pin
+	multipartMaxIdentityLen = 256
+)
 
 // multipartChunkSize resolves the effective chunk size. The admin setting is
 // the ceiling: a client may suggest a smaller chunk via X-Chunk-Size but never
@@ -80,12 +84,6 @@ func MultipartInit(c *gin.Context) {
 		}
 	}
 	overwrite := c.GetHeader("Overwrite") != "false"
-	if !overwrite {
-		if res, _ := fs.Get(c.Request.Context(), path, &fs.GetArgs{NoLog: true}); res != nil {
-			common.ErrorStrResp(c, "file exists", 403)
-			return
-		}
-	}
 	dir, name := stdpath.Split(path)
 	if shouldIgnoreSystemFile(name) {
 		common.ErrorStrResp(c, errs.IgnoredSystemFile.Error(), 403)
@@ -115,15 +113,32 @@ func MultipartInit(c *gin.Context) {
 	if len(mimetype) == 0 {
 		mimetype = utils.GetMimeType(name)
 	}
-	snap, resumed, err := multipart.DefaultManager.Init(multipart.InitReq{
+	// opaque to the server: it only ever compares it with the value the
+	// session was created with, to decide whether resuming is safe
+	identity := c.GetHeader("X-File-Identity")
+	if len(identity) > multipartMaxIdentityLen {
+		common.ErrorStrResp(c, "X-File-Identity is too long", 400)
+		return
+	}
+	req := multipart.InitReq{
 		User:      user,
 		Path:      path,
 		Size:      size,
 		ChunkSize: multipartChunkSize(requestedChunkSize),
 		Mimetype:  mimetype,
 		Modified:  getLastModified(c),
 		Hashes:    h,
-	})
+		Identity:  identity,
+	}
+	// an interrupted upload may resume even though in-place drivers have
+	// already written its partial output to the destination
+	if !overwrite && !multipart.DefaultManager.ResumesSameFile(req) {
+		if res, _ := fs.Get(c.Request.Context(), path, &fs.GetArgs{NoLog: true}); res != nil {
+			common.ErrorStrResp(c, "file exists", 403)
+			return
+		}
+	}
+	snap, resumed, err := multipart.DefaultManager.Init(req)
 	if err != nil {
 		common.ErrorResp(c, err, 500)
 		return