filer: record the size of a multipart upload saved inside the entry (#11734)

* filer: record the size of a multipart upload saved inside the entry

* filer: guard save-inside multipart append and check Content-MD5

Addressing review feedback:

- Reject a save-inside multipart append when the target is chunk
  backed. The buffered content cannot be attached to existing chunks,
  so previously the request grew FileSize without storing the appended
  bytes, and reads returned zeros for the added range.
- Compare the request Content-MD5 against the buffered content before
  saving, matching the regular multipart path, so a bad digest no
  longer commits the file.
- Extend the size test to check the saved entry MD5, and cover the
  rejected append against chunk-backed and inline targets plus the
  bad-digest rejection.

* filer: return 400 for unsupported append modes

The save-inside append rejection and the existing inline append and offset write rejections are deterministic client errors, but the autoChunk default case mapped them to 500. Give them shared constants and return 400 instead, and assert the status in the tests.

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/server/filer_server_handlers_write_autochunk.go

@@ -58,7 +58,10 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
 			writeJsonError(w, r, util.HttpStatusCancelled, err)
 		case strings.HasSuffix(errStr, "is a file") || strings.HasSuffix(errStr, "already exists"):
 			writeJsonError(w, r, http.StatusConflict, err)
-		case errStr == constants.ErrMsgBadDigest:
+		case errStr == constants.ErrMsgBadDigest,
+			errStr == constants.ErrMsgAppendSaveInsideChunked,
+			errStr == constants.ErrMsgAppendInlineMode,
+			errStr == constants.ErrMsgOffsetWriteInline:
 			writeJsonError(w, r, http.StatusBadRequest, err)
 		default:
 			writeJsonError(w, r, http.StatusInternalServerError, err)
@@ -99,8 +102,17 @@ func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWrite
 	if so.SaveInside {
 		buf := bufPool.Get().(*bytes.Buffer)
 		buf.Reset()
-		buf.ReadFrom(part1)
-		filerResult, replyerr, _ = fs.saveMetaData(ctx, r, fileName, contentType, so, chunkSize, nil, nil, 0, buf.Bytes())
+		if _, err := buf.ReadFrom(part1); err != nil {
+			bufPool.Put(buf)
+			return nil, nil, fmt.Errorf("read input: %w", err)
+		}
+		md5bytes = util.Md5(buf.Bytes())
+		headerMd5 := r.Header.Get("Content-Md5")
+		if headerMd5 != "" && !(util.Base64Encode(md5bytes) == headerMd5 || fmt.Sprintf("%x", md5bytes) == headerMd5) {
+			bufPool.Put(buf)
+			return nil, nil, errors.New(constants.ErrMsgBadDigest)
+		}
+		filerResult, replyerr, _ = fs.saveMetaData(ctx, r, fileName, contentType, so, chunkSize, md5bytes, nil, int64(buf.Len()), buf.Bytes())
 		bufPool.Put(buf)
 		return
 	}
@@ -293,7 +305,11 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 	if entry != nil {
 		inlineCandidate := len(entry.Content) > 0 ||
 			(isAppend && entry.FileSize == 0 && entry.Remote == nil && len(entry.HardLinkId) == 0 && len(entry.GetChunks()) == 0)
-		if isAppend && !so.SaveInside && content != nil && !inlineCandidate {
+		if isAppend && content != nil && !inlineCandidate {
+			if so.SaveInside {
+				replyerr = errors.New(constants.ErrMsgAppendSaveInsideChunked)
+				return
+			}
 			replyerr = fmt.Errorf("inline file changed while preparing append; retry the request")
 			return
 		}
@@ -301,11 +317,11 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 		if inlineCandidate {
 			if isOffsetWrite {
 				// TODO: support inline offset writes separately from append semantics.
-				replyerr = fmt.Errorf("offset write to inline small file is not supported yet")
+				replyerr = errors.New(constants.ErrMsgOffsetWriteInline)
 				return
 			}
 			if !isAppend || so.SaveInside || entry.Remote != nil || len(entry.HardLinkId) != 0 || len(entry.GetChunks()) != 0 {
-				replyerr = fmt.Errorf("append to inline content with this storage mode is not supported")
+				replyerr = errors.New(constants.ErrMsgAppendInlineMode)
 				return
 			}
 			if entry.FileSize > uint64(len(entry.Content)) {

weed/server/filer_server_handlers_write_autochunk_inline_append_test.go

@@ -270,6 +270,164 @@ func TestFilerInlineAppendMultipartPostStaysInline(t *testing.T) {
 	}
 }
 
+func TestFilerSaveInsideMultipartPostRecordsSize(t *testing.T) {
+	ctx := context.Background()
+	content := []byte(`{"identities":[]}`)
+	store := newRenameTestStore()
+	f := newRenameTestFiler(t, store)
+	fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
+
+	var body bytes.Buffer
+	multipartWriter := multipart.NewWriter(&body)
+	part, err := multipartWriter.CreateFormFile("file", "identity.json")
+	if err != nil {
+		t.Fatalf("create multipart file part: %v", err)
+	}
+	if _, err = part.Write(content); err != nil {
+		t.Fatalf("write multipart file part: %v", err)
+	}
+	if err = multipartWriter.Close(); err != nil {
+		t.Fatalf("close multipart body: %v", err)
+	}
+	r := httptest.NewRequest(http.MethodPost, "/", &body)
+	r.Header.Set("Content-Type", multipartWriter.FormDataContentType())
+	result, requestMD5, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, int64(body.Len()), &operation.StorageOption{SaveInside: true})
+	if err != nil {
+		t.Fatalf("POST saveInside: %v", err)
+	}
+	if result == nil || result.Size != int64(len(content)) {
+		t.Fatalf("result = %#v, want size %d", result, len(content))
+	}
+	wantMD5 := md5.Sum(content)
+	if !bytes.Equal(requestMD5, wantMD5[:]) {
+		t.Fatalf("request MD5 = %x, want %x", requestMD5, wantMD5)
+	}
+	saved, err := f.FindEntry(ctx, "/identity.json")
+	if err != nil {
+		t.Fatalf("find saved entry: %v", err)
+	}
+	if !bytes.Equal(saved.Content, content) || saved.FileSize != uint64(len(content)) {
+		t.Fatalf("saved entry: content=%q size=%d, want %q size %d", saved.Content, saved.FileSize, content, len(content))
+	}
+	if !bytes.Equal(saved.Md5, wantMD5[:]) {
+		t.Fatalf("saved MD5 = %x, want %x", saved.Md5, wantMD5)
+	}
+}
+
+func saveInsideMultipartRequest(t *testing.T, target, fileName string, content []byte) *http.Request {
+	t.Helper()
+	var body bytes.Buffer
+	multipartWriter := multipart.NewWriter(&body)
+	part, err := multipartWriter.CreateFormFile("file", fileName)
+	if err != nil {
+		t.Fatalf("create multipart file part: %v", err)
+	}
+	if _, err = part.Write(content); err != nil {
+		t.Fatalf("write multipart file part: %v", err)
+	}
+	if err = multipartWriter.Close(); err != nil {
+		t.Fatalf("close multipart body: %v", err)
+	}
+	r := httptest.NewRequest(http.MethodPost, target, &body)
+	r.Header.Set("Content-Type", multipartWriter.FormDataContentType())
+	return r
+}
+
+func TestFilerSaveInsideMultipartPostRejectsBadDigest(t *testing.T) {
+	ctx := context.Background()
+	content := []byte(`{"identities":[]}`)
+	store := newRenameTestStore()
+	f := newRenameTestFiler(t, store)
+	fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
+
+	r := saveInsideMultipartRequest(t, "/", "identity.json", content)
+	r.Header.Set("Content-Md5", util.Base64Encode([]byte("not the digest")))
+	if _, _, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, r.ContentLength, &operation.StorageOption{SaveInside: true}); err == nil || !strings.Contains(err.Error(), "Content-Md5") {
+		t.Fatalf("POST saveInside error = %v, want bad Content-MD5 error", err)
+	}
+	if _, err := f.FindEntry(ctx, "/identity.json"); !errors.Is(err, filer_pb.ErrNotFound) {
+		t.Fatalf("bad digest still saved the entry: %v", err)
+	}
+}
+
+func TestFilerSaveInsideMultipartAppendRejected(t *testing.T) {
+	ctx := context.Background()
+	const filePath = "/append-target.txt"
+	appendContent := []byte("new")
+
+	t.Run("chunk backed target", func(t *testing.T) {
+		oldChunk := &filer_pb.FileChunk{FileId: "1,00000001", Offset: 0, Size: 4}
+		store := newRenameTestStore()
+		f := newRenameTestFiler(t, store)
+		if err := store.InsertEntry(ctx, &filer.Entry{
+			FullPath: util.FullPath(filePath),
+			Attr: filer.Attr{
+				Mtime:    time.Unix(1, 0),
+				Crtime:   time.Unix(1, 0),
+				FileSize: 4,
+			},
+			Chunks: []*filer_pb.FileChunk{oldChunk},
+		}); err != nil {
+			t.Fatalf("insert chunk-backed entry: %v", err)
+		}
+
+		fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
+		r := saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
+		if _, _, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, r.ContentLength, &operation.StorageOption{SaveInside: true}); err == nil {
+			t.Fatal("saveInside append to chunk-backed entry succeeded")
+		}
+		recorder := httptest.NewRecorder()
+		r = saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
+		fs.autoChunk(ctx, recorder, r, r.ContentLength, &operation.StorageOption{SaveInside: true})
+		if recorder.Code != http.StatusBadRequest {
+			t.Fatalf("rejected saveInside append status = %d, want %d", recorder.Code, http.StatusBadRequest)
+		}
+		unchanged, err := f.FindEntry(ctx, util.FullPath(filePath))
+		if err != nil {
+			t.Fatalf("find entry after rejected append: %v", err)
+		}
+		if unchanged.FileSize != 4 || len(unchanged.Chunks) != 1 || len(unchanged.Content) != 0 {
+			t.Fatalf("rejected append changed entry: size=%d chunks=%d content=%q", unchanged.FileSize, len(unchanged.Chunks), unchanged.Content)
+		}
+	})
+
+	t.Run("inline target", func(t *testing.T) {
+		oldContent := []byte("old-")
+		store := newRenameTestStore()
+		f := newRenameTestFiler(t, store)
+		if err := store.InsertEntry(ctx, &filer.Entry{
+			FullPath: util.FullPath(filePath),
+			Attr: filer.Attr{
+				Mtime:    time.Unix(1, 0),
+				Crtime:   time.Unix(1, 0),
+				FileSize: uint64(len(oldContent)),
+			},
+			Content: append([]byte(nil), oldContent...),
+		}); err != nil {
+			t.Fatalf("insert inline entry: %v", err)
+		}
+
+		fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
+		r := saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
+		if _, _, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, r.ContentLength, &operation.StorageOption{SaveInside: true}); err == nil {
+			t.Fatal("saveInside append to inline entry succeeded")
+		}
+		recorder := httptest.NewRecorder()
+		r = saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
+		fs.autoChunk(ctx, recorder, r, r.ContentLength, &operation.StorageOption{SaveInside: true})
+		if recorder.Code != http.StatusBadRequest {
+			t.Fatalf("rejected saveInside append status = %d, want %d", recorder.Code, http.StatusBadRequest)
+		}
+		unchanged, err := f.FindEntry(ctx, util.FullPath(filePath))
+		if err != nil {
+			t.Fatalf("find entry after rejected append: %v", err)
+		}
+		if !bytes.Equal(unchanged.Content, oldContent) || unchanged.FileSize != uint64(len(oldContent)) {
+			t.Fatalf("rejected append changed entry: content=%q size=%d", unchanged.Content, unchanged.FileSize)
+		}
+	})
+}
+
 func TestFilerInlineOffsetWriteRemainsUnsupported(t *testing.T) {
 	ctx := context.Background()
 	const filePath = "/offset.txt"

weed/util/constants/filer.go

@@ -7,6 +7,10 @@ const (
 	ErrMsgOperationNotPermitted = "operation not permitted"
 	ErrMsgBadDigest             = "The Content-Md5 you specified did not match what we received."
 	ErrMsgEntryNameTooLong      = "entry name too long"
+	// Unsupported write-mode combinations; deterministic client errors.
+	ErrMsgAppendSaveInsideChunked = "append saved inside the entry to a chunked file is not supported"
+	ErrMsgAppendInlineMode        = "append to inline content with this storage mode is not supported"
+	ErrMsgOffsetWriteInline       = "offset write to inline small file is not supported yet"
 	// Suffix appended to a path, e.g. "/buckets/b/foo/bar is a file"
 	ErrMsgIsAFile = " is a file"
 	// Prefix + suffix, e.g. "existing /buckets/b/foo is a directory"