diff --git a/internal/bootstrap/data/setting.go b/internal/bootstrap/data/setting.go index 4526aa5234..54d8bb91f9 100644 --- a/internal/bootstrap/data/setting.go +++ b/internal/bootstrap/data/setting.go @@ -249,6 +249,8 @@ func InitialSettings() []model.SettingItem { {Key: conf.StreamMaxClientUploadSpeed, Value: "-1", Type: conf.TypeNumber, Group: model.TRAFFIC, Flag: model.PRIVATE}, {Key: conf.StreamMaxServerDownloadSpeed, Value: "-1", Type: conf.TypeNumber, Group: model.TRAFFIC, Flag: model.PRIVATE}, {Key: conf.StreamMaxServerUploadSpeed, Value: "-1", Type: conf.TypeNumber, Group: model.TRAFFIC, Flag: model.PRIVATE}, + {Key: conf.MultipartEnabled, Value: "true", Type: conf.TypeBool, Group: model.TRAFFIC, Flag: model.PUBLIC}, + {Key: conf.MultipartChunkSize, Value: "10", Type: conf.TypeNumber, Group: model.TRAFFIC, Flag: model.PUBLIC, Help: `chunk size of multipart upload in MB (positive integer), keep it under your CDN's request body limit; each active session buffers up to 8 chunks on the server's disk`}, } additionalSettingItems := tool.Tools.Items() // 固定顺序 diff --git a/internal/conf/const.go b/internal/conf/const.go index b99d8849cb..06d9851a75 100644 --- a/internal/conf/const.go +++ b/internal/conf/const.go @@ -161,6 +161,8 @@ const ( StreamMaxClientUploadSpeed = "max_client_upload_speed" StreamMaxServerDownloadSpeed = "max_server_download_speed" StreamMaxServerUploadSpeed = "max_server_upload_speed" + MultipartEnabled = "multipart_enabled" + MultipartChunkSize = "multipart_chunk_size" ) const ( diff --git a/internal/multipart/session.go b/internal/multipart/session.go new file mode 100644 index 0000000000..e25d1d1c7e --- /dev/null +++ b/internal/multipart/session.go @@ -0,0 +1,607 @@ +package multipart + +import ( + "context" + "errors" + "fmt" + "io" + "math" + "os" + stdpath "path" + "path/filepath" + "sync" + "sync/atomic" + "time" + + "github.com/OpenListTeam/OpenList/v4/internal/conf" + "github.com/OpenListTeam/OpenList/v4/internal/driver" + "github.com/OpenListTeam/OpenList/v4/internal/errs" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/OpenList/v4/internal/op" + "github.com/OpenListTeam/OpenList/v4/internal/stream" + "github.com/OpenListTeam/OpenList/v4/pkg/utils" + "github.com/google/uuid" +) + +type State string + +const ( + StateReceiving State = "receiving" + StateCompleted State = "completed" + StateFailedRetriable State = "failed_retriable" + StateFailedPermanent State = "failed_permanent" + StateAborted State = "aborted" +) + +// WindowSlots bounds the per-session disk footprint to WindowSlots*ChunkSize. +var WindowSlots = 8 + +const ( + // defaultSessionTTL is the sliding inactivity timeout; it also serves as the + // grace period during which finished sessions remain queryable. + defaultSessionTTL = 30 * time.Minute + gcInterval = time.Minute +) + +var ( + ErrSessionNotFound = errors.New("multipart upload session not found") + ErrNotOwner = errors.New("multipart upload session belongs to another user") + // errAborted wraps context.Canceled so a driver blocked on the stream sees + // the abort as a canceled request and runs its cancellation cleanup. + errAborted = fmt.Errorf("multipart upload aborted: %w", context.Canceled) +) + +// putFile is the pipeline tail: resolve the storage and run the regular upload +// path. It mirrors the checks of fs.putDirectly (internal/fs/put.go) but calls +// op.Put directly so the driver's progress callback can be observed. +// It is a variable so session tests can stub the storage layer out. +var putFile = func(ctx context.Context, dstDirPath string, fs *stream.FileStream, up driver.UpdateProgress) error { + storage, dstDirActualPath, err := op.GetStorageAndActualPath(dstDirPath) + if err != nil { + return err + } + if storage.Config().NoUpload { + return errs.UploadNotSupported + } + return op.Put(ctx, storage, dstDirActualPath, fs, up) +} + +// Session is one multipart upload: metadata survives pipeline attempts, the +// window (chunk data) does not. +type Session struct { + ID string + Path string // full destination path (dir + name), already user-joined + DstDir string + Name string + Size int64 + ChunkSize int64 + Total int + Mimetype string + Modified time.Time + Hashes map[*utils.HashType]string + Creator *model.User + + mu sync.Mutex + state State + err error + attempt int + win *Window + done chan struct{} + cancel context.CancelFunc + prevCRCs []uint32 + prevSet []bool + + storagePct atomic.Uint64 // math.Float64bits of the driver progress (0-100) + lastActive atomic.Int64 // unix nano +} + +func (s *Session) touch() { s.lastActive.Store(time.Now().UnixNano()) } + +func (s *Session) setStoragePct(p float64) { s.storagePct.Store(math.Float64bits(p)) } + +// Snapshot is the wire representation of a session used by all endpoints. +type SessionSnapshot struct { + ID string `json:"upload_id"` + State State `json:"state"` + Attempt int `json:"attempt"` + Path string `json:"path"` + Size int64 `json:"size"` + ChunkSize int64 `json:"chunk_size"` + TotalChunks int `json:"total_chunks"` + Received [][2]int `json:"received"` + ReceivedBytes int64 `json:"received_bytes"` + Frontier int `json:"frontier"` + StorageProgress float64 `json:"storage_progress"` + Error string `json:"error,omitempty"` +} + +func (s *Session) snapshotLocked() SessionSnapshot { + snap := SessionSnapshot{ + ID: s.ID, + State: s.state, + Attempt: s.attempt, + Path: s.Path, + Size: s.Size, + ChunkSize: s.ChunkSize, + TotalChunks: s.Total, + Received: [][2]int{}, + StorageProgress: math.Float64frombits(s.storagePct.Load()), + } + if s.err != nil { + snap.Error = s.err.Error() + } + switch { + case s.state == StateCompleted: + snap.Received = [][2]int{{0, s.Total - 1}} + snap.ReceivedBytes = s.Size + snap.Frontier = s.Total + snap.StorageProgress = 100 + case s.win != nil: + ws := s.win.Snapshot() + snap.Received = ws.Received + snap.ReceivedBytes = ws.ReceivedBytes + snap.Frontier = ws.Frontier + } + return snap +} + +func (s *Session) Snapshot() SessionSnapshot { + s.mu.Lock() + defer s.mu.Unlock() + return s.snapshotLocked() +} + +// InitReq carries everything the handler parsed from the init request. +type InitReq struct { + User *model.User + Path string // full destination path, already user-joined + Size int64 + ChunkSize int64 // final chunk size in bytes, already clamped by the handler + Mimetype string + Modified time.Time + Hashes map[*utils.HashType]string +} + +// Manager owns all live sessions. Sessions are in-memory only (aligned with +// upload tasks not being persisted); a restart drops them and the ring files +// are swept on the next start. +type Manager struct { + mu sync.Mutex + byID map[string]*Session + byKey map[string]string + gcOnce sync.Once + ttl time.Duration // 0 means defaultSessionTTL; tests shrink it per instance +} + +func (m *Manager) sessionTTL() time.Duration { + if m.ttl > 0 { + return m.ttl + } + return defaultSessionTTL +} + +var DefaultManager = &Manager{ + byID: make(map[string]*Session), + byKey: make(map[string]string), +} + +func (m *Manager) dir() string { + return filepath.Join(conf.Conf.TempDir, "multipart") +} + +func sessionKey(userID uint, path string, size int64) string { + return fmt.Sprintf("%d|%s|%d", userID, path, size) +} + +// 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. +func hashesQualifyResume(old, new map[*utils.HashType]string) bool { + shared := false + for t, ov := range old { + if nv, ok := new[t]; ok { + if ov != nv { + return false + } + shared = true + } + } + return shared +} + +// StartGC sweeps ring files orphaned by a previous run and starts the expiry +// loop. It is called at server startup so orphans are reclaimed even if no +// multipart upload ever happens again; Init also calls it, so embedders that +// skip the server wiring still get GC lazily. +func (m *Manager) StartGC() { + m.ensureGC() +} + +func (m *Manager) ensureGC() { + m.gcOnce.Do(func() { + // sweep ring files orphaned by a previous run; bootstrap's CleanTempDir + // only runs when no transfer tasks are pending, so do not rely on it + _ = os.RemoveAll(m.dir()) + go func() { + ticker := time.NewTicker(gcInterval) + for range ticker.C { + m.gc() + } + }() + }) +} + +func (m *Manager) gc() { + deadline := time.Now().Add(-m.sessionTTL()).UnixNano() + m.mu.Lock() + var expired []*Session + for _, s := range m.byID { + if s.lastActive.Load() < deadline { + expired = append(expired, s) + } + } + m.mu.Unlock() + for _, s := range expired { + m.terminate(s, errors.New("multipart upload session expired")) + } +} + +// terminate aborts a session (if still receiving) and drops it from the maps. +func (m *Manager) terminate(s *Session, cause error) { + s.mu.Lock() + if s.state == StateReceiving { + s.state = StateAborted + s.err = cause + } + s.killAttemptLocked() + s.mu.Unlock() + m.remove(s) +} + +// killAttemptLocked stops the running pipeline attempt: the context cancel +// interrupts drivers blocked on network I/O, and closing the window wakes a +// driver blocked in Read (context cancellation cannot interrupt cond.Wait). +// The caller must hold s.mu and must have set the final state first. +func (s *Session) killAttemptLocked() { + if s.cancel != nil { + s.cancel() + } + if s.win != nil { + _ = s.win.CloseWithError(errAborted) + } +} + +func (m *Manager) remove(s *Session) { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.byID, s.ID) + key := sessionKey(s.Creator.ID, s.Path, s.Size) + if m.byKey[key] == s.ID { + delete(m.byKey, key) + } +} + +func (m *Manager) get(user *model.User, id string) (*Session, error) { + m.mu.Lock() + s, ok := m.byID[id] + m.mu.Unlock() + if !ok { + return nil, ErrSessionNotFound + } + if s.Creator.ID != user.ID { + return nil, ErrNotOwner + } + return s, nil +} + +// Init creates a session and starts its pipeline, or returns the live session +// for the same (user, path, size) so an interrupted client resumes implicitly. +func (m *Manager) Init(req InitReq) (SessionSnapshot, bool, error) { + m.ensureGC() + if req.Size <= 0 { + return SessionSnapshot{}, false, fmt.Errorf("multipart upload requires a positive X-File-Size, got %d", req.Size) + } + if req.ChunkSize <= 0 { + return SessionSnapshot{}, false, fmt.Errorf("invalid chunk size %d", req.ChunkSize) + } + key := sessionKey(req.User.ID, req.Path, req.Size) + + m.mu.Lock() + if id, ok := m.byKey[key]; ok { + if s, ok := m.byID[id]; ok { + s.mu.Lock() + st := s.state + s.mu.Unlock() + // 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)) { + m.mu.Unlock() + s.touch() + return s.Snapshot(), true, nil + } + // finished session, or same path+size without proof of identity: + // drop the old session and start fresh + m.mu.Unlock() + m.terminate(s, errors.New("superseded by a new upload of the same path and size")) + m.mu.Lock() + } + } + m.mu.Unlock() + + dstDir, name := stdpath.Split(req.Path) + s := &Session{ + ID: uuid.NewString(), + Path: req.Path, + DstDir: dstDir, + Name: name, + Size: req.Size, + ChunkSize: req.ChunkSize, + Mimetype: req.Mimetype, + Modified: req.Modified, + Hashes: req.Hashes, + Creator: req.User, + state: StateReceiving, + } + s.touch() + + s.mu.Lock() + if err := m.startAttemptLocked(s); err != nil { + s.mu.Unlock() + return SessionSnapshot{}, false, err + } + s.Total = s.win.TotalChunks() + snap := s.snapshotLocked() + s.mu.Unlock() + + m.mu.Lock() + m.byID[s.ID] = s + m.byKey[key] = s.ID + m.mu.Unlock() + return snap, false, nil +} + +// startAttemptLocked builds a fresh window and spawns the pipeline goroutine. +// The caller must hold s.mu. +func (m *Manager) startAttemptLocked(s *Session) error { + win, err := NewWindow(m.dir(), fmt.Sprintf("%s.%d", s.ID, s.attempt), s.ChunkSize, s.Size, WindowSlots) + if err != nil { + return err + } + ctx, cancel := context.WithCancel(context.WithValue(context.Background(), conf.UserKey, s.Creator)) + done := make(chan struct{}) + s.win = win + s.cancel = cancel + s.done = done + s.state = StateReceiving + s.err = nil + s.setStoragePct(0) + + fileStream := &stream.FileStream{ + Obj: &model.Object{ + Name: s.Name, + Size: s.Size, + Modified: s.Modified, + HashInfo: utils.NewHashInfoByMap(s.Hashes), + }, + Reader: win, + Mimetype: s.Mimetype, + } + fileStream.Add(win) + + dstDir := s.DstDir + put := putFile // capture: the seam must not be read after spawn + go func() { + err := put(ctx, dstDir, fileStream, s.setStoragePct) + s.finishAttempt(win, err) + close(done) + }() + return nil +} + +// finishAttempt records the pipeline outcome and harvests the CRC table for +// re-fill verification. The window's data is gone at this point (op.Put closed +// it); only metadata survives. +func (s *Session) finishAttempt(win *Window, err error) { + crcs, set := win.CRCs() + _ = win.Close() // op.Put already closed it; make sure the ring file is gone anyway + + s.mu.Lock() + defer s.mu.Unlock() + s.touch() + if s.win == win { + s.win = nil + } + s.prevCRCs, s.prevSet = crcs, set + switch { + case err == nil: + s.state = StateCompleted + s.err = nil + s.setStoragePct(100) + case s.state != StateReceiving: + // Abort/expiry already labeled this attempt; keep that state. + if s.err == nil { + s.err = err + } + case isPermanentPutError(err): + s.state = StateFailedPermanent + s.err = err + default: + s.state = StateFailedRetriable + s.err = err + s.attempt++ + } +} + +func isPermanentPutError(err error) bool { + if errors.Is(err, context.Canceled) { + return false + } + for _, target := range []error{ + errs.UploadNotSupported, + errs.PermissionDenied, + errs.StorageNotFound, + errs.ObjectAlreadyExists, + errs.RelativePath, + errs.IgnoredSystemFile, + } { + if errors.Is(err, target) { + return true + } + } + return false +} + +// Chunk feeds one chunk into the session. Re-sending chunk 0 to a +// failed_retriable session re-fills it: a fresh window and pipeline attempt. +func (m *Manager) Chunk(user *model.User, id string, idx int, body io.Reader) (SessionSnapshot, error) { + s, err := m.get(user, id) + if err != nil { + return SessionSnapshot{}, err + } + s.touch() + + s.mu.Lock() + switch s.state { + case StateReceiving: + case StateFailedRetriable: + if idx != 0 { + snap := s.snapshotLocked() + s.mu.Unlock() + return snap, fmt.Errorf("upload attempt failed, resend from chunk 0 to retry: %w", s.err) + } + if err := m.startAttemptLocked(s); err != nil { + snap := s.snapshotLocked() + s.mu.Unlock() + return snap, err + } + case StateCompleted: + snap := s.snapshotLocked() + s.mu.Unlock() + return snap, nil // idempotent: stragglers after rapid-upload/finish succeed + default: + snap := s.snapshotLocked() + s.mu.Unlock() + return snap, fmt.Errorf("session is %s: %w", s.state, s.err) + } + win := s.win + var prevCRC uint32 + hasPrev := false + if idx < len(s.prevSet) && s.prevSet[idx] { + prevCRC, hasPrev = s.prevCRCs[idx], true + } + s.mu.Unlock() + + crc, err := win.WriteChunk(idx, body) + if err != nil { + // A closed window means the pipeline ended while this chunk was in + // flight — rapid upload makes this the NORMAL case: the driver + // succeeds off the hash alone with chunks still arriving. The window + // closes (op.Put's defer) moments before the verdict is recorded, so + // wait for the verdict instead of racing it, then absorb the chunk + // idempotently if the upload in fact succeeded. + if errors.Is(err, ErrClosed) || errors.Is(err, context.Canceled) { + s.mu.Lock() + done := s.done + s.mu.Unlock() + select { + case <-done: + case <-time.After(10 * time.Second): // pipeline teardown is µs-scale; never expected + } + } + s.mu.Lock() + completed := s.state == StateCompleted + snap := s.snapshotLocked() + s.mu.Unlock() + if completed { + return snap, nil + } + return snap, err + } + if hasPrev && crc != prevCRC { + err := fmt.Errorf("chunk %d content changed between attempts, aborting", idx) + s.mu.Lock() + s.state = StateFailedPermanent + s.err = err + s.killAttemptLocked() + s.mu.Unlock() + return s.Snapshot(), err + } + return s.Snapshot(), nil +} + +// Complete waits for the pipeline outcome. It refuses to block while chunks +// are still missing, so a buggy client cannot park a connection for the TTL. +func (m *Manager) Complete(ctx context.Context, user *model.User, id string) (SessionSnapshot, error) { + s, err := m.get(user, id) + if err != nil { + return SessionSnapshot{}, err + } + s.touch() + for { + s.mu.Lock() + st := s.state + done := s.done + if st == StateReceiving && s.win != nil { + if ws := s.win.Snapshot(); ws.ReceivedBytes < s.Size { + snap := s.snapshotLocked() + s.mu.Unlock() + return snap, fmt.Errorf("cannot complete: %d of %d bytes received", ws.ReceivedBytes, s.Size) + } + } + s.mu.Unlock() + if st != StateReceiving { + break + } + select { + case <-done: + case <-ctx.Done(): + return s.Snapshot(), ctx.Err() + } + } + s.touch() + s.mu.Lock() + st, serr := s.state, s.err + snap := s.snapshotLocked() + s.mu.Unlock() + if st == StateCompleted { + m.remove(s) // served its purpose; frees the key for future uploads + return snap, nil + } + return snap, fmt.Errorf("upload failed (%s): %w", st, serr) +} + +// Status looks a session up by id. +func (m *Manager) Status(user *model.User, id string) (SessionSnapshot, error) { + s, err := m.get(user, id) + if err != nil { + return SessionSnapshot{}, err + } + s.touch() + return s.Snapshot(), nil +} + +// Find looks a live session up by destination path and size, for resume discovery. +func (m *Manager) Find(user *model.User, path string, size int64) (SessionSnapshot, error) { + m.mu.Lock() + id, ok := m.byKey[sessionKey(user.ID, path, size)] + m.mu.Unlock() + if !ok { + return SessionSnapshot{}, ErrSessionNotFound + } + return m.Status(user, id) +} + +// Abort cancels the pipeline and forgets the session immediately. +func (m *Manager) Abort(user *model.User, id string) error { + s, err := m.get(user, id) + if err != nil { + return err + } + m.terminate(s, errors.New("multipart upload aborted by client")) + return nil +} diff --git a/internal/multipart/session_test.go b/internal/multipart/session_test.go new file mode 100644 index 0000000000..282d6a488c --- /dev/null +++ b/internal/multipart/session_test.go @@ -0,0 +1,650 @@ +package multipart + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + "sync" + "testing" + "time" + + "github.com/OpenListTeam/OpenList/v4/internal/conf" + "github.com/OpenListTeam/OpenList/v4/internal/driver" + "github.com/OpenListTeam/OpenList/v4/internal/errs" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/OpenList/v4/internal/stream" + "github.com/OpenListTeam/OpenList/v4/pkg/utils" +) + +// Tests in this file share the putFile seam and must not run in parallel. + +func setupSessionTest(t *testing.T) *Manager { + t.Helper() + oldConf := conf.Conf + conf.Conf = conf.DefaultConfig(t.TempDir()) + t.Cleanup(func() { conf.Conf = oldConf }) + return &Manager{byID: make(map[string]*Session), byKey: make(map[string]string)} +} + +func stubPut(t *testing.T, fn func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error) { + t.Helper() + orig := putFile + putFile = fn + t.Cleanup(func() { putFile = orig }) +} + +func testUser() *model.User { return &model.User{ID: 7, Username: "tester"} } + +func initReq(user *model.User, size, chunkSize int64) InitReq { + return InitReq{ + User: user, + Path: "/local/test.bin", + Size: size, + ChunkSize: chunkSize, + Mimetype: "application/octet-stream", + Modified: time.Unix(1700000000, 0), + } +} + +func sendChunk(t *testing.T, m *Manager, user *model.User, id string, data []byte, idx int, chunkSize int64) SessionSnapshot { + t.Helper() + snap, err := m.Chunk(user, id, idx, bytes.NewReader(chunkOf(data, idx, chunkSize))) + if err != nil { + t.Fatalf("Chunk(%d): %v", idx, err) + } + return snap +} + +func waitState(t *testing.T, m *Manager, user *model.User, id string, want State) SessionSnapshot { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + snap, err := m.Status(user, id) + if err != nil { + t.Fatalf("Status: %v", err) + } + if snap.State == want { + return snap + } + if time.Now().After(deadline) { + t.Fatalf("session state = %s, want %s (err: %s)", snap.State, want, snap.Error) + } + time.Sleep(2 * time.Millisecond) + } +} + +func TestSessionHappyPath(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(3*chunkSize + 300) + data := genData(totalSize) + user := testUser() + + got := make(chan []byte, 1) + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + if dst != "/local/" { + return fmt.Errorf("unexpected dst dir %q", dst) + } + if fs.GetName() != "test.bin" || fs.GetSize() != totalSize { + return fmt.Errorf("unexpected stream meta %s/%d", fs.GetName(), fs.GetSize()) + } + b, err := io.ReadAll(fs) + if err != nil { + return err + } + up(100) + got <- b + return nil + }) + + snap, resumed, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil || resumed { + t.Fatalf("Init = (resumed=%v, err=%v)", resumed, err) + } + if snap.TotalChunks != 4 || snap.State != StateReceiving { + t.Fatalf("init snapshot = %+v", snap) + } + for _, idx := range []int{1, 0, 3, 2} { // out of order on purpose + sendChunk(t, m, user, snap.ID, data, idx, chunkSize) + } + final, err := m.Complete(context.Background(), user, snap.ID) + if err != nil { + t.Fatalf("Complete: %v", err) + } + if final.State != StateCompleted || final.StorageProgress != 100 { + t.Fatalf("final snapshot = %+v", final) + } + if !bytes.Equal(<-got, data) { + t.Fatal("driver received different bytes") + } + if _, err := m.Status(user, snap.ID); !errors.Is(err, ErrSessionNotFound) { + t.Fatalf("session should be removed after Complete, got %v", err) + } +} + +func TestSessionRetriableRefill(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + user := testUser() + + attempts := 0 + got := make(chan []byte, 1) + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + attempts++ + if attempts == 1 { + buf := make([]byte, chunkSize) + if _, err := io.ReadFull(fs, buf); err != nil { + return err + } + return errors.New("transient storage hiccup") + } + b, err := io.ReadAll(fs) + if err != nil { + return err + } + got <- b + return nil + }) + + snap, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + sendChunk(t, m, user, snap.ID, data, 0, chunkSize) + failed := waitState(t, m, user, snap.ID, StateFailedRetriable) + if failed.Attempt != 1 { + t.Fatalf("attempt = %d, want 1", failed.Attempt) + } + if len(failed.Received) != 0 { + t.Fatalf("failed session must report nothing received, got %v", failed.Received) + } + + // chunks other than 0 are rejected until the client restarts the fill + if _, err := m.Chunk(user, snap.ID, 1, bytes.NewReader(chunkOf(data, 1, chunkSize))); err == nil { + t.Fatal("chunk 1 on failed_retriable session: expected error") + } + // re-fill from chunk 0 respawns the pipeline + sendChunk(t, m, user, snap.ID, data, 0, chunkSize) + sendChunk(t, m, user, snap.ID, data, 1, chunkSize) + final, err := m.Complete(context.Background(), user, snap.ID) + if err != nil { + t.Fatalf("Complete after refill: %v", err) + } + if final.State != StateCompleted || attempts != 2 { + t.Fatalf("state=%s attempts=%d, want completed/2", final.State, attempts) + } + if !bytes.Equal(<-got, data) { + t.Fatal("driver received different bytes after refill") + } +} + +func TestSessionPermanentFailure(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + data := genData(chunkSize) + 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) + }) + + snap, _, err := m.Init(initReq(user, chunkSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + waitState(t, m, user, snap.ID, StateFailedPermanent) + if _, err := m.Chunk(user, snap.ID, 0, bytes.NewReader(data)); err == nil { + t.Fatal("chunk on failed_permanent session: expected error") + } + if _, err := m.Complete(context.Background(), user, snap.ID); err == nil { + t.Fatal("Complete on failed_permanent session: expected error") + } +} + +func TestRefillCRCMismatch(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + user := testUser() + + attempts := 0 + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + attempts++ + if attempts == 1 { + buf := make([]byte, chunkSize) + if _, err := io.ReadFull(fs, buf); err != nil { + return err + } + return errors.New("transient") + } + _, err := io.ReadAll(fs) + return err + }) + + snap, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + sendChunk(t, m, user, snap.ID, data, 0, chunkSize) + waitState(t, m, user, snap.ID, StateFailedRetriable) + + tampered := genData(chunkSize + 5)[:chunkSize] // different content, same length + if _, err := m.Chunk(user, snap.ID, 0, bytes.NewReader(tampered)); err == nil { + t.Fatal("re-fill with changed content: expected error") + } + waitState(t, m, user, snap.ID, StateFailedPermanent) +} + +func TestCompleteRefusesIncomplete(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(3 * chunkSize) + data := genData(totalSize) + user := testUser() + + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + _, err := io.ReadAll(fs) + return err + }) + + snap, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + sendChunk(t, m, user, snap.ID, data, 0, chunkSize) + if _, err := m.Complete(context.Background(), user, snap.ID); err == nil { + t.Fatal("Complete with missing chunks: expected error") + } + // unblock the pipeline goroutine before the test tears down + if err := m.Abort(user, snap.ID); err != nil { + t.Fatalf("Abort: %v", err) + } +} + +func TestAbortAndOwnership(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(4 * chunkSize) + data := genData(totalSize) + user := testUser() + + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + _, err := io.ReadAll(fs) + return err + }) + + snap, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + sendChunk(t, m, user, snap.ID, data, 0, chunkSize) + + stranger := &model.User{ID: 99, Username: "stranger"} + if _, err := m.Status(stranger, snap.ID); !errors.Is(err, ErrNotOwner) { + t.Fatalf("stranger Status err = %v, want ErrNotOwner", err) + } + if err := m.Abort(stranger, snap.ID); !errors.Is(err, ErrNotOwner) { + t.Fatalf("stranger Abort err = %v, want ErrNotOwner", err) + } + if err := m.Abort(user, snap.ID); err != nil { + t.Fatalf("Abort: %v", err) + } + if _, err := m.Status(user, snap.ID); !errors.Is(err, ErrSessionNotFound) { + t.Fatalf("Status after Abort err = %v, want ErrSessionNotFound", err) + } +} + +func TestExpiry(t *testing.T) { + m := setupSessionTest(t) + m.ttl = 30 * time.Millisecond + + const chunkSize = 1024 + user := testUser() + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + _, err := io.ReadAll(fs) + return err + }) + + snap, _, err := m.Init(initReq(user, 4*chunkSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + time.Sleep(60 * time.Millisecond) + m.gc() + if _, err := m.Status(user, snap.ID); !errors.Is(err, ErrSessionNotFound) { + t.Fatalf("Status after expiry err = %v, want ErrSessionNotFound", err) + } + // the terminated pipeline must release the ring file + deadline := time.Now().Add(5 * time.Second) + for { + entries, _ := os.ReadDir(m.dir()) + if len(entries) == 0 { + break + } + if time.Now().After(deadline) { + names := make([]string, 0, len(entries)) + for _, e := range entries { + names = append(names, e.Name()) + } + t.Fatalf("ring files not cleaned up after expiry: %v", names) + } + time.Sleep(5 * time.Millisecond) + } +} + +func TestInitResume(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(4 * chunkSize) + data := genData(totalSize) + user := testUser() + + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + _, err := io.ReadAll(fs) + return err + }) + + hashed := initReq(user, totalSize, chunkSize) + hashed.Hashes = map[*utils.HashType]string{utils.MD5: "0123456789abcdef0123456789abcdef"} + snap, _, err := m.Init(hashed) + if err != nil { + t.Fatalf("Init: %v", err) + } + sendChunk(t, m, user, snap.ID, data, 2, chunkSize) + + // same hashes prove the same file: resume with buffered chunks skippable + again, resumed, err := m.Init(hashed) + if err != nil || !resumed { + t.Fatalf("hashed re-Init = (resumed=%v, err=%v), want resumed", resumed, err) + } + if again.ID != snap.ID { + t.Fatalf("resumed session id = %s, want %s", again.ID, snap.ID) + } + if len(again.Received) != 1 || again.Received[0] != [2]int{2, 2} { + t.Fatalf("resumed received = %v, want [[2,2]]", again.Received) + } + + // a different size is a different upload + other, resumed, err := m.Init(initReq(user, totalSize+1, chunkSize)) + if err != nil || resumed || other.ID == snap.ID { + t.Fatalf("different-size Init = (id=%s, resumed=%v, err=%v)", other.ID, resumed, err) + } + + if _, err := m.Find(user, "/local/test.bin", totalSize); err != nil { + t.Fatalf("Find: %v", err) + } + if _, err := m.Find(user, "/local/nope.bin", totalSize); !errors.Is(err, ErrSessionNotFound) { + t.Fatalf("Find miss err = %v, want ErrSessionNotFound", err) + } + + // without hashes, path+size cannot prove identity against a receiving + // session holding buffered data — the old session must be dropped, or a + // same-sized different file would be silently mixed into the result + bare := initReq(user, totalSize, chunkSize) + fresh, resumed, err := m.Init(bare) + if err != nil || resumed || fresh.ID == snap.ID { + t.Fatalf("bare re-Init = (id=%s, resumed=%v, err=%v), want a fresh session", fresh.ID, resumed, err) + } + if _, err := m.Status(user, snap.ID); !errors.Is(err, ErrSessionNotFound) { + t.Fatalf("superseded session err = %v, want ErrSessionNotFound", err) + } +} + +func TestResumeFailedRetriableWithoutHash(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + user := testUser() + + attempts := 0 + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + attempts++ + if attempts == 1 { + buf := make([]byte, chunkSize) + if _, err := io.ReadFull(fs, buf); err != nil { + return err + } + return errors.New("transient") + } + _, err := io.ReadAll(fs) + return err + }) + + snap, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + sendChunk(t, m, user, snap.ID, data, 0, chunkSize) + waitState(t, m, user, snap.ID, StateFailedRetriable) + + // a failed_retriable session keeps no chunk data, so identity proof is not + // required to resume: the retry-button flow works without rapid hashing, + // and the re-fill CRC check still catches a changed file + again, resumed, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil || !resumed || again.ID != snap.ID { + t.Fatalf("re-Init on failed_retriable = (id=%s, resumed=%v, err=%v), want resumed same session", again.ID, resumed, err) + } + sendChunk(t, m, user, snap.ID, data, 0, chunkSize) + sendChunk(t, m, user, snap.ID, data, 1, chunkSize) + if _, err := m.Complete(context.Background(), user, snap.ID); err != nil { + t.Fatalf("Complete after hashless refill: %v", err) + } +} + +func TestRapidUploadShortCircuit(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(6 * chunkSize) + data := genData(totalSize) + user := testUser() + + wantMD5 := "0123456789abcdef0123456789abcdef" + wantSHA1 := "da39a3ee5e6b4b0d3255bfef95601890afd80709" + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + // the hash provided at init must reach the driver through the stream, + // that is what lets PutRapid-style drivers skip the transfer entirely + if got := fs.GetHash().GetHash(utils.MD5); got != wantMD5 { + return fmt.Errorf("md5 not propagated: %q", got) + } + if got := fs.GetHash().GetHash(utils.SHA1); got != wantSHA1 { + return fmt.Errorf("sha1 not propagated: %q", got) + } + time.Sleep(50 * time.Millisecond) // simulated rapid-upload API round trip + up(100) + return nil // rapid upload hit: succeed without reading the stream + }) + + req := initReq(user, totalSize, chunkSize) + req.Hashes = map[*utils.HashType]string{utils.MD5: wantMD5, utils.SHA1: wantSHA1} + snap, _, err := m.Init(req) + if err != nil { + t.Fatalf("Init: %v", err) + } + + // spam chunks across the completion moment: some land while receiving, + // some race the window close, some arrive after completion — with the + // completed-session absorption none of them may surface an error + var wg sync.WaitGroup + errCh := make(chan error, 64) + for w := 0; w < 4; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for round := 0; round < 8; round++ { + idx := (w*8 + round) % 6 + _, err := m.Chunk(user, snap.ID, idx, bytes.NewReader(chunkOf(data, idx, chunkSize))) + // flow-control signals are part of the protocol, not failures + if err != nil && !errors.Is(err, ErrChunkInFlight) && !errors.Is(err, ErrOutOfWindow) { + errCh <- fmt.Errorf("worker %d chunk %d: %w", w, idx, err) + return + } + time.Sleep(5 * time.Millisecond) + } + }(w) + } + wg.Wait() + close(errCh) + for err := range errCh { + t.Error(err) + } + + final, err := m.Complete(context.Background(), user, snap.ID) + if err != nil { + t.Fatalf("Complete: %v", err) + } + if final.State != StateCompleted || final.StorageProgress != 100 { + t.Fatalf("final snapshot = %+v", final) + } +} + +// TestChunkRacesRapidCompletion pins the exact race the completed-session +// absorption exists for: a chunk request grabs the live window, stalls while +// receiving its body, the pipeline completes off the hash alone (rapid +// upload) and closes the window — the stalled chunk must then succeed +// idempotently instead of surfacing "window closed" to a client whose upload +// in fact just finished. +func TestChunkRacesRapidCompletion(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + user := testUser() + + proceed := make(chan struct{}) + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + <-proceed // the rapid-upload verdict arrives when the test says so + return nil + }) + + snap, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + + gate := make(chan struct{}) + type result struct { + snap SessionSnapshot + err error + } + resCh := make(chan result, 1) + go func() { + s, e := m.Chunk(user, snap.ID, 0, + &gatedReader{release: gate, inner: bytes.NewReader(chunkOf(data, 0, chunkSize))}) + resCh <- result{s, e} + }() + + // wait until the chunk writer holds slot 0 in the filling state + m.mu.Lock() + sess := m.byID[snap.ID] + m.mu.Unlock() + sess.mu.Lock() + win := sess.win + sess.mu.Unlock() + deadline := time.Now().Add(5 * time.Second) + for { + win.mu.Lock() + filling := win.slotState[0] == slotFilling + win.mu.Unlock() + if filling { + break + } + if time.Now().After(deadline) { + t.Fatal("chunk writer never reached the filling state") + } + time.Sleep(time.Millisecond) + } + + close(proceed) // rapid upload succeeds, window closes + waitState(t, m, user, snap.ID, StateCompleted) // completion recorded + close(gate) // stalled chunk finishes against the closed window + + res := <-resCh + if res.err != nil { + t.Fatalf("in-flight chunk across rapid completion must be absorbed, got: %v", res.err) + } + if res.snap.State != StateCompleted { + t.Fatalf("absorbed chunk snapshot state = %s, want completed", res.snap.State) + } +} + +// TestChunkDuringCompletionGap covers the moment between op.Put closing the +// window (its defer) and the verdict being recorded: a chunk hitting the +// closed window inside that gap must wait for the verdict and be absorbed, +// not bounce a "window closed" error at a client whose upload just succeeded. +func TestChunkDuringCompletionGap(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + user := testUser() + + windowClosed := make(chan struct{}) + allowVerdict := make(chan struct{}) + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + _ = fs.Close() // what op.Put's defer does before Put returns + close(windowClosed) + <-allowVerdict // hold the pipeline return open: this IS the gap + return nil + }) + + snap, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + <-windowClosed + + type result struct { + snap SessionSnapshot + err error + } + resCh := make(chan result, 1) + go func() { + s, e := m.Chunk(user, snap.ID, 0, bytes.NewReader(chunkOf(data, 0, chunkSize))) + resCh <- result{s, e} + }() + + select { + case r := <-resCh: + t.Fatalf("chunk inside the gap returned early with (%s, %v); it must wait for the verdict", r.snap.State, r.err) + case <-time.After(150 * time.Millisecond): + // still waiting on the verdict, as designed + } + + close(allowVerdict) + select { + case r := <-resCh: + if r.err != nil { + t.Fatalf("gap chunk must be absorbed after completion, got: %v", r.err) + } + if r.snap.State != StateCompleted { + t.Fatalf("gap chunk snapshot state = %s, want completed", r.snap.State) + } + case <-time.After(5 * time.Second): + t.Fatal("gap chunk never returned after the verdict") + } +} + +func TestInitRejectsBadSize(t *testing.T) { + m := setupSessionTest(t) + if _, _, err := m.Init(initReq(testUser(), 0, 1024)); err == nil { + t.Fatal("Init with size 0: expected error") + } +} diff --git a/internal/multipart/window.go b/internal/multipart/window.go new file mode 100644 index 0000000000..01e7267012 --- /dev/null +++ b/internal/multipart/window.go @@ -0,0 +1,377 @@ +package multipart + +import ( + "errors" + "fmt" + "hash/crc32" + "io" + "os" + "path/filepath" + "sync" + "time" + + "github.com/OpenListTeam/OpenList/v4/pkg/utils" +) + +const ( + slotFree uint8 = iota + slotFilling + slotReady +) + +var ( + // ErrClosed is the sticky error after Close; all pending and future reads/writes fail with it. + ErrClosed = errors.New("multipart upload window closed") + // ErrChunkInFlight means another request is uploading the same chunk right now. + ErrChunkInFlight = errors.New("chunk is being uploaded by another request") + // ErrOutOfWindow means the chunk is still too far ahead of the consumption + // frontier after waiting WindowWaitTimeout; the client should back off and + // resend it later (flow control, not a failure). + ErrOutOfWindow = errors.New("chunk is out of the receiving window") +) + +// WindowWaitTimeout bounds how long WriteChunk blocks waiting for its slot. +// Browsers cannot reliably read responses sent before the request body is +// consumed (they report a network error), so under backpressure it is far +// better to hold the request until a slot frees — the wait must just stay +// well below CDN request deadlines (Cloudflare: ~100s). Tests shrink this. +var WindowWaitTimeout = 10 * time.Second + +// Window reassembles concurrently uploaded chunks into a sequential stream. +// Chunks land in a ring file of slots*chunkSize bytes (chunk i -> slot i%slots), +// and Read serves bytes in order, blocking until the next needed chunk arrives. +// A chunk slot is released as soon as the reader crosses its boundary, so the +// disk footprint is bounded by slots*chunkSize regardless of the file size. +// +// WriteChunk is safe for concurrent use; Read must be called from a single +// goroutine (the same contract as the FileStreamer it backs). +type Window struct { + mu sync.Mutex + cond *sync.Cond + + f *os.File + path string + + chunkSize int64 + totalSize int64 + total int + slots int + + slotState []uint8 + slotChunk []int + readPos int64 + + crcs []uint32 + crcSet []bool + + err error +} + +// Snapshot describes the receiving state, used for status responses and resume. +type Snapshot struct { + // Frontier is the next chunk index to be consumed (== TotalChunks when the stream is fully consumed). + Frontier int + // ReadPos is the number of bytes already consumed by the pipeline. + ReadPos int64 + // ReceivedBytes is the number of payload bytes received from the client (consumed + buffered). + ReceivedBytes int64 + // Received holds inclusive ranges of chunk indexes the client does not need to resend. + Received [][2]int +} + +func NewWindow(dir, id string, chunkSize, totalSize int64, slots int) (*Window, error) { + if chunkSize <= 0 || totalSize <= 0 || slots <= 0 { + return nil, fmt.Errorf("invalid window params: chunkSize=%d totalSize=%d slots=%d", chunkSize, totalSize, slots) + } + if err := os.MkdirAll(dir, 0o700); err != nil { + return nil, err + } + path := filepath.Join(dir, id+".ring") + f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0o600) + if err != nil { + return nil, err + } + total := int((totalSize + chunkSize - 1) / chunkSize) + w := &Window{ + f: f, + path: path, + chunkSize: chunkSize, + totalSize: totalSize, + total: total, + slots: slots, + slotState: make([]uint8, slots), + slotChunk: make([]int, slots), + crcs: make([]uint32, total), + crcSet: make([]bool, total), + } + for i := range w.slotChunk { + w.slotChunk[i] = -1 + } + w.cond = sync.NewCond(&w.mu) + return w, nil +} + +func (w *Window) TotalChunks() int { return w.total } + +// ChunkLen returns the payload length of chunk idx (the last chunk may be short). +func (w *Window) ChunkLen(idx int) int64 { + if idx == w.total-1 { + return w.totalSize - int64(idx)*w.chunkSize + } + return w.chunkSize +} + +func stopTimer(t *time.Timer) { + if t != nil { + t.Stop() + } +} + +// frontier returns the next chunk index to be consumed. Callers must hold mu. +func (w *Window) frontier() int { + if w.readPos >= w.totalSize { + return w.total + } + return int(w.readPos / w.chunkSize) +} + +// WriteChunk reads exactly the chunk payload from r into the ring and returns its CRC32 (IEEE). +// Re-sending an already buffered or consumed chunk succeeds immediately without touching data. +func (w *Window) WriteChunk(idx int, r io.Reader) (uint32, error) { + w.mu.Lock() + if w.err != nil { + w.mu.Unlock() + return 0, w.err + } + if idx < 0 || idx >= w.total { + w.mu.Unlock() + return 0, fmt.Errorf("chunk index %d out of range [0,%d)", idx, w.total) + } + length := w.ChunkLen(idx) + slot := idx % w.slots + // Admission control with a bounded wait: instead of bouncing a chunk the + // moment its slot is busy, park the request until the reader frees the + // slot. Rejecting fast would answer before the request body is read, which + // browsers surface as a network error — so under backpressure, waiting IS + // the flow control. sync.Cond has no timed wait; a timer broadcast wakes + // the loop at the deadline. + deadline := time.Now().Add(WindowWaitTimeout) + var timer *time.Timer + for { + if w.err != nil { + w.mu.Unlock() + stopTimer(timer) + return 0, w.err + } + if int64(idx)*w.chunkSize+length <= w.readPos { + // already fully consumed + crc := w.crcs[idx] + w.mu.Unlock() + stopTimer(timer) + return crc, nil + } + if w.slotChunk[slot] == idx { + if w.slotState[slot] == slotReady { + crc := w.crcs[idx] + w.mu.Unlock() + stopTimer(timer) + return crc, nil + } + if w.slotState[slot] == slotFilling { + w.mu.Unlock() + stopTimer(timer) + return 0, ErrChunkInFlight + } + } + if w.slotState[slot] == slotFree && idx < w.frontier()+w.slots { + break // admissible + } + if !time.Now().Before(deadline) { + w.mu.Unlock() + stopTimer(timer) + return 0, ErrOutOfWindow + } + if timer == nil { + timer = time.AfterFunc(time.Until(deadline), func() { + w.mu.Lock() + w.cond.Broadcast() + w.mu.Unlock() + }) + } + w.cond.Wait() + } + stopTimer(timer) + w.slotState[slot] = slotFilling + w.slotChunk[slot] = idx + f := w.f + w.mu.Unlock() + + h := crc32.NewIEEE() + n, err := utils.CopyWithBufferN(io.NewOffsetWriter(f, int64(slot)*w.chunkSize), io.TeeReader(r, h), length) + if err == nil { + // the body must contain exactly one chunk + var b [1]byte + if m, _ := io.ReadFull(r, b[:]); m > 0 { + err = fmt.Errorf("chunk %d larger than expected %d bytes", idx, length) + } + } else { + err = fmt.Errorf("incomplete chunk %d: got %d of %d bytes: %w", idx, n, length, err) + } + + w.mu.Lock() + defer w.mu.Unlock() + if w.slotState[slot] != slotFilling || w.slotChunk[slot] != idx { + // the window was closed and reset the slot while we were writing + if w.err != nil { + return 0, w.err + } + return 0, ErrClosed + } + if err == nil && w.err != nil { + err = w.err + } + if err != nil { + w.slotState[slot] = slotFree + w.slotChunk[slot] = -1 + return 0, err + } + w.slotState[slot] = slotReady + w.crcs[idx] = h.Sum32() + w.crcSet[idx] = true + w.cond.Broadcast() + return w.crcs[idx], nil +} + +// Read serves the reassembled stream in order, blocking until the next chunk +// is available, the window is closed, or the stream ends (io.EOF). +func (w *Window) Read(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + w.mu.Lock() + for { + if w.err != nil { + w.mu.Unlock() + return 0, w.err + } + if w.readPos >= w.totalSize { + w.mu.Unlock() + return 0, io.EOF + } + cur := int(w.readPos / w.chunkSize) + slot := cur % w.slots + if w.slotState[slot] == slotReady && w.slotChunk[slot] == cur { + chunkStart := int64(cur) * w.chunkSize + chunkEnd := chunkStart + w.ChunkLen(cur) + n := int64(len(p)) + if avail := chunkEnd - w.readPos; n > avail { + n = avail + } + off := int64(slot)*w.chunkSize + (w.readPos - chunkStart) + f := w.f + w.mu.Unlock() + + read, err := f.ReadAt(p[:n], off) + + w.mu.Lock() + if read > 0 { + w.readPos += int64(read) + if w.readPos >= chunkEnd { + w.slotState[slot] = slotFree + w.slotChunk[slot] = -1 + w.cond.Broadcast() // writers may be parked waiting for this slot + } + } + sticky := w.err + w.mu.Unlock() + if read > 0 { + return read, nil + } + if sticky != nil { + return 0, sticky + } + if err == nil { + err = io.ErrUnexpectedEOF + } + return 0, err + } + w.cond.Wait() + } +} + +// Close makes all pending and future operations fail with ErrClosed and removes +// the ring file. It is invoked by op.Put via FileStream.Closers when the +// pipeline ends, and is safe to call multiple times. +func (w *Window) Close() error { + return w.CloseWithError(ErrClosed) +} + +// CloseWithError is Close with a caller-chosen sticky error. The session's +// abort path passes an error wrapping context.Canceled so that a driver woken +// up from a blocked Read treats the abort exactly like a canceled request +// (e.g. the local driver only removes partially written files in that case). +func (w *Window) CloseWithError(sticky error) error { + w.mu.Lock() + if w.err == nil { + w.err = sticky + } + f := w.f + w.f = nil + for i := range w.slotState { + w.slotState[i] = slotFree + w.slotChunk[i] = -1 + } + w.cond.Broadcast() + w.mu.Unlock() + if f == nil { + return nil + } + err := f.Close() + if rmErr := os.Remove(w.path); rmErr != nil && err == nil { + err = rmErr + } + return err +} + +// Snapshot reports the receiving state for status responses and resume discovery. +func (w *Window) Snapshot() Snapshot { + w.mu.Lock() + defer w.mu.Unlock() + snap := Snapshot{Frontier: w.frontier(), ReadPos: w.readPos, ReceivedBytes: w.readPos} + var ranges [][2]int + if snap.Frontier > 0 { + ranges = append(ranges, [2]int{0, snap.Frontier - 1}) + } + for idx := snap.Frontier; idx < snap.Frontier+w.slots && idx < w.total; idx++ { + slot := idx % w.slots + if w.slotState[slot] != slotReady || w.slotChunk[slot] != idx { + continue + } + start := int64(idx) * w.chunkSize + var consumed int64 + if w.readPos > start { + consumed = w.readPos - start + } + snap.ReceivedBytes += w.ChunkLen(idx) - consumed + if len(ranges) > 0 && ranges[len(ranges)-1][1] == idx-1 { + ranges[len(ranges)-1][1] = idx + } else { + ranges = append(ranges, [2]int{idx, idx}) + } + } + snap.Received = ranges + return snap +} + +// CRCs returns a copy of the per-chunk CRC32 table and which entries are set. +// It remains readable after Close, so the session can compare re-filled chunks +// against a previous attempt. +func (w *Window) CRCs() ([]uint32, []bool) { + w.mu.Lock() + defer w.mu.Unlock() + crcs := make([]uint32, len(w.crcs)) + set := make([]bool, len(w.crcSet)) + copy(crcs, w.crcs) + copy(set, w.crcSet) + return crcs, set +} diff --git a/internal/multipart/window_test.go b/internal/multipart/window_test.go new file mode 100644 index 0000000000..d368672c95 --- /dev/null +++ b/internal/multipart/window_test.go @@ -0,0 +1,527 @@ +package multipart + +import ( + "bytes" + "crypto/sha256" + "errors" + "hash/crc32" + "io" + "math/rand" + "sync" + "sync/atomic" + "testing" + "time" +) + +func genData(size int64) []byte { + data := make([]byte, size) + rnd := rand.New(rand.NewSource(size*7919 + 13)) + rnd.Read(data) + return data +} + +func newTestWindow(t *testing.T, chunkSize, totalSize int64, slots int) *Window { + t.Helper() + w, err := NewWindow(t.TempDir(), "test", chunkSize, totalSize, slots) + if err != nil { + t.Fatalf("NewWindow: %v", err) + } + t.Cleanup(func() { _ = w.Close() }) + return w +} + +func chunkOf(data []byte, idx int, chunkSize int64) []byte { + start := int64(idx) * chunkSize + end := start + chunkSize + if end > int64(len(data)) { + end = int64(len(data)) + } + return data[start:end] +} + +func writeChunkOK(t *testing.T, w *Window, data []byte, idx int) uint32 { + t.Helper() + crc, err := w.WriteChunk(idx, bytes.NewReader(chunkOf(data, idx, w.chunkSize))) + if err != nil { + t.Fatalf("WriteChunk(%d): %v", idx, err) + } + if want := crc32.ChecksumIEEE(chunkOf(data, idx, w.chunkSize)); crc != want { + t.Fatalf("WriteChunk(%d) crc = %08x, want %08x", idx, crc, want) + } + return crc +} + +// readAllWithin reads the whole stream in a goroutine and fails the test on timeout, +// so a reassembly bug cannot hang the suite. +func readAllWithin(t *testing.T, w *Window, timeout time.Duration) []byte { + t.Helper() + type result struct { + data []byte + err error + } + ch := make(chan result, 1) + go func() { + data, err := io.ReadAll(w) + ch <- result{data, err} + }() + select { + case res := <-ch: + if res.err != nil { + t.Fatalf("ReadAll: %v", res.err) + } + return res.data + case <-time.After(timeout): + t.Fatal("ReadAll timed out") + return nil + } +} + +func TestSequentialReadWrite(t *testing.T) { + const chunkSize = 64 * 1024 + totalSize := int64(4*chunkSize + 32*1024) // last chunk is short + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 8) + + go func() { + for i := 0; i < w.TotalChunks(); i++ { + if _, err := w.WriteChunk(i, bytes.NewReader(chunkOf(data, i, chunkSize))); err != nil { + t.Errorf("WriteChunk(%d): %v", i, err) + return + } + time.Sleep(time.Millisecond) + } + }() + + got := readAllWithin(t, w, 10*time.Second) + if !bytes.Equal(got, data) { + t.Fatalf("reassembled stream differs: got %d bytes, want %d", len(got), len(data)) + } +} + +func TestOutOfOrderWrites(t *testing.T) { + const chunkSize = 16 * 1024 + totalSize := int64(5*chunkSize - 100) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 8) + + for _, idx := range []int{3, 0, 4, 2, 1} { + writeChunkOK(t, w, data, idx) + } + got := readAllWithin(t, w, 10*time.Second) + if !bytes.Equal(got, data) { + t.Fatal("reassembled stream differs after out-of-order writes") + } +} + +func TestConcurrentWriters(t *testing.T) { + const chunkSize = 32 * 1024 + totalSize := int64(8 * chunkSize) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 8) + + var wg sync.WaitGroup + for i := 0; i < w.TotalChunks(); i++ { + wg.Add(1) + go func(idx int) { + defer wg.Done() + if _, err := w.WriteChunk(idx, bytes.NewReader(chunkOf(data, idx, chunkSize))); err != nil { + t.Errorf("WriteChunk(%d): %v", idx, err) + } + }(i) + } + got := readAllWithin(t, w, 10*time.Second) + wg.Wait() + if !bytes.Equal(got, data) { + t.Fatal("reassembled stream differs after concurrent writes") + } +} + +func setWaitTimeout(t *testing.T, d time.Duration) { + t.Helper() + old := WindowWaitTimeout + WindowWaitTimeout = d + t.Cleanup(func() { WindowWaitTimeout = old }) +} + +func TestBackpressure(t *testing.T) { + setWaitTimeout(t, 50*time.Millisecond) // assert the post-deadline rejection + const chunkSize = 1024 + totalSize := int64(5 * chunkSize) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 2) + + if _, err := w.WriteChunk(2, bytes.NewReader(chunkOf(data, 2, chunkSize))); !errors.Is(err, ErrOutOfWindow) { + t.Fatalf("chunk 2 with frontier 0: err = %v, want ErrOutOfWindow", err) + } + writeChunkOK(t, w, data, 0) + writeChunkOK(t, w, data, 1) + // both slots occupied, frontier still 0 + if _, err := w.WriteChunk(2, bytes.NewReader(chunkOf(data, 2, chunkSize))); !errors.Is(err, ErrOutOfWindow) { + t.Fatalf("chunk 2 with full window: err = %v, want ErrOutOfWindow", err) + } + // consume chunk 0 -> slot released, frontier advances + buf := make([]byte, chunkSize) + if _, err := io.ReadFull(w, buf); err != nil { + t.Fatalf("ReadFull chunk 0: %v", err) + } + if !bytes.Equal(buf, chunkOf(data, 0, chunkSize)) { + t.Fatal("chunk 0 content differs") + } + writeChunkOK(t, w, data, 2) + // chunk 3 maps to the slot still holding buffered chunk 1 + if _, err := w.WriteChunk(3, bytes.NewReader(chunkOf(data, 3, chunkSize))); !errors.Is(err, ErrOutOfWindow) { + t.Fatalf("chunk 3 with occupied slot: err = %v, want ErrOutOfWindow", err) + } +} + +// TestWriteChunkWaitsForSlot pins the browser-friendly flow control: a chunk +// whose slot is occupied parks until the reader frees it instead of bouncing +// with an immediate rejection (early responses read as network errors in +// browsers). +func TestWriteChunkWaitsForSlot(t *testing.T) { + setWaitTimeout(t, 5*time.Second) + const chunkSize = 1024 + totalSize := int64(5 * chunkSize) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 2) + + writeChunkOK(t, w, data, 0) + writeChunkOK(t, w, data, 1) + + done := make(chan error, 1) + go func() { + _, err := w.WriteChunk(2, bytes.NewReader(chunkOf(data, 2, chunkSize))) + done <- err + }() + select { + case err := <-done: + t.Fatalf("chunk 2 should be parked while the window is full, returned %v", err) + case <-time.After(100 * time.Millisecond): + // parked, as designed + } + + // consuming chunk 0 frees its slot and must wake the parked writer + buf := make([]byte, chunkSize) + if _, err := io.ReadFull(w, buf); err != nil { + t.Fatalf("ReadFull: %v", err) + } + select { + case err := <-done: + if err != nil { + t.Fatalf("parked chunk after slot freed: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("parked chunk never admitted after its slot freed") + } +} + +func TestIdempotentResend(t *testing.T) { + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 4) + + first := writeChunkOK(t, w, data, 0) + again := writeChunkOK(t, w, data, 0) // buffered, not yet consumed + if first != again { + t.Fatalf("resend crc = %08x, want %08x", again, first) + } + buf := make([]byte, chunkSize) + if _, err := io.ReadFull(w, buf); err != nil { + t.Fatalf("ReadFull: %v", err) + } + consumed := writeChunkOK(t, w, data, 0) // already consumed + if consumed != first { + t.Fatalf("post-consume resend crc = %08x, want %08x", consumed, first) + } +} + +// gatedReader blocks the first Read until released, to hold a chunk in the filling state. +type gatedReader struct { + release <-chan struct{} + inner io.Reader +} + +func (g *gatedReader) Read(p []byte) (int, error) { + <-g.release + return g.inner.Read(p) +} + +func TestInFlightConflict(t *testing.T) { + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 4) + + release := make(chan struct{}) + done := make(chan error, 1) + go func() { + _, err := w.WriteChunk(0, &gatedReader{release: release, inner: bytes.NewReader(chunkOf(data, 0, chunkSize))}) + done <- err + }() + + // wait until the writer marked the slot as filling + deadline := time.Now().Add(5 * time.Second) + for { + w.mu.Lock() + filling := w.slotState[0] == slotFilling + w.mu.Unlock() + if filling { + break + } + if time.Now().After(deadline) { + t.Fatal("writer never reached filling state") + } + time.Sleep(time.Millisecond) + } + + if _, err := w.WriteChunk(0, bytes.NewReader(chunkOf(data, 0, chunkSize))); !errors.Is(err, ErrChunkInFlight) { + t.Fatalf("concurrent same-chunk write: err = %v, want ErrChunkInFlight", err) + } + close(release) + if err := <-done; err != nil { + t.Fatalf("gated WriteChunk: %v", err) + } + writeChunkOK(t, w, data, 0) // idempotent after settle +} + +func TestShortBodyRecovers(t *testing.T) { + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 4) + + if _, err := w.WriteChunk(0, bytes.NewReader(chunkOf(data, 0, chunkSize)[:100])); err == nil { + t.Fatal("short body: expected error") + } + writeChunkOK(t, w, data, 0) // slot must have been recycled +} + +func TestOversizeBodyRejected(t *testing.T) { + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 4) + + oversize := append(append([]byte{}, chunkOf(data, 0, chunkSize)...), 0xFF) + if _, err := w.WriteChunk(0, bytes.NewReader(oversize)); err == nil { + t.Fatal("oversize body: expected error") + } + writeChunkOK(t, w, data, 0) + + // the short last chunk must also reject a full-size body + last := w.TotalChunks() - 1 + if last == 0 { + t.Fatal("test needs at least 2 chunks") + } +} + +func TestLastChunkShortStrict(t *testing.T) { + const chunkSize = 1024 + totalSize := int64(chunkSize + 100) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 4) + + if _, err := w.WriteChunk(1, bytes.NewReader(genData(chunkSize))); err == nil { + t.Fatal("full-size body for short last chunk: expected error") + } + writeChunkOK(t, w, data, 1) + writeChunkOK(t, w, data, 0) + got := readAllWithin(t, w, 10*time.Second) + if !bytes.Equal(got, data) { + t.Fatal("reassembled stream differs") + } +} + +func TestCloseUnblocksReader(t *testing.T) { + w := newTestWindow(t, 1024, 4096, 4) + errCh := make(chan error, 1) + go func() { + buf := make([]byte, 16) + _, err := w.Read(buf) + errCh <- err + }() + time.Sleep(20 * time.Millisecond) + _ = w.Close() + select { + case err := <-errCh: + if !errors.Is(err, ErrClosed) { + t.Fatalf("blocked Read after Close: err = %v, want ErrClosed", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Read still blocked after Close") + } +} + +func TestCloseWithErrorPropagatesSticky(t *testing.T) { + w := newTestWindow(t, 1024, 4096, 4) + cause := errors.New("aborted for a reason") + errCh := make(chan error, 1) + go func() { + buf := make([]byte, 16) + _, err := w.Read(buf) + errCh <- err + }() + time.Sleep(20 * time.Millisecond) + _ = w.CloseWithError(cause) + select { + case err := <-errCh: + if !errors.Is(err, cause) { + t.Fatalf("blocked Read after CloseWithError: err = %v, want %v", err, cause) + } + case <-time.After(5 * time.Second): + t.Fatal("Read still blocked after CloseWithError") + } + if _, err := w.WriteChunk(0, bytes.NewReader(make([]byte, 1024))); !errors.Is(err, cause) { + t.Fatalf("WriteChunk after CloseWithError: err = %v, want %v", err, cause) + } +} + +func TestWriteAfterClose(t *testing.T) { + w := newTestWindow(t, 1024, 4096, 4) + _ = w.Close() + if _, err := w.WriteChunk(0, bytes.NewReader(make([]byte, 1024))); !errors.Is(err, ErrClosed) { + t.Fatalf("WriteChunk after Close: err = %v, want ErrClosed", err) + } +} + +func TestCloseUnblocksInFlightWriter(t *testing.T) { + const chunkSize = 1024 + data := genData(2 * chunkSize) + w := newTestWindow(t, chunkSize, 2*chunkSize, 4) + + release := make(chan struct{}) + done := make(chan error, 1) + go func() { + _, err := w.WriteChunk(0, &gatedReader{release: release, inner: bytes.NewReader(chunkOf(data, 0, chunkSize))}) + done <- err + }() + time.Sleep(20 * time.Millisecond) + _ = w.Close() + close(release) + if err := <-done; !errors.Is(err, ErrClosed) { + t.Fatalf("in-flight WriteChunk across Close: err = %v, want ErrClosed", err) + } +} + +func TestEOFExact(t *testing.T) { + const chunkSize = 1024 + totalSize := int64(chunkSize + 5) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 4) + writeChunkOK(t, w, data, 0) + writeChunkOK(t, w, data, 1) + + got := readAllWithin(t, w, 10*time.Second) + if !bytes.Equal(got, data) { + t.Fatal("reassembled stream differs") + } + buf := make([]byte, 1) + if n, err := w.Read(buf); n != 0 || err != io.EOF { + t.Fatalf("Read at EOF = (%d, %v), want (0, io.EOF)", n, err) + } +} + +func TestManyLapsSmallWindow(t *testing.T) { + const chunkSize = 8 * 1024 + const chunks = 64 + totalSize := int64(chunks*chunkSize - 777) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 3) + + var next atomic.Int64 + var wg sync.WaitGroup + for i := 0; i < 3; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + idx := int(next.Add(1) - 1) + if idx >= w.TotalChunks() { + return + } + for { + _, err := w.WriteChunk(idx, bytes.NewReader(chunkOf(data, idx, chunkSize))) + if err == nil { + break + } + if errors.Is(err, ErrOutOfWindow) || errors.Is(err, ErrChunkInFlight) { + time.Sleep(200 * time.Microsecond) + continue + } + t.Errorf("WriteChunk(%d): %v", idx, err) + return + } + } + }() + } + + got := readAllWithin(t, w, 30*time.Second) + wg.Wait() + if wantSum, gotSum := sha256.Sum256(data), sha256.Sum256(got); wantSum != gotSum { + t.Fatalf("reassembled stream differs: got %d bytes, want %d", len(got), len(data)) + } +} + +func TestSnapshot(t *testing.T) { + const chunkSize = 1024 + totalSize := int64(5*chunkSize + 512) + data := genData(totalSize) + w := newTestWindow(t, chunkSize, totalSize, 4) + + writeChunkOK(t, w, data, 0) + writeChunkOK(t, w, data, 1) + writeChunkOK(t, w, data, 3) + + snap := w.Snapshot() + if snap.Frontier != 0 || snap.ReadPos != 0 { + t.Fatalf("snapshot frontier/readPos = %d/%d, want 0/0", snap.Frontier, snap.ReadPos) + } + wantRanges := [][2]int{{0, 1}, {3, 3}} + if len(snap.Received) != len(wantRanges) || snap.Received[0] != wantRanges[0] || snap.Received[1] != wantRanges[1] { + t.Fatalf("snapshot received = %v, want %v", snap.Received, wantRanges) + } + if snap.ReceivedBytes != 3*chunkSize { + t.Fatalf("snapshot receivedBytes = %d, want %d", snap.ReceivedBytes, 3*chunkSize) + } + + buf := make([]byte, chunkSize) + if _, err := io.ReadFull(w, buf); err != nil { + t.Fatalf("ReadFull: %v", err) + } + snap = w.Snapshot() + if snap.Frontier != 1 || snap.ReadPos != chunkSize { + t.Fatalf("snapshot frontier/readPos = %d/%d, want 1/%d", snap.Frontier, snap.ReadPos, chunkSize) + } + if len(snap.Received) != 2 || snap.Received[0] != [2]int{0, 1} || snap.Received[1] != [2]int{3, 3} { + t.Fatalf("snapshot received = %v, want [[0,1],[3,3]]", snap.Received) + } + if snap.ReceivedBytes != 3*chunkSize { + t.Fatalf("snapshot receivedBytes = %d, want %d", snap.ReceivedBytes, 3*chunkSize) + } +} + +func TestIndexOutOfRange(t *testing.T) { + w := newTestWindow(t, 1024, 4096, 4) + if _, err := w.WriteChunk(-1, bytes.NewReader(nil)); err == nil { + t.Fatal("negative index: expected error") + } + if _, err := w.WriteChunk(4, bytes.NewReader(nil)); err == nil { + t.Fatal("index == total: expected error") + } +} + +func TestCRCsSurviveClose(t *testing.T) { + const chunkSize = 1024 + data := genData(2 * chunkSize) + w := newTestWindow(t, chunkSize, 2*chunkSize, 4) + want := writeChunkOK(t, w, data, 0) + _ = w.Close() + crcs, set := w.CRCs() + if !set[0] || crcs[0] != want { + t.Fatalf("CRCs after Close = (%08x, %v), want (%08x, true)", crcs[0], set[0], want) + } + if set[1] { + t.Fatal("chunk 1 crc should not be set") + } +} diff --git a/internal/op/hook.go b/internal/op/hook.go index 5cf01730d2..3d8530f933 100644 --- a/internal/op/hook.go +++ b/internal/op/hook.go @@ -2,7 +2,9 @@ package op import ( "context" + "fmt" "regexp" + "strconv" "strings" "github.com/OpenListTeam/OpenList/v4/internal/conf" @@ -83,6 +85,17 @@ var settingItemHooks = map[string]SettingItemHook{ conf.SlicesMap[conf.IgnoreDirectLinkParams] = strings.Split(item.Value, ",") return nil }, + conf.MultipartChunkSize: func(item *model.SettingItem) error { + size, err := strconv.Atoi(strings.TrimSpace(item.Value)) + if err != nil || size < 1 { + // deliberately a plain error: SaveSettings formats hook errors + // with %+v, which would dump a full stack trace into the UI + // notification for stack-carrying errors + return fmt.Errorf("multipart chunk size must be a positive integer (MB), got %q", item.Value) + } + item.Value = strconv.Itoa(size) + return nil + }, } func RegisterSettingItemHook(key string, hook SettingItemHook) { diff --git a/server/handles/multipart.go b/server/handles/multipart.go new file mode 100644 index 0000000000..991a102bcf --- /dev/null +++ b/server/handles/multipart.go @@ -0,0 +1,232 @@ +package handles + +import ( + "errors" + "io" + "net/url" + stdpath "path" + "strconv" + + "github.com/OpenListTeam/OpenList/v4/internal/conf" + "github.com/OpenListTeam/OpenList/v4/internal/errs" + "github.com/OpenListTeam/OpenList/v4/internal/fs" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/OpenList/v4/internal/multipart" + "github.com/OpenListTeam/OpenList/v4/internal/op" + "github.com/OpenListTeam/OpenList/v4/internal/setting" + "github.com/OpenListTeam/OpenList/v4/pkg/utils" + "github.com/OpenListTeam/OpenList/v4/server/common" + "github.com/gin-gonic/gin" +) + +const multipartMinChunkSize = int64(1) << 20 // 1MB + +// 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 +// a larger one — the server buffers a window of several chunks per session, so +// an unbounded client suggestion would translate directly into server-side +// disk usage. +func multipartChunkSize(requested int64) int64 { + ceiling := int64(setting.GetInt(conf.MultipartChunkSize, 10)) << 20 + if ceiling < multipartMinChunkSize { + ceiling = multipartMinChunkSize + } + size := ceiling + if requested > 0 && requested < ceiling { + size = max(requested, multipartMinChunkSize) + } + return size +} + +type MultipartInitResp struct { + multipart.SessionSnapshot + Resumed bool `json:"resumed"` +} + +// MultipartInit creates (or resumes) a multipart upload session and starts its +// upload pipeline. Headers mirror FsStream (fsup.go). +func MultipartInit(c *gin.Context) { + if !setting.GetBool(conf.MultipartEnabled) { + common.ErrorStrResp(c, "multipart upload is disabled", 403) + return + } + path := c.GetHeader("File-Path") + path, err := url.PathUnescape(path) + if err != nil { + common.ErrorResp(c, err, 400) + return + } + user := c.Request.Context().Value(conf.UserKey).(*model.User) + path, err = user.JoinPath(path) + if err != nil { + common.ErrorResp(c, err, 403) + return + } + size, err := strconv.ParseInt(c.GetHeader("X-File-Size"), 10, 64) + if err != nil { + common.ErrorStrResp(c, "multipart upload requires a valid X-File-Size header", 400) + return + } + if size <= 0 { + common.ErrorStrResp(c, "multipart upload requires a positive X-File-Size; upload empty files via /fs/put", 400) + return + } + var requestedChunkSize int64 + if v := c.GetHeader("X-Chunk-Size"); v != "" { + requestedChunkSize, err = strconv.ParseInt(v, 10, 64) + if err != nil { + common.ErrorResp(c, err, 400) + return + } + } + 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) + return + } + // fail fast on unusable destinations instead of letting the pipeline discover it + storage, _, err := op.GetStorageAndActualPath(dir) + if err != nil { + common.ErrorResp(c, err, 500) + return + } + if storage.Config().NoUpload { + common.ErrorResp(c, errs.UploadNotSupported, 405) + return + } + h := make(map[*utils.HashType]string) + if md5 := c.GetHeader("X-File-Md5"); md5 != "" { + h[utils.MD5] = md5 + } + if sha1 := c.GetHeader("X-File-Sha1"); sha1 != "" { + h[utils.SHA1] = sha1 + } + if sha256 := c.GetHeader("X-File-Sha256"); sha256 != "" { + h[utils.SHA256] = sha256 + } + mimetype := c.GetHeader("Content-Type") + if len(mimetype) == 0 { + mimetype = utils.GetMimeType(name) + } + snap, resumed, err := multipart.DefaultManager.Init(multipart.InitReq{ + User: user, + Path: path, + Size: size, + ChunkSize: multipartChunkSize(requestedChunkSize), + Mimetype: mimetype, + Modified: getLastModified(c), + Hashes: h, + }) + if err != nil { + common.ErrorResp(c, err, 500) + return + } + common.SuccessResp(c, MultipartInitResp{SessionSnapshot: snap, Resumed: resumed}) +} + +// MultipartChunk ingests one chunk. Chunks are idempotent and may be sent +// concurrently and out of order within the receiving window. +// code 429 = window full (flow control, retry after a short delay), +// code 409 = the same chunk is already in flight on another connection. +func MultipartChunk(c *gin.Context) { + user := c.Request.Context().Value(conf.UserKey).(*model.User) + id := c.GetHeader("X-Upload-Id") + idx, err := strconv.Atoi(c.GetHeader("X-Chunk-Index")) + if err != nil { + common.ErrorStrResp(c, "invalid X-Chunk-Index header", 400) + return + } + snap, err := multipart.DefaultManager.Chunk(user, id, idx, c.Request.Body) + // Answer only after the request body is consumed — on EVERY path. Flow + // control (429), absorbed chunks and validation errors would otherwise + // respond while the browser is still streaming the body, which it reports + // as a network error and which poisons its connection pool. A rejected + // chunk gets resent anyway, so draining costs no extra round trip. The + // drain is bounded so a malformed request cannot pin the handler. + limit := multipartChunkSize(0) + 64*1024 + if snap.ChunkSize > 0 { + limit = snap.ChunkSize + 64*1024 + } + _, _ = utils.CopyWithBuffer(io.Discard, io.LimitReader(c.Request.Body, limit)) + if err != nil { + common.ErrorWithDataResp(c, err, multipartErrCode(err), snap) + return + } + common.SuccessResp(c, snap) +} + +// MultipartComplete waits for the pipeline outcome and reports it, mirroring +// how /fs/put only responds once the driver upload finished. +func MultipartComplete(c *gin.Context) { + user := c.Request.Context().Value(conf.UserKey).(*model.User) + id := c.GetHeader("X-Upload-Id") + snap, err := multipart.DefaultManager.Complete(c.Request.Context(), user, id) + if err != nil { + common.ErrorWithDataResp(c, err, multipartErrCode(err), snap) + return + } + common.SuccessResp(c, snap) +} + +// MultipartStatus looks a session up by upload_id, or by path+size so an +// interrupted client can discover a resumable session. +func MultipartStatus(c *gin.Context) { + user := c.Request.Context().Value(conf.UserKey).(*model.User) + if id := c.Query("upload_id"); id != "" { + snap, err := multipart.DefaultManager.Status(user, id) + if err != nil { + common.ErrorResp(c, err, multipartErrCode(err)) + return + } + common.SuccessResp(c, snap) + return + } + path, err := user.JoinPath(c.Query("path")) + if err != nil { + common.ErrorResp(c, err, 403) + return + } + size, err := strconv.ParseInt(c.Query("size"), 10, 64) + if err != nil { + common.ErrorStrResp(c, "status lookup requires upload_id, or path and size", 400) + return + } + snap, err := multipart.DefaultManager.Find(user, path, size) + if err != nil { + common.ErrorResp(c, err, multipartErrCode(err)) + return + } + common.SuccessResp(c, snap) +} + +// MultipartAbort cancels the pipeline and discards the session. +func MultipartAbort(c *gin.Context) { + user := c.Request.Context().Value(conf.UserKey).(*model.User) + if err := multipart.DefaultManager.Abort(user, c.GetHeader("X-Upload-Id")); err != nil { + common.ErrorResp(c, err, multipartErrCode(err)) + return + } + common.SuccessResp(c) +} + +func multipartErrCode(err error) int { + switch { + case errors.Is(err, multipart.ErrOutOfWindow): + return 429 + case errors.Is(err, multipart.ErrChunkInFlight): + return 409 + case errors.Is(err, multipart.ErrSessionNotFound): + return 404 + case errors.Is(err, multipart.ErrNotOwner): + return 403 + default: + return 400 + } +} diff --git a/server/router.go b/server/router.go index 7330bd2c33..6bcb01e14c 100644 --- a/server/router.go +++ b/server/router.go @@ -4,6 +4,7 @@ import ( "github.com/OpenListTeam/OpenList/v4/cmd/flags" "github.com/OpenListTeam/OpenList/v4/internal/conf" "github.com/OpenListTeam/OpenList/v4/internal/message" + multipartPkg "github.com/OpenListTeam/OpenList/v4/internal/multipart" "github.com/OpenListTeam/OpenList/v4/internal/sign" "github.com/OpenListTeam/OpenList/v4/internal/stream" "github.com/OpenListTeam/OpenList/v4/pkg/utils" @@ -212,6 +213,13 @@ func _fs(g *gin.RouterGroup) { uploadLimiter := middlewares.UploadRateLimiter(stream.ClientUploadLimit) g.PUT("/put", middlewares.FsUp, uploadLimiter, handles.FsStream) g.PUT("/form", middlewares.FsUp, uploadLimiter, handles.FsForm) + multipartPkg.DefaultManager.StartGC() // reclaim ring files orphaned by a previous run + multipart := g.Group("/multipart") + multipart.POST("/init", middlewares.FsUp, handles.MultipartInit) + multipart.PUT("/chunk", uploadLimiter, handles.MultipartChunk) + multipart.POST("/complete", handles.MultipartComplete) + multipart.GET("/status", handles.MultipartStatus) + multipart.POST("/abort", handles.MultipartAbort) g.POST("/link", middlewares.AuthAdmin, handles.Link) // g.POST("/add_aria2", handles.AddOfflineDownload) // g.POST("/add_qbit", handles.AddQbittorrent)