diff --git a/internal/multipart/session.go b/internal/multipart/session.go index e25d1d1c7e..ac48b931df 100644 --- a/internal/multipart/session.go +++ b/internal/multipart/session.go @@ -9,6 +9,7 @@ import ( "os" stdpath "path" "path/filepath" + "runtime/debug" "sync" "sync/atomic" "time" @@ -36,11 +37,21 @@ const ( // WindowSlots bounds the per-session disk footprint to WindowSlots*ChunkSize. var WindowSlots = 8 +// AttemptIdleTimeout releases a pipeline attempt whose driver has been starved +// of data that long: the client stopped sending, so the attempt stops holding +// the destination, while the session stays failed_retriable and resumable for +// the rest of its TTL. Tests shrink it. +var AttemptIdleTimeout = 10 * time.Minute + 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 + // supersedeJoinTimeout bounds the wait for a superseded attempt to return. + // Teardown is prompt (the context is canceled and the window closed); the + // bound only keeps a driver stuck on network I/O from blocking init. + supersedeJoinTimeout = 10 * time.Second ) var ( @@ -49,6 +60,8 @@ var ( // 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) + // errAttemptIdle marks an attempt released by AttemptIdleTimeout. + errAttemptIdle = errors.New("no data from the client, upload attempt released") ) // putFile is the pipeline tail: resolve the storage and run the regular upload @@ -90,6 +103,9 @@ type Session struct { cancel context.CancelFunc prevCRCs []uint32 prevSet []bool + // released is set when the attempt is let go for idleness; the attempt + // turns it into failed_retriable once it has actually returned + released error storagePct atomic.Uint64 // math.Float64bits of the driver progress (0-100) lastActive atomic.Int64 // unix nano @@ -97,6 +113,24 @@ type Session struct { func (s *Session) touch() { s.lastActive.Store(time.Now().UnixNano()) } +// awaitAttempt blocks until the running pipeline attempt has fully returned, +// or the timeout elapses. Callers must not hold Manager.mu. It is only a +// handover barrier: the attempt is already being torn down by terminate. +func (s *Session) awaitAttempt(timeout time.Duration) { + s.mu.Lock() + done := s.done + s.mu.Unlock() + if done == nil { + return + } + timer := time.NewTimer(timeout) + defer timer.Stop() + select { + case <-done: + case <-timer.C: + } +} + func (s *Session) setStoragePct(p float64) { s.storagePct.Store(math.Float64bits(p)) } // Snapshot is the wire representation of a session used by all endpoints. @@ -234,18 +268,28 @@ func (m *Manager) ensureGC() { } func (m *Manager) gc() { - deadline := time.Now().Add(-m.sessionTTL()).UnixNano() + now := time.Now() + deadline := now.Add(-m.sessionTTL()).UnixNano() m.mu.Lock() - var expired []*Session + var expired, live []*Session for _, s := range m.byID { if s.lastActive.Load() < deadline { expired = append(expired, s) + } else { + live = append(live, s) } } m.mu.Unlock() for _, s := range expired { m.terminate(s, errors.New("multipart upload session expired")) } + for _, s := range live { + s.mu.Lock() + if s.state == StateReceiving && s.released == nil && s.win != nil && s.win.StarvedFor(now) > AttemptIdleTimeout { + s.releaseIdleAttemptLocked() + } + s.mu.Unlock() + } } // terminate aborts a session (if still receiving) and drops it from the maps. @@ -273,6 +317,21 @@ func (s *Session) killAttemptLocked() { } } +// releaseIdleAttemptLocked stops an attempt whose client went silent. The +// session only becomes failed_retriable (and refillable) once the attempt +// has returned, so a refill never overlaps the released attempt. The window +// error wraps context.Canceled, so the driver cleans up as it does for a +// canceled request. The caller must hold s.mu. +func (s *Session) releaseIdleAttemptLocked() { + s.released = fmt.Errorf("%w after %s: %w", errAttemptIdle, AttemptIdleTimeout, context.Canceled) + if s.cancel != nil { + s.cancel() + } + if s.win != nil { + _ = s.win.CloseWithError(s.released) + } +} + func (m *Manager) remove(s *Session) { m.mu.Lock() defer m.mu.Unlock() @@ -328,6 +387,10 @@ func (m *Manager) Init(req InitReq) (SessionSnapshot, bool, error) { // 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")) + // wait for the old attempt to let go of the destination before a + // new one starts writing it: two drivers writing the same path + // interleave their output + s.awaitAttempt(supersedeJoinTimeout) m.mu.Lock() } } @@ -379,6 +442,7 @@ func (m *Manager) startAttemptLocked(s *Session) error { s.done = done s.state = StateReceiving s.err = nil + s.released = nil s.setStoragePct(0) fileStream := &stream.FileStream{ @@ -396,9 +460,19 @@ func (m *Manager) startAttemptLocked(s *Session) error { 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) + var err error + defer func() { + // the driver runs on this goroutine, outside gin's Recovery and the + // task workers' recover: a panicking driver must fail the upload, + // not take the whole server down + if r := recover(); r != nil { + utils.Log.Errorf("multipart upload pipeline panicked: %v\n%s", r, debug.Stack()) + err = fmt.Errorf("upload pipeline panicked: %v", r) + } + s.finishAttempt(win, err) + close(done) + }() + err = put(ctx, dstDir, fileStream, s.setStoragePct) }() return nil } @@ -427,6 +501,12 @@ func (s *Session) finishAttempt(win *Window, err error) { if s.err == nil { s.err = err } + case s.released != nil: + // released for idleness: retriable whatever the driver made of the + // closed stream, and with a reason the client can act on + s.state = StateFailedRetriable + s.err = s.released + s.attempt++ case isPermanentPutError(err): s.state = StateFailedPermanent s.err = err @@ -514,12 +594,16 @@ func (m *Manager) Chunk(user *model.User, id string, idx int, body io.Reader) (S } } s.mu.Lock() - completed := s.state == StateCompleted + st, cause := s.state, s.err snap := s.snapshotLocked() s.mu.Unlock() - if completed { + if st == StateCompleted { return snap, nil } + if st != StateReceiving && cause != nil { + // the window only closed because the attempt ended; say why + return snap, fmt.Errorf("upload failed (%s): %w", st, cause) + } return snap, err } if hasPrev && crc != prevCRC { diff --git a/internal/multipart/session_test.go b/internal/multipart/session_test.go index daec7f8062..2e99b8c7eb 100644 --- a/internal/multipart/session_test.go +++ b/internal/multipart/session_test.go @@ -7,7 +7,9 @@ import ( "fmt" "io" "os" + "strings" "sync" + "sync/atomic" "testing" "time" @@ -667,3 +669,321 @@ func TestInitRejectsBadSize(t *testing.T) { t.Fatal("Init with size 0: expected error") } } + +// A panicking driver must fail its upload attempt, not crash the server: the +// pipeline runs the driver on its own goroutine, outside gin's Recovery and +// the task workers' recover (see #3201 for such a driver panic). +func TestDriverPanicFailsTheAttempt(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + user := testUser() + + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + if _, err := io.ReadAll(fs); err != nil { + return err + } + var result *struct{ FileID string } // the driver got a nil upload result + return fmt.Errorf("uploaded %s", result.FileID) + }) + + 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) + sendChunk(t, m, user, snap.ID, data, 1, chunkSize) + + final, err := m.Complete(context.Background(), user, snap.ID) + if err == nil || !strings.Contains(err.Error(), "upload pipeline panicked") { + t.Fatalf("Complete err = %v, want the recovered panic", err) + } + if final.State != StateFailedRetriable { + t.Fatalf("state = %s, want %s", final.State, StateFailedRetriable) + } +} + +// Superseding must hand the destination over: the old attempt has to return +// before its replacement starts, or two drivers write the same path at once. +func TestSupersedeWaitsForTheOldAttempt(t *testing.T) { + m := setupSessionTest(t) + const chunkSize = 1024 + totalSize := int64(4 * chunkSize) + user := testUser() + + var mu sync.Mutex + var events []string + running := make(chan struct{}) + replacementStarted := make(chan struct{}) + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + mu.Lock() + first := len(events) == 0 + events = append(events, "start") + mu.Unlock() + if !first { + close(replacementStarted) + <-ctx.Done() + return ctx.Err() + } + close(running) + <-ctx.Done() + // a real driver needs a moment to unwind before it lets go of the + // destination; the handover must cover that window + time.Sleep(50 * time.Millisecond) + mu.Lock() + events = append(events, "old-returned") + mu.Unlock() + return ctx.Err() + }) + + if _, _, err := m.Init(initReq(user, totalSize, chunkSize)); err != nil { + t.Fatalf("Init: %v", err) + } + <-running + // same user, path and size without hashes: the old session is superseded + again, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("superseding Init: %v", err) + } + t.Cleanup(func() { _ = m.Abort(user, again.ID) }) + select { + case <-replacementStarted: + case <-time.After(5 * time.Second): + t.Fatal("the replacement attempt never started") + } + + mu.Lock() + defer mu.Unlock() + if len(events) != 3 || events[1] != "old-returned" { + t.Fatalf("events = %v, want the superseded attempt to return before its replacement starts", events) + } +} + +// When the storage fails, a chunk caught in flight by the closing window must +// report the storage's error, not the generic "window closed" (#3309). +func TestChunkReportsStorageError(t *testing.T) { + m := setupSessionTest(t) + oldSlots := WindowSlots + WindowSlots = 1 + t.Cleanup(func() { WindowSlots = oldSlots }) + const chunkSize = 1024 + totalSize := int64(4 * chunkSize) + data := genData(totalSize) + user := testUser() + cause := errors.New("upload part 7 rejected by the storage") + + halfRead := make(chan struct{}) + fail := make(chan struct{}) + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + buf := make([]byte, chunkSize/2) + if _, err := io.ReadFull(fs, buf); err != nil { + return err + } + close(halfRead) + <-fail + return cause + }) + + 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) + <-halfRead + type result struct { + snap SessionSnapshot + err error + } + parked := make(chan result, 1) + go func() { + // chunk 0 still holds the only slot (half read), so chunk 1 waits + s, err := m.Chunk(user, snap.ID, 1, bytes.NewReader(chunkOf(data, 1, chunkSize))) + parked <- result{s, err} + }() + time.Sleep(200 * time.Millisecond) + close(fail) + + r := <-parked + if r.err == nil { + t.Fatal("chunk 1 succeeded on a failed upload") + } + if errors.Is(r.err, ErrClosed) || !strings.Contains(r.err.Error(), cause.Error()) { + t.Fatalf("chunk error = %q, want it to carry the storage error %q", r.err, cause) + } +} + +func setAttemptIdleTimeout(t *testing.T, d time.Duration) { + t.Helper() + old := AttemptIdleTimeout + AttemptIdleTimeout = d + t.Cleanup(func() { AttemptIdleTimeout = old }) +} + +// readAllPut is a driver that consumes the whole stream before succeeding. +func readAllPut(got chan<- []byte) func(context.Context, string, *stream.FileStream, driver.UpdateProgress) error { + return func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + b, err := io.ReadAll(fs) + if err != nil { + return err + } + if got != nil { + got <- b + } + return nil + } +} + +// An attempt whose client went silent is released after AttemptIdleTimeout, +// so it stops holding the destination for the whole session TTL; the session +// becomes failed_retriable and resumes through the refill protocol. The +// refill must not start before the released attempt has let go. +func TestIdleAttemptIsReleased(t *testing.T) { + m := setupSessionTest(t) + setAttemptIdleTimeout(t, 50*time.Millisecond) + const chunkSize = 1024 + totalSize := int64(3 * chunkSize) + data := genData(totalSize) + user := testUser() + got := make(chan []byte, 1) + var running, overlaps atomic.Int32 + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + if running.Add(1) > 1 { + overlaps.Add(1) + } + defer running.Add(-1) + b, err := io.ReadAll(fs) + if err != nil { + time.Sleep(50 * time.Millisecond) // a driver takes a moment to unwind + 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) + time.Sleep(100 * time.Millisecond) // the client goes silent + m.gc() + released := waitState(t, m, user, snap.ID, StateFailedRetriable) + if !strings.Contains(released.Error, "no data from the client") { + t.Fatalf("released session error = %q, want it to say why", released.Error) + } + + // the client comes back and refills from chunk 0 + again, resumed, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil || !resumed || again.ID != snap.ID { + t.Fatalf("re-Init = (id=%s, resumed=%v, err=%v), want the released session resumed", again.ID, resumed, err) + } + for idx := 0; idx < 3; idx++ { + sendChunk(t, m, user, snap.ID, data, idx, chunkSize) + } + final, err := m.Complete(context.Background(), user, snap.ID) + if err != nil || final.State != StateCompleted { + t.Fatalf("Complete after refill = (%s, %v), want completed", final.State, err) + } + if b := <-got; !bytes.Equal(b, data) { + t.Fatal("the refilled upload delivered different content") + } + if n := overlaps.Load(); n != 0 { + t.Fatalf("the refill attempt overlapped the released one %d time(s)", n) + } +} + +// tricklingReader yields one byte per read, every interval. +type tricklingReader struct { + data []byte + interval time.Duration +} + +func (r *tricklingReader) Read(p []byte) (int, error) { + if len(r.data) == 0 { + return 0, io.EOF + } + time.Sleep(r.interval) + p[0] = r.data[0] + r.data = r.data[1:] + return 1, nil +} + +// A slow client that keeps sending is not idle, however long one chunk takes. +func TestSlowClientIsNotReleased(t *testing.T) { + m := setupSessionTest(t) + setAttemptIdleTimeout(t, 50*time.Millisecond) + const chunkSize = 32 + totalSize := int64(2 * chunkSize) + data := genData(totalSize) + user := testUser() + stubPut(t, readAllPut(nil)) + + snap, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + sent := make(chan error, 1) + go func() { + _, err := m.Chunk(user, snap.ID, 0, &tricklingReader{data: chunkOf(data, 0, chunkSize), interval: 10 * time.Millisecond}) + sent <- err + }() + for i := 0; i < 12; i++ { // ~240ms of a 320ms trickle + time.Sleep(20 * time.Millisecond) + m.gc() + if s, err := m.Status(user, snap.ID); err != nil || s.State != StateReceiving { + t.Fatalf("a client that was still sending got released (state %s, err %v)", s.State, err) + } + } + if err := <-sent; err != nil { + t.Fatalf("trickled chunk: %v", err) + } + sendChunk(t, m, user, snap.ID, data, 1, chunkSize) + if _, err := m.Complete(context.Background(), user, snap.ID); err != nil { + t.Fatalf("Complete: %v", err) + } +} + +// A storage that is slow to consume is backpressure, not an idle client: +// chunks parked behind a full window must not get the attempt released. +func TestBackpressureIsNotIdle(t *testing.T) { + m := setupSessionTest(t) + setAttemptIdleTimeout(t, 50*time.Millisecond) + const chunkSize = 1024 + slots := WindowSlots + totalSize := int64((slots + 2) * chunkSize) + data := genData(totalSize) + user := testUser() + release := make(chan struct{}) + stubPut(t, func(ctx context.Context, dst string, fs *stream.FileStream, up driver.UpdateProgress) error { + defer fs.Close() + <-release // the storage is busy and not reading yet + _, err := io.ReadAll(fs) + return err + }) + + snap, _, err := m.Init(initReq(user, totalSize, chunkSize)) + if err != nil { + t.Fatalf("Init: %v", err) + } + for idx := 0; idx < slots; idx++ { + sendChunk(t, m, user, snap.ID, data, idx, chunkSize) + } + time.Sleep(100 * time.Millisecond) + m.gc() + if s, err := m.Status(user, snap.ID); err != nil || s.State != StateReceiving { + t.Fatalf("an attempt waiting on the storage got released (state %s, err %v)", s.State, err) + } + close(release) + for idx := slots; idx < slots+2; idx++ { + sendChunk(t, m, user, snap.ID, data, idx, chunkSize) + } + if _, err := m.Complete(context.Background(), user, snap.ID); err != nil { + t.Fatalf("Complete: %v", err) + } +} diff --git a/internal/multipart/window.go b/internal/multipart/window.go index 01e7267012..598fd65b98 100644 --- a/internal/multipart/window.go +++ b/internal/multipart/window.go @@ -8,6 +8,7 @@ import ( "os" "path/filepath" "sync" + "sync/atomic" "time" "github.com/OpenListTeam/OpenList/v4/pkg/utils" @@ -65,6 +66,12 @@ type Window struct { crcSet []bool err error + + // idle detection (see StarvedFor): waitSince is when Read began waiting + // for data and is zero while it is not waiting; lastFill is when the last + // chunk byte arrived, in unix nanoseconds + waitSince time.Time + lastFill atomic.Int64 } // Snapshot describes the receiving state, used for status responses and resume. @@ -207,7 +214,7 @@ func (w *Window) WriteChunk(idx int, r io.Reader) (uint32, error) { w.mu.Unlock() h := crc32.NewIEEE() - n, err := utils.CopyWithBufferN(io.NewOffsetWriter(f, int64(slot)*w.chunkSize), io.TeeReader(r, h), length) + n, err := utils.CopyWithBufferN(io.NewOffsetWriter(f, int64(slot)*w.chunkSize), io.TeeReader(fillReader{r, &w.lastFill}, h), length) if err == nil { // the body must contain exactly one chunk var b [1]byte @@ -251,10 +258,12 @@ func (w *Window) Read(p []byte) (int, error) { w.mu.Lock() for { if w.err != nil { + w.waitSince = time.Time{} w.mu.Unlock() return 0, w.err } if w.readPos >= w.totalSize { + w.waitSince = time.Time{} w.mu.Unlock() return 0, io.EOF } @@ -269,6 +278,7 @@ func (w *Window) Read(p []byte) (int, error) { } off := int64(slot)*w.chunkSize + (w.readPos - chunkStart) f := w.f + w.waitSince = time.Time{} w.mu.Unlock() read, err := f.ReadAt(p[:n], off) @@ -295,10 +305,47 @@ func (w *Window) Read(p []byte) (int, error) { } return 0, err } + if w.waitSince.IsZero() { + w.waitSince = time.Now() + } w.cond.Wait() } } +// fillReader records when the last chunk byte arrived (see StarvedFor). +type fillReader struct { + r io.Reader + last *atomic.Int64 +} + +func (f fillReader) Read(p []byte) (int, error) { + n, err := f.r.Read(p) + if n > 0 { + f.last.Store(time.Now().UnixNano()) + } + return n, err +} + +// StarvedFor reports how long the reader has been blocked waiting for data +// while no chunk byte arrived: the time since it began waiting or since the +// last byte came in, whichever is later. It is zero while the reader is not +// waiting — a driver busy with the storage is backpressure, not starvation. +func (w *Window) StarvedFor(now time.Time) time.Duration { + w.mu.Lock() + since := w.waitSince + w.mu.Unlock() + if since.IsZero() { + return 0 + } + if last := time.Unix(0, w.lastFill.Load()); last.After(since) { + since = last + } + if d := now.Sub(since); d > 0 { + return d + } + return 0 +} + // 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. diff --git a/internal/multipart/window_test.go b/internal/multipart/window_test.go index d368672c95..2c6e1ac5e9 100644 --- a/internal/multipart/window_test.go +++ b/internal/multipart/window_test.go @@ -525,3 +525,60 @@ func TestCRCsSurviveClose(t *testing.T) { t.Fatal("chunk 1 crc should not be set") } } + +// StarvedFor measures how long the reader has waited for data no client is +// sending: it starts when Read blocks, restarts with every byte that arrives, +// and is zero while the reader is not waiting. +func TestWindowStarvedFor(t *testing.T) { + const chunkSize = 64 + w := newTestWindow(t, chunkSize, 2*chunkSize, 2) + if d := w.StarvedFor(time.Now()); d != 0 { + t.Fatalf("StarvedFor before any read = %v, want 0", d) + } + read := make(chan struct{}) + go func() { + buf := make([]byte, chunkSize) + _, _ = io.ReadFull(w, buf) + close(read) + }() + waitFor(t, "the blocked reader to be reported as starved", func() bool { return w.StarvedFor(time.Now()) > 0 }) + if d := w.StarvedFor(time.Now().Add(time.Hour)); d < time.Hour { + t.Fatalf("StarvedFor an hour later = %v, want at least 1h", d) + } + + // a chunk trickling in keeps the reader fed: starvation restarts with + // every byte received + pr, pw := io.Pipe() + written := make(chan error, 1) + go func() { + _, err := w.WriteChunk(0, pr) + written <- err + }() + time.Sleep(30 * time.Millisecond) + if _, err := pw.Write([]byte{1}); err != nil { + t.Fatal(err) + } + waitFor(t, "starvation to restart after a byte arrived", func() bool { return w.StarvedFor(time.Now()) < 20*time.Millisecond }) + if _, err := pw.Write(make([]byte, chunkSize-1)); err != nil { + t.Fatal(err) + } + _ = pw.Close() + if err := <-written; err != nil { + t.Fatalf("WriteChunk: %v", err) + } + <-read + if d := w.StarvedFor(time.Now()); d != 0 { + t.Fatalf("StarvedFor while the reader is not waiting = %v, want 0", d) + } +} + +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(time.Millisecond) + } +} diff --git a/server/handles/multipart.go b/server/handles/multipart.go index 991a102bcf..e6745b4873 100644 --- a/server/handles/multipart.go +++ b/server/handles/multipart.go @@ -156,12 +156,21 @@ func MultipartChunk(c *gin.Context) { } _, _ = utils.CopyWithBuffer(io.Discard, io.LimitReader(c.Request.Body, limit)) if err != nil { - common.ErrorWithDataResp(c, err, multipartErrCode(err), snap) + common.ErrorWithDataResp(c, err, multipartErrCode(err), snapshotData(snap)) return } common.SuccessResp(c, snap) } +// snapshotData keeps error responses free of an all-zero snapshot: a session +// that was never found has nothing to report, so the envelope carries null. +func snapshotData(snap multipart.SessionSnapshot) any { + if snap.ID == "" { + return nil + } + return 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) { @@ -169,7 +178,7 @@ func MultipartComplete(c *gin.Context) { 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) + common.ErrorWithDataResp(c, err, multipartErrCode(err), snapshotData(snap)) return } common.SuccessResp(c, snap) diff --git a/server/handles/multipart_test.go b/server/handles/multipart_test.go new file mode 100644 index 0000000000..dc47c5c0aa --- /dev/null +++ b/server/handles/multipart_test.go @@ -0,0 +1,20 @@ +package handles + +import ( + "testing" + + "github.com/OpenListTeam/OpenList/v4/internal/multipart" +) + +// A session that was never found has nothing to report: its error response +// carries null data, as documented, rather than an all-zero snapshot. +func TestSnapshotDataOmitsUnknownSession(t *testing.T) { + if got := snapshotData(multipart.SessionSnapshot{}); got != nil { + t.Fatalf("snapshotData(zero snapshot) = %#v, want nil", got) + } + snap := multipart.SessionSnapshot{ID: "3f2c9e2a", State: multipart.StateFailedRetriable} + got, ok := snapshotData(snap).(multipart.SessionSnapshot) + if !ok || got.ID != snap.ID || got.State != snap.State { + t.Fatalf("snapshotData(known snapshot) = %#v, want the snapshot itself", snapshotData(snap)) + } +}