From dd8247e5780651894faa99a3a60a856f0bc2e08b Mon Sep 17 00:00:00 2001 From: MadDogOwner Date: Tue, 21 Jul 2026 00:22:24 +0800 Subject: [PATCH] feat(server/s3): support multipart upload - Implement gofakes3 MultipartBackend (Create/UploadPart/Complete/Abort) on s3Backend so multipart parts stream to local temp files instead of being buffered in memory - Track in-progress uploads via a new uploads sync.Map on s3Backend - Refactor the PutObject body into a reusable putStream helper shared with the multipart Complete path - Validate part ordering, etags and existence, reject short reads, and return S3-style multipart etags - Add end-to-end multipart tests against a real Local driver Co-authored-by: Codex <267193182+codex@users.noreply.github.com> --- server/s3/backend.go | 39 ++++-- server/s3/multipart.go | 272 ++++++++++++++++++++++++++++++++++++ server/s3/multipart_test.go | 245 ++++++++++++++++++++++++++++++++ 3 files changed, 541 insertions(+), 15 deletions(-) create mode 100644 server/s3/multipart.go create mode 100644 server/s3/multipart_test.go diff --git a/server/s3/backend.go b/server/s3/backend.go index 6125a5a4b..108c2846d 100644 --- a/server/s3/backend.go +++ b/server/s3/backend.go @@ -33,17 +33,21 @@ var ( timeFormat = "Mon, 2 Jan 2006 15:04:05 GMT" ) -// s3Backend implements the gofacess3.Backend interface to make an S3 -// backend for gofakes3 +// s3Backend implements the gofakes3.Backend interface to make an S3 +// backend for gofakes3. It also implements gofakes3.MultipartBackend so that +// multipart uploads are streamed to local temp files part-by-part and +// assembled into storage on completion, instead of being buffered in memory. type s3Backend struct { meta *sync.Map listDir func(context.Context, string) ([]model.Obj, error) + uploads *sync.Map // map[gofakes3.UploadID]*multipartState } // newBackend creates a new SimpleBucketBackend. func newBackend() gofakes3.Backend { return &s3Backend{ meta: new(sync.Map), + uploads: new(sync.Map), listDir: getDirEntries, } } @@ -233,9 +237,20 @@ func (b *s3Backend) PutObject( meta map[string]string, input io.Reader, size int64, ) (result gofakes3.PutObjectResult, err error) { + return result, b.putStream(ctx, bucketName, objectName, meta, input, size) +} + +// putStream stores the given object into the underlying storage. It is shared +// by PutObject and the multipart-upload Complete step so both paths apply the +// same directory creation, metadata and ignore rules. +func (b *s3Backend) putStream( + ctx context.Context, bucketName, objectName string, + meta map[string]string, + input io.Reader, size int64, +) error { bucket, err := getBucketByName(bucketName) if err != nil { - return result, err + return err } bucketPath := bucket.Path @@ -261,15 +276,15 @@ func (b *s3Backend) PutObject( log.Debugf("reqPath: %s not found and objectName contains /, need to makeDir", reqPath) err = fs.MakeDir(ctx, reqPath) if err != nil { - return result, errors.WithMessagef(err, "failed to makeDir, reqPath: %s", reqPath) + return errors.WithMessagef(err, "failed to makeDir, reqPath: %s", reqPath) } } else { - return result, gofakes3.KeyNotFound(objectName) + return gofakes3.KeyNotFound(objectName) } } if isDir { - return result, nil + return nil } var ti time.Time @@ -295,7 +310,7 @@ func (b *s3Backend) PutObject( } // Check if system file should be ignored if setting.GetBool(conf.IgnoreSystemFiles) && utils.IsSystemFile(obj.Name) { - return result, errs.IgnoredSystemFile + return errs.IgnoredSystemFile } stream := &stream.FileStream{ Obj: &obj, @@ -305,18 +320,12 @@ func (b *s3Backend) PutObject( err = fs.PutDirectly(ctx, reqPath, stream) if err != nil { - return result, err + return err } - // if err := stream.Close(); err != nil { - // // remove file when close error occurred (FsPutErr) - // _ = fs.Remove(ctx, fp) - // return result, err - // } - b.meta.Store(fp, meta) - return result, nil + return nil } // DeleteMulti deletes multiple objects in a single request. diff --git a/server/s3/multipart.go b/server/s3/multipart.go new file mode 100644 index 000000000..38bfe4de1 --- /dev/null +++ b/server/s3/multipart.go @@ -0,0 +1,272 @@ +// Credits: https://pkg.go.dev/github.com/rclone/rclone@v1.65.2/cmd/serve/s3 +// Package s3 implements a fake s3 server for openlist +package s3 + +import ( + "context" + "crypto/md5" + "encoding/hex" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/OpenListTeam/OpenList/v4/internal/conf" + "github.com/OpenListTeam/OpenList/v4/pkg/utils" + "github.com/OpenListTeam/gofakes3" + log "github.com/sirupsen/logrus" +) + +// Compile-time assertions that s3Backend implements both the base Backend and +// the optional MultipartBackend interface from gofakes3. +var ( + _ gofakes3.Backend = (*s3Backend)(nil) + _ gofakes3.MultipartBackend = (*s3Backend)(nil) +) + +// multipartPart records a single uploaded part on disk. +type multipartPart struct { + path string + size int64 + md5hex string // unquoted lowercase hex + updated time.Time +} + +// multipartState tracks one in-progress multipart upload. +// +// gofakes3 does not serialize multipart operations for a given UploadID, so +// the parts map is guarded by mu. Each part is written to its own file inside +// dir, so concurrent UploadPart calls for different part numbers are safe. +type multipartState struct { + bucket string + object string + meta map[string]string + dir string + created time.Time + + mu sync.Mutex + parts map[int]*multipartPart +} + +// CreateMultipartUpload begins a new multipart upload. Parts are streamed to +// local temp files so that large uploads do not need to be buffered in memory. +// +// It implements gofakes3.MultipartBackend. +func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucket, object string, meta map[string]string) (gofakes3.UploadID, error) { + if _, err := getBucketByName(bucket); err != nil { + return "", err + } + + tempDir := conf.Conf.TempDir + if tempDir == "" { + tempDir = os.TempDir() + } + dir, err := os.MkdirTemp(tempDir, "s3-multipart-*") + if err != nil { + return "", fmt.Errorf("create multipart upload dir: %w", err) + } + + uploadID := gofakes3.UploadID(strings.ReplaceAll(uuid.NewString(), "-", "")) + state := &multipartState{ + bucket: bucket, + object: object, + meta: meta, + dir: dir, + created: time.Now(), + parts: map[int]*multipartPart{}, + } + + b.uploads.Store(uploadID, state) + log.Debugf("s3 multipart: created upload %s for %s/%s", uploadID, bucket, object) + return uploadID, nil +} + +// UploadPart writes a single part to disk and returns its (quoted) MD5 etag. +// +// It implements gofakes3.MultipartBackend. The body must contain exactly +// contentLength bytes; a short read is reported as ErrIncompleteBody so a +// truncated client request is never silently stored. +func (b *s3Backend) UploadPart(ctx context.Context, bucket, object string, uploadID gofakes3.UploadID, partNumber int, contentLength int64, body io.Reader) (string, error) { + if partNumber <= 0 || partNumber > gofakes3.MaxUploadPartNumber { + return "", gofakes3.ErrInvalidPart + } + + val, ok := b.uploads.Load(uploadID) + if !ok { + return "", gofakes3.ErrNoSuchUpload + } + state := val.(*multipartState) + + partPath := filepath.Join(state.dir, fmt.Sprintf("part-%05d", partNumber)) + f, err := os.Create(partPath) + if err != nil { + return "", fmt.Errorf("create part file: %w", err) + } + // Remove a half-written file on any failure path. + partFailed := true + defer func() { + if partFailed { + _ = f.Close() + _ = os.Remove(partPath) + } + }() + + hash := md5.New() + // io.TeeReader feeds the hasher while the part is streamed to disk, so the + // etag costs no extra pass over the data. + n, err := utils.CopyWithBuffer(io.MultiWriter(f, hash), io.LimitReader(body, contentLength)) + if err != nil { + return "", fmt.Errorf("write part %d: %w", partNumber, err) + } + if err := f.Close(); err != nil { + return "", fmt.Errorf("close part %d: %w", partNumber, err) + } + if n != contentLength { + // The client under-delivered (truncated/aborted request). gofakes3 does + // not validate this for streaming backends, so we must. + return "", gofakes3.ErrIncompleteBody + } + + md5hex := hex.EncodeToString(hash.Sum(nil)) + etag := fmt.Sprintf("%q", md5hex) + + state.mu.Lock() + if old := state.parts[partNumber]; old != nil && old.path != partPath { + _ = os.Remove(old.path) + } + state.parts[partNumber] = &multipartPart{ + path: partPath, + size: n, + md5hex: md5hex, + updated: time.Now(), + } + state.mu.Unlock() + + partFailed = false + log.Debugf("s3 multipart: stored part %d for %s (%d bytes)", partNumber, uploadID, n) + return etag, nil +} + +// CompleteMultipartUpload assembles the uploaded parts in ascending part-number +// order and streams the result into storage via the shared putStream path. +// +// It implements gofakes3.MultipartBackend. Part ordering and etags are +// validated against the parts actually received. +func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucket, object string, uploadID gofakes3.UploadID, input *gofakes3.CompleteMultipartUploadRequest) (gofakes3.VersionID, string, error) { + val, ok := b.uploads.Load(uploadID) + if !ok { + return "", "", gofakes3.ErrNoSuchUpload + } + state := val.(*multipartState) + + if input == nil || len(input.Parts) == 0 { + return "", "", gofakes3.ErrorMessagef(gofakes3.ErrMalformedXML, "complete multipart upload has no parts") + } + // S3 requires the parts in a CompleteMultipartUpload request to be listed + // in ascending part-number order. + for i := 1; i < len(input.Parts); i++ { + if input.Parts[i].PartNumber <= input.Parts[i-1].PartNumber { + return "", "", gofakes3.ErrInvalidPartOrder + } + } + + // Validate every requested part exists with a matching etag, and collect + // them in the order requested by the client (which is sorted ascending). + state.mu.Lock() + ordered := make([]*multipartPart, 0, len(input.Parts)) + var concat []byte + for _, p := range input.Parts { + stored := state.parts[p.PartNumber] + if stored == nil { + state.mu.Unlock() + return "", "", gofakes3.ErrorMessagef(gofakes3.ErrInvalidPart, "unexpected part number %d in complete request", p.PartNumber) + } + if strings.Trim(p.ETag, "\"") != stored.md5hex { + state.mu.Unlock() + return "", "", gofakes3.ErrorMessagef(gofakes3.ErrInvalidPart, "unexpected part etag for number %d in complete request", p.PartNumber) + } + ordered = append(ordered, stored) + // S3 multipart etag = hex(md5(concat(part_md5_digests)))-N + concat = append(concat, stored.md5Bytes()...) + } + // Hold the lock until the part files are opened so an abort racing with + // complete cannot delete them out from under us. + readers := make([]io.Reader, 0, len(ordered)) + closers := make([]io.Closer, 0, len(ordered)) + var total int64 + for _, part := range ordered { + f, err := os.Open(part.path) + if err != nil { + for _, c := range closers { + _ = c.Close() + } + state.mu.Unlock() + return "", "", fmt.Errorf("open part %s: %w", part.path, err) + } + readers = append(readers, f) + closers = append(closers, f) + total += part.size + } + state.mu.Unlock() + + combined := utils.NewReadCloser(io.MultiReader(readers...), func() error { + var firstErr error + for _, c := range closers { + if err := c.Close(); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr + }) + + err := b.putStream(ctx, bucket, object, state.meta, combined, total) + _ = combined.Close() + if err != nil { + // Leave the upload in place so the client may retry completion, per + // the gofakes3 MultipartBackend contract. + return "", "", err + } + + // Success: drop bookkeeping and clean up part files. + b.removeUpload(uploadID) + + sum := md5.Sum(concat) + etag := fmt.Sprintf("%q", fmt.Sprintf("%s-%d", hex.EncodeToString(sum[:]), len(ordered))) + log.Debugf("s3 multipart: completed upload %s -> %s/%s (%d bytes)", uploadID, bucket, object, total) + return "", etag, nil +} + +// AbortMultipartUpload discards an in-progress upload and its parts. +// +// It implements gofakes3.MultipartBackend and is idempotent: aborting an +// unknown upload succeeds so retries do not fail. +func (b *s3Backend) AbortMultipartUpload(ctx context.Context, bucket, object string, uploadID gofakes3.UploadID) error { + b.removeUpload(uploadID) + return nil +} + +// removeUpload deletes the upload's temp directory and drops its bookkeeping. +// Missing uploads are ignored to keep abort/complete idempotent. +func (b *s3Backend) removeUpload(uploadID gofakes3.UploadID) { + val, ok := b.uploads.LoadAndDelete(uploadID) + if !ok { + return + } + state := val.(*multipartState) + if state.dir != "" { + if err := os.RemoveAll(state.dir); err != nil { + log.Warnf("s3 multipart: failed to clean up %s: %v", state.dir, err) + } + } +} + +// md5Bytes returns the raw 16-byte MD5 digest of the part. +func (p *multipartPart) md5Bytes() []byte { + b, _ := hex.DecodeString(p.md5hex) + return b +} diff --git a/server/s3/multipart_test.go b/server/s3/multipart_test.go new file mode 100644 index 000000000..39f4561cd --- /dev/null +++ b/server/s3/multipart_test.go @@ -0,0 +1,245 @@ +package s3 + +import ( + "bytes" + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + + _ "github.com/OpenListTeam/OpenList/v4/drivers/local" + "github.com/OpenListTeam/OpenList/v4/internal/conf" + "github.com/OpenListTeam/OpenList/v4/internal/db" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/OpenList/v4/internal/op" + + "github.com/OpenListTeam/gofakes3" + "github.com/glebarez/sqlite" + "gorm.io/gorm" +) + +func init() { + dataDir, err := os.MkdirTemp("", "openlist-s3-mp-*") + if err != nil { + panic(err) + } + conf.Conf = conf.DefaultConfig(dataDir) + if err := os.MkdirAll(conf.Conf.TempDir, 0o755); err != nil { + panic("mkdir temp dir: " + err.Error()) + } + dB, err := gorm.Open(sqlite.Open("file::memory:?cache=shared"), &gorm.Config{}) + if err != nil { + panic("failed to connect database: " + err.Error()) + } + db.Init(dB) +} + +// s3ErrorCode extracts the gofakes3 ErrorCode from an error returned by the +// MultipartBackend methods. +func s3ErrorCode(err error) gofakes3.ErrorCode { + if err == nil { + return gofakes3.ErrNone + } + var s3err interface{ ErrorCode() gofakes3.ErrorCode } + if errors.As(err, &s3err) { + return s3err.ErrorCode() + } + return gofakes3.ErrNone +} + +// setupMultipartBackend prepares a Local storage mounted at /mpbucket and an +// s3Backend with an "mp" bucket pointing at it. It returns the backend, the +// local root directory on disk, and a cleanup function. +func setupMultipartBackend(t *testing.T) (*s3Backend, string) { + t.Helper() + ctx := context.Background() + + // Unique mount path and bucket per test: the in-memory sqlite is shared + // across tests in this package, so a fixed mount path would clash. + mount := "/" + sanitizeTestName(t.Name()) + bucket := "mp" + + localRoot, err := os.MkdirTemp("", "openlist-s3-local-*") + if err != nil { + t.Fatalf("mkdir local root: %v", err) + } + t.Cleanup(func() { _ = os.RemoveAll(localRoot) }) + + _, err = op.CreateStorage(ctx, model.Storage{ + Driver: "Local", + MountPath: mount, + Addition: `{"root_folder_path":"` + localRoot + `","thumbnail":false}`, + }) + if err != nil { + t.Fatalf("create local storage: %+v", err) + } + + if err := op.SaveSettingItem(&model.SettingItem{ + Key: conf.S3Buckets, + Value: `[{"name":"` + bucket + `","path":"` + mount + `"}]`, + }); err != nil { + t.Fatalf("save s3 buckets setting: %+v", err) + } + + return newBackend().(*s3Backend), localRoot +} + +func sanitizeTestName(name string) string { + r := strings.NewReplacer("/", "_", " ", "_") + return r.Replace(name) +} + +func TestMultipartUploadEndToEnd(t *testing.T) { + ctx := context.Background() + b, localRoot := setupMultipartBackend(t) + + meta := map[string]string{"Content-Type": "text/plain"} + uploadID, err := b.CreateMultipartUpload(ctx, "mp", "dir/hello.txt", meta) + if err != nil { + t.Fatalf("CreateMultipartUpload: %+v", err) + } + if uploadID == "" { + t.Fatal("empty upload id") + } + + part := func(n int, body string) string { + t.Helper() + etag, err := b.UploadPart(ctx, "mp", "dir/hello.txt", uploadID, n, int64(len(body)), strings.NewReader(body)) + if err != nil { + t.Fatalf("UploadPart %d: %+v", n, err) + } + return etag + } + + etag1 := part(1, "Hello, ") + etag2 := part(2, "multipart ") + etag3 := part(3, "world!") + + // Re-uploading the same part number overwrites it and returns a fresh etag. + if e := part(2, "multipart "); e != etag2 { + t.Fatalf("re-upload part 2 etag = %q, want %q", e, etag2) + } + + // Short read (body smaller than declared Content-Length) must fail. + _, shortErr := b.UploadPart(ctx, "mp", "dir/hello.txt", uploadID, 4, 10, strings.NewReader("abc")) + shortErr = s3ErrorCode(shortErr) + if shortErr != gofakes3.ErrIncompleteBody { + t.Fatalf("short read error = %v, want IncompleteBody", shortErr) + } + + // Unknown upload id. + _, uerr := b.UploadPart(ctx, "mp", "dir/hello.txt", "does-not-exist", 1, 1, strings.NewReader("x")) + if code := s3ErrorCode(uerr); code != gofakes3.ErrNoSuchUpload { + t.Fatalf("unknown upload error = %v, want NoSuchUpload", code) + } + // Out-of-range part number. + _, perr := b.UploadPart(ctx, "mp", "dir/hello.txt", uploadID, 0, 1, strings.NewReader("x")) + if code := s3ErrorCode(perr); code != gofakes3.ErrInvalidPart { + t.Fatalf("part 0 error = %v, want InvalidPart", code) + } + + // Parts out of order. + _, _, err = b.CompleteMultipartUpload(ctx, "mp", "dir/hello.txt", uploadID, &gofakes3.CompleteMultipartUploadRequest{ + Parts: []gofakes3.CompletedPart{ + {PartNumber: 2, ETag: etag2}, + {PartNumber: 1, ETag: etag1}, + }, + }) + if code := s3ErrorCode(err); code != gofakes3.ErrInvalidPartOrder { + t.Fatalf("out-of-order complete error = %v, want InvalidPartOrder", code) + } + + // Wrong etag. + _, _, err = b.CompleteMultipartUpload(ctx, "mp", "dir/hello.txt", uploadID, &gofakes3.CompleteMultipartUploadRequest{ + Parts: []gofakes3.CompletedPart{ + {PartNumber: 1, ETag: etag1}, + {PartNumber: 2, ETag: `"deadbeef"`}, + {PartNumber: 3, ETag: etag3}, + }, + }) + if code := s3ErrorCode(err); code != gofakes3.ErrInvalidPart { + t.Fatalf("wrong etag complete error = %v, want InvalidPart", code) + } + + // Missing part number. + _, _, err = b.CompleteMultipartUpload(ctx, "mp", "dir/hello.txt", uploadID, &gofakes3.CompleteMultipartUploadRequest{ + Parts: []gofakes3.CompletedPart{ + {PartNumber: 1, ETag: etag1}, + {PartNumber: 99, ETag: etag3}, + }, + }) + if code := s3ErrorCode(err); code != gofakes3.ErrInvalidPart { + t.Fatalf("missing part complete error = %v, want InvalidPart", code) + } + + // The failed completes must leave the upload available for retry. + if _, ok := b.uploads.Load(uploadID); !ok { + t.Fatal("upload was removed after a failed complete") + } + + // Successful complete. + _, etag, err := b.CompleteMultipartUpload(ctx, "mp", "dir/hello.txt", uploadID, &gofakes3.CompleteMultipartUploadRequest{ + Parts: []gofakes3.CompletedPart{ + {PartNumber: 1, ETag: etag1}, + {PartNumber: 2, ETag: etag2}, + {PartNumber: 3, ETag: etag3}, + }, + }) + if err != nil { + t.Fatalf("CompleteMultipartUpload: %+v", err) + } + if !strings.HasSuffix(etag, `-3"`) || !strings.HasPrefix(etag, `"`) { + t.Fatalf("complete etag = %q, want a quoted \"-3\" multipart etag", etag) + } + + // The object must exist on disk with the concatenated content. + got, err := os.ReadFile(filepath.Join(localRoot, "dir", "hello.txt")) + if err != nil { + t.Fatalf("read resulting file: %+v", err) + } + want := []byte("Hello, multipart world!") + if !bytes.Equal(got, want) { + t.Fatalf("resulting file content = %q, want %q", got, want) + } + + // Bookkeeping and temp files must be cleaned up on success. + if _, ok := b.uploads.Load(uploadID); ok { + t.Fatal("upload still tracked after successful complete") + } +} + +func TestMultipartAbort(t *testing.T) { + ctx := context.Background() + b, _ := setupMultipartBackend(t) + + uploadID, err := b.CreateMultipartUpload(ctx, "mp", "abort.txt", nil) + if err != nil { + t.Fatalf("CreateMultipartUpload: %+v", err) + } + if _, err := b.UploadPart(ctx, "mp", "abort.txt", uploadID, 1, 3, strings.NewReader("abc")); err != nil { + t.Fatalf("UploadPart: %+v", err) + } + + state, _ := b.uploads.Load(uploadID) + dir := state.(*multipartState).dir + if _, err := os.Stat(dir); err != nil { + t.Fatalf("temp dir missing before abort: %+v", err) + } + + if err := b.AbortMultipartUpload(ctx, "mp", "abort.txt", uploadID); err != nil { + t.Fatalf("AbortMultipartUpload: %+v", err) + } + if _, ok := b.uploads.Load(uploadID); ok { + t.Fatal("upload still tracked after abort") + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Fatalf("temp dir still exists after abort (err=%v)", err) + } + + // Aborting an unknown upload must be idempotent. + if err := b.AbortMultipartUpload(ctx, "mp", "abort.txt", "nope"); err != nil { + t.Fatalf("abort unknown upload returned error: %+v", err) + } +}