From ac3797ba6b8b706e50a74bfee855c5fe349128f1 Mon Sep 17 00:00:00 2001 From: MadDogOwner Date: Sun, 12 Jul 2026 00:50:46 +0800 Subject: [PATCH 1/3] refactor(server/s3): use OpenListTeam/gofakes3 Replace itsHenry35/gofakes3 with OpenListTeam/gofakes3 Signed-off-by: MadDogOwner --- go.mod | 2 +- go.sum | 4 ++-- server/s3/backend.go | 2 +- server/s3/list.go | 2 +- server/s3/logger.go | 2 +- server/s3/pager.go | 2 +- server/s3/redirect.go | 2 +- server/s3/server.go | 2 +- server/s3/utils.go | 2 +- 9 files changed, 10 insertions(+), 10 deletions(-) diff --git a/go.mod b/go.mod index d0fd7fd7da..f2803517e4 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/KarpelesLab/reflink v1.0.2 github.com/KirCute/zip v1.0.1 github.com/OpenListTeam/go-cache v0.1.0 + github.com/OpenListTeam/gofakes3 v0.8.0 github.com/OpenListTeam/sftpd-openlist v1.0.1 github.com/OpenListTeam/tache v0.2.2 github.com/OpenListTeam/times v0.1.0 @@ -51,7 +52,6 @@ require ( github.com/hekmon/transmissionrpc/v3 v3.0.0 github.com/henrybear327/go-proton-api v1.0.0 github.com/ipfs/go-ipfs-api v0.7.0 - github.com/itsHenry35/gofakes3 v0.0.8 github.com/jlaffaye/ftp v0.2.1-0.20251026020404-6602e981a1bb github.com/json-iterator/go v1.1.12 github.com/kdomanski/iso9660 v0.4.0 diff --git a/go.sum b/go.sum index 924bf79cad..828b1a4e54 100644 --- a/go.sum +++ b/go.sum @@ -39,6 +39,8 @@ github.com/OpenListTeam/115-sdk-go v0.2.5 h1:E4O7GZmEXlGnouZ/e9//xU7NepT9JJkWtnF github.com/OpenListTeam/115-sdk-go v0.2.5/go.mod h1:cfvitk2lwe6036iNi2h+iNxwxWDifKZsSvNtrur5BqU= github.com/OpenListTeam/go-cache v0.1.0 h1:eV2+FCP+rt+E4OCJqLUW7wGccWZNJMV0NNkh+uChbAI= github.com/OpenListTeam/go-cache v0.1.0/go.mod h1:AHWjKhNK3LE4rorVdKyEALDHoeMnP8SjiNyfVlB+Pz4= +github.com/OpenListTeam/gofakes3 v0.8.0 h1:Y+Stbb29qfDUrGYmK0Tqzk8c0QXwK4Pkmk6lG5zpWwk= +github.com/OpenListTeam/gofakes3 v0.8.0/go.mod h1:DsJxAVrGCaleujFKrAYyGAjB5FtbPVF8IwAFYZ6pBKM= github.com/OpenListTeam/gsync v0.1.0 h1:ywzGybOvA3lW8K1BUjKZ2IUlT2FSlzPO4DOazfYXjcs= github.com/OpenListTeam/gsync v0.1.0/go.mod h1:h/Rvv9aX/6CdW/7B8di3xK3xNV8dUg45Fehrd/ksZ9s= github.com/OpenListTeam/reflink v0.0.0-20260701021214-78760eaeafef h1:67uGHancMF/abMrnkc8abVUWQiG73Wk5d8CKt3RzkFo= @@ -427,8 +429,6 @@ github.com/ipfs/go-cid v0.5.0 h1:goEKKhaGm0ul11IHA7I6p1GmKz8kEYniqFopaB5Otwg= github.com/ipfs/go-cid v0.5.0/go.mod h1:0L7vmeNXpQpUS9vt+yEARkJ8rOg43DF3iPgn4GIN0mk= github.com/ipfs/go-ipfs-api v0.7.0 h1:CMBNCUl0b45coC+lQCXEVpMhwoqjiaCwUIrM+coYW2Q= github.com/ipfs/go-ipfs-api v0.7.0/go.mod h1:AIxsTNB0+ZhkqIfTZpdZ0VR/cpX5zrXjATa3prSay3g= -github.com/itsHenry35/gofakes3 v0.0.8 h1:1AgOl04IgoUV5r/WSK7ycnvwfpgharYLfVTmnzk5miw= -github.com/itsHenry35/gofakes3 v0.0.8/go.mod h1:gQwOJ7LoH5QSpCVmjzC6oKp+MS71utLS7GHtonsvD0c= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= diff --git a/server/s3/backend.go b/server/s3/backend.go index f00d2149c5..50bdeba1b4 100644 --- a/server/s3/backend.go +++ b/server/s3/backend.go @@ -23,7 +23,7 @@ import ( "github.com/OpenListTeam/OpenList/v4/internal/stream" "github.com/OpenListTeam/OpenList/v4/pkg/http_range" "github.com/OpenListTeam/OpenList/v4/pkg/utils" - "github.com/itsHenry35/gofakes3" + "github.com/OpenListTeam/gofakes3" "github.com/ncw/swift/v2" log "github.com/sirupsen/logrus" ) diff --git a/server/s3/list.go b/server/s3/list.go index d783099a51..2b9d79693d 100644 --- a/server/s3/list.go +++ b/server/s3/list.go @@ -7,7 +7,7 @@ import ( "strings" "time" - "github.com/itsHenry35/gofakes3" + "github.com/OpenListTeam/gofakes3" log "github.com/sirupsen/logrus" ) diff --git a/server/s3/logger.go b/server/s3/logger.go index daf0aec24c..1715a40040 100644 --- a/server/s3/logger.go +++ b/server/s3/logger.go @@ -6,7 +6,7 @@ import ( "fmt" "github.com/OpenListTeam/OpenList/v4/pkg/utils" - "github.com/itsHenry35/gofakes3" + "github.com/OpenListTeam/gofakes3" ) // logger output formatted message diff --git a/server/s3/pager.go b/server/s3/pager.go index 2682429710..51e710b656 100644 --- a/server/s3/pager.go +++ b/server/s3/pager.go @@ -5,7 +5,7 @@ package s3 import ( "sort" - "github.com/itsHenry35/gofakes3" + "github.com/OpenListTeam/gofakes3" ) // pager splits the object list into smulitply pages. diff --git a/server/s3/redirect.go b/server/s3/redirect.go index 4031e171d9..6d8d430bc3 100644 --- a/server/s3/redirect.go +++ b/server/s3/redirect.go @@ -14,7 +14,7 @@ import ( "github.com/OpenListTeam/OpenList/v4/internal/op" "github.com/OpenListTeam/OpenList/v4/pkg/utils" "github.com/OpenListTeam/OpenList/v4/server/common" - "github.com/itsHenry35/gofakes3/signature" + "github.com/OpenListTeam/gofakes3/signature" ) func redirectHandler(next http.Handler, authPairs map[string]string) http.Handler { diff --git a/server/s3/server.go b/server/s3/server.go index d33b63ddde..451eb9f383 100644 --- a/server/s3/server.go +++ b/server/s3/server.go @@ -7,7 +7,7 @@ import ( "math/rand" "net/http" - "github.com/itsHenry35/gofakes3" + "github.com/OpenListTeam/gofakes3" ) // Make a new S3 Server to serve the remote diff --git a/server/s3/utils.go b/server/s3/utils.go index eb3dd56850..ec406e54e7 100644 --- a/server/s3/utils.go +++ b/server/s3/utils.go @@ -13,7 +13,7 @@ import ( "github.com/OpenListTeam/OpenList/v4/internal/model" "github.com/OpenListTeam/OpenList/v4/internal/op" "github.com/OpenListTeam/OpenList/v4/internal/setting" - "github.com/itsHenry35/gofakes3" + "github.com/OpenListTeam/gofakes3" ) type Bucket struct { From 190f07b9918b4877e2db661154eb9ed85abcf25b Mon Sep 17 00:00:00 2001 From: MadDogOwner Date: Tue, 21 Jul 2026 00:22:24 +0800 Subject: [PATCH 2/3] 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 | 43 +++--- server/s3/multipart.go | 272 ++++++++++++++++++++++++++++++++++++ server/s3/multipart_test.go | 245 ++++++++++++++++++++++++++++++++ 3 files changed, 543 insertions(+), 17 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 50bdeba1b4..902870d5a4 100644 --- a/server/s3/backend.go +++ b/server/s3/backend.go @@ -33,16 +33,20 @@ 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 + meta *sync.Map + uploads *sync.Map // map[gofakes3.UploadID]*multipartState } // newBackend creates a new SimpleBucketBackend. func newBackend() gofakes3.Backend { return &s3Backend{ - meta: new(sync.Map), + meta: new(sync.Map), + uploads: new(sync.Map), } } @@ -232,9 +236,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 @@ -260,15 +275,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 @@ -294,7 +309,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, @@ -304,18 +319,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 0000000000..38bfe4de1c --- /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 0000000000..39f4561cd8 --- /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) + } +} From 9a51b73299a472e9dea42c15ed7ac94b5b5b6813 Mon Sep 17 00:00:00 2001 From: MadDogOwner Date: Tue, 21 Jul 2026 14:46:16 +0800 Subject: [PATCH 3/3] feat(server/s3): reap abandoned multipart uploads - Track lastActivity on each multipart upload, updated on create and every part upload, so idle uploads can be detected - Add a background reaper per backend instance that removes uploads inactive for longer than the TTL (default 24h) and cleans their temp directories - Add a startup sweep that removes leftover s3-multipart-* directories older than the TTL, recovering part files from a previous crash - Add a configurable s3.multipart_ttl (env S3_MULTIPART_TTL) duration, parsed via time.ParseDuration with a 24h default - Add tests for TTL-based reaping and stale-directory cleanup Co-authored-by: Codex <267193182+codex@users.noreply.github.com> --- internal/conf/config.go | 7 +- server/s3/backend.go | 4 +- server/s3/multipart.go | 131 ++++++++++++++++++++++++++++++++---- server/s3/multipart_test.go | 76 +++++++++++++++++++++ 4 files changed, 202 insertions(+), 16 deletions(-) diff --git a/internal/conf/config.go b/internal/conf/config.go index 0f66761168..53e79f82e7 100644 --- a/internal/conf/config.go +++ b/internal/conf/config.go @@ -83,9 +83,10 @@ type Cors struct { } type S3 struct { - Enable bool `json:"enable" env:"ENABLE"` - Port int `json:"port" env:"PORT"` - SSL bool `json:"ssl" env:"SSL"` + Enable bool `json:"enable" env:"ENABLE"` + Port int `json:"port" env:"PORT"` + SSL bool `json:"ssl" env:"SSL"` + MultipartTTL string `json:"multipart_ttl" env:"MULTIPART_TTL"` } type FTP struct { diff --git a/server/s3/backend.go b/server/s3/backend.go index 902870d5a4..0a13db3176 100644 --- a/server/s3/backend.go +++ b/server/s3/backend.go @@ -44,10 +44,12 @@ type s3Backend struct { // newBackend creates a new SimpleBucketBackend. func newBackend() gofakes3.Backend { - return &s3Backend{ + b := &s3Backend{ meta: new(sync.Map), uploads: new(sync.Map), } + b.startReaper() + return b } // ListBuckets always returns the default bucket. diff --git a/server/s3/multipart.go b/server/s3/multipart.go index 38bfe4de1c..8069864709 100644 --- a/server/s3/multipart.go +++ b/server/s3/multipart.go @@ -43,11 +43,12 @@ type multipartPart struct { // 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 + bucket string + object string + meta map[string]string + dir string + created time.Time + lastActivity time.Time // updated under mu on create and each part upload mu sync.Mutex parts map[int]*multipartPart @@ -72,13 +73,15 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucket, object st } uploadID := gofakes3.UploadID(strings.ReplaceAll(uuid.NewString(), "-", "")) + now := time.Now() state := &multipartState{ - bucket: bucket, - object: object, - meta: meta, - dir: dir, - created: time.Now(), - parts: map[int]*multipartPart{}, + bucket: bucket, + object: object, + meta: meta, + dir: dir, + created: now, + lastActivity: now, + parts: map[int]*multipartPart{}, } b.uploads.Store(uploadID, state) @@ -135,6 +138,7 @@ func (b *s3Backend) UploadPart(ctx context.Context, bucket, object string, uploa md5hex := hex.EncodeToString(hash.Sum(nil)) etag := fmt.Sprintf("%q", md5hex) + now := time.Now() state.mu.Lock() if old := state.parts[partNumber]; old != nil && old.path != partPath { _ = os.Remove(old.path) @@ -143,8 +147,9 @@ func (b *s3Backend) UploadPart(ctx context.Context, bucket, object string, uploa path: partPath, size: n, md5hex: md5hex, - updated: time.Now(), + updated: now, } + state.lastActivity = now state.mu.Unlock() partFailed = false @@ -270,3 +275,105 @@ func (p *multipartPart) md5Bytes() []byte { b, _ := hex.DecodeString(p.md5hex) return b } + +// Defaults for reaping abandoned multipart uploads. A client that never sends +// CompleteMultipartUpload or AbortMultipartUpload would otherwise leave part +// files on disk forever; the reaper drops uploads inactive for longer than the +// TTL. +const ( + defaultMultipartTTL = 24 * time.Hour + multipartDirPrefix = "s3-multipart-" +) + +// multipartTTL returns the configured max idle time for an upload before the +// reaper reclaims it. It parses conf.Conf.S3.MultipartTTL as a Go duration +// (e.g. "24h", "30m"); an empty or invalid value falls back to the default. +func multipartTTL() time.Duration { + if v := conf.Conf.S3.MultipartTTL; v != "" { + if d, err := time.ParseDuration(v); err == nil && d > 0 { + return d + } + } + return defaultMultipartTTL +} + +// reapInterval derives the reaper tick interval from the TTL: a quarter of the +// TTL, clamped to [10s, 1h]. +func reapInterval(ttl time.Duration) time.Duration { + d := ttl / 4 + if d < 10*time.Second { + d = 10 * time.Second + } + if d > time.Hour { + d = time.Hour + } + return d +} + +// startReaper removes leftover part directories from a previous process crash +// and then launches a background goroutine that periodically reclaims uploads +// inactive for longer than the TTL. The goroutine runs for the lifetime of the +// process; NewServer is called once at startup, so there is one reaper per +// backend instance. +func (b *s3Backend) startReaper() { + b.cleanupStaleDirs(time.Now(), multipartTTL()) + interval := reapInterval(multipartTTL()) + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for now := range ticker.C { + b.reapExpired(now, multipartTTL()) + } + }() +} + +// reapExpired removes uploads whose lastActivity is older than ttl. It is safe +// to call concurrently with UploadPart/Complete/Abort: each candidate is +// re-checked under its own lock and removed atomically via removeUpload. +func (b *s3Backend) reapExpired(now time.Time, ttl time.Duration) { + b.uploads.Range(func(key, val any) bool { + state := val.(*multipartState) + state.mu.Lock() + expired := now.Sub(state.lastActivity) > ttl + state.mu.Unlock() + if !expired { + return true + } + b.removeUpload(key.(gofakes3.UploadID)) + log.Infof("s3 multipart: reaped abandoned upload %s (%s/%s)", key, state.bucket, state.object) + return true + }) +} + +// cleanupStaleDirs removes s3-multipart-* directories under TempDir that are +// older than ttl. This reclaims part files left behind by a previous process +// crash; dirs younger than ttl are left alone so a concurrently-starting +// sibling backend instance is never disturbed. +func (b *s3Backend) cleanupStaleDirs(now time.Time, ttl time.Duration) { + tempDir := conf.Conf.TempDir + if tempDir == "" { + tempDir = os.TempDir() + } + entries, err := os.ReadDir(tempDir) + if err != nil { + return + } + cutoff := now.Add(-ttl) + for _, e := range entries { + if !e.IsDir() || !strings.HasPrefix(e.Name(), multipartDirPrefix) { + continue + } + info, err := e.Info() + if err != nil { + continue + } + if info.ModTime().After(cutoff) { + continue + } + if err := os.RemoveAll(filepath.Join(tempDir, e.Name())); err != nil { + log.Warnf("s3 multipart: failed to clean up stale dir %s: %v", e.Name(), err) + } else { + log.Infof("s3 multipart: removed stale multipart dir %s", e.Name()) + } + } +} diff --git a/server/s3/multipart_test.go b/server/s3/multipart_test.go index 39f4561cd8..7adc4c5e3b 100644 --- a/server/s3/multipart_test.go +++ b/server/s3/multipart_test.go @@ -8,6 +8,7 @@ import ( "path/filepath" "strings" "testing" + "time" _ "github.com/OpenListTeam/OpenList/v4/drivers/local" "github.com/OpenListTeam/OpenList/v4/internal/conf" @@ -243,3 +244,78 @@ func TestMultipartAbort(t *testing.T) { t.Fatalf("abort unknown upload returned error: %+v", err) } } + +func TestMultipartReapExpired(t *testing.T) { + ctx := context.Background() + b, _ := setupMultipartBackend(t) + + // An active upload (fresh lastActivity) must be kept. + freshID, err := b.CreateMultipartUpload(ctx, "mp", "fresh.txt", nil) + if err != nil { + t.Fatalf("create fresh upload: %+v", err) + } + + // An abandoned upload (stale lastActivity) must be reaped. + staleID, err := b.CreateMultipartUpload(ctx, "mp", "stale.txt", nil) + if err != nil { + t.Fatalf("create stale upload: %+v", err) + } + if _, err := b.UploadPart(ctx, "mp", "stale.txt", staleID, 1, 3, strings.NewReader("abc")); err != nil { + t.Fatalf("upload stale part: %+v", err) + } + staleState, _ := b.uploads.Load(staleID) + staleDir := staleState.(*multipartState).dir + if _, err := os.Stat(staleDir); err != nil { + t.Fatalf("stale temp dir missing: %+v", err) + } + + // Force the stale upload's lastActivity well into the past. + ttl := 30 * time.Minute + now := time.Now() + staleState.(*multipartState).mu.Lock() + staleState.(*multipartState).lastActivity = now.Add(-2 * ttl) + staleState.(*multipartState).mu.Unlock() + + b.reapExpired(now, ttl) + + if _, ok := b.uploads.Load(staleID); ok { + t.Fatal("stale upload still tracked after reap") + } + if _, err := os.Stat(staleDir); !os.IsNotExist(err) { + t.Fatalf("stale temp dir still exists after reap (err=%v)", err) + } + if _, ok := b.uploads.Load(freshID); !ok { + t.Fatal("fresh upload was reaped, should have been kept") + } +} + +func TestMultipartCleanupStaleDirs(t *testing.T) { + b, _ := setupMultipartBackend(t) + + tempDir := conf.Conf.TempDir + staleDir, err := os.MkdirTemp(tempDir, multipartDirPrefix+"*") + if err != nil { + t.Fatalf("mkdir stale dir: %v", err) + } + freshDir, err := os.MkdirTemp(tempDir, multipartDirPrefix+"*") + if err != nil { + t.Fatalf("mkdir fresh dir: %v", err) + } + // Age the stale dir beyond the TTL; leave the fresh dir young. + ttl := 30 * time.Minute + now := time.Now() + past := now.Add(-2 * ttl) + if err := os.Chtimes(staleDir, past, past); err != nil { + t.Fatalf("chtimes stale dir: %v", err) + } + + b.cleanupStaleDirs(now, ttl) + + if _, err := os.Stat(staleDir); !os.IsNotExist(err) { + t.Fatalf("stale dir should have been removed (err=%v)", err) + } + if _, err := os.Stat(freshDir); err != nil { + t.Fatalf("fresh dir should have been kept (err=%v)", err) + } + _ = os.RemoveAll(freshDir) +}