From 195489421481e9784cd65492e1bf6e932ac9c397 Mon Sep 17 00:00:00 2001 From: thetapilla Date: Fri, 30 Oct 2026 14:44:55 +0800 Subject: [PATCH 1/4] fix(multipart): recover from driver panics in the upload pipeline - Run the storage upload behind a recover so a panicking driver fails the attempt (failed_retriable) instead of crashing the whole process. The pipeline goroutine sits outside gin's Recovery and the task workers' recover, so it was the one upload path without a safety net. - Log the recovered panic with its stack trace. - Add TestDriverPanicFailsTheAttempt. Co-Authored-By: Claude --- internal/multipart/session.go | 17 +++++++++++--- internal/multipart/session_test.go | 36 ++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 3 deletions(-) diff --git a/internal/multipart/session.go b/internal/multipart/session.go index e25d1d1c7e..f1eded260e 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" @@ -396,9 +397,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 } diff --git a/internal/multipart/session_test.go b/internal/multipart/session_test.go index daec7f8062..628cd7cb1b 100644 --- a/internal/multipart/session_test.go +++ b/internal/multipart/session_test.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "os" + "strings" "sync" "testing" "time" @@ -667,3 +668,38 @@ 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) + } +} From affb4b3c2eb762e584e70c778245a8880a3b63aa Mon Sep 17 00:00:00 2001 From: thetapilla Date: Fri, 30 Oct 2026 14:45:24 +0800 Subject: [PATCH 2/4] fix(multipart): wait for a superseded attempt before starting its replacement - When a new init supersedes a live session of the same path and size, wait (bounded by 10s) for the old pipeline attempt to return before starting the new one. Drivers that write the destination in place could otherwise have the old attempt's cancellation cleanup remove or interleave with the file the new attempt is writing. - Add TestSupersedeWaitsForTheOldAttempt. Co-Authored-By: Claude --- internal/multipart/session.go | 26 ++++++++++++++ internal/multipart/session_test.go | 57 ++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+) diff --git a/internal/multipart/session.go b/internal/multipart/session.go index f1eded260e..a00c2c395f 100644 --- a/internal/multipart/session.go +++ b/internal/multipart/session.go @@ -42,6 +42,10 @@ const ( // 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 ( @@ -98,6 +102,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. @@ -329,6 +351,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() } } diff --git a/internal/multipart/session_test.go b/internal/multipart/session_test.go index 628cd7cb1b..6f51176bdc 100644 --- a/internal/multipart/session_test.go +++ b/internal/multipart/session_test.go @@ -703,3 +703,60 @@ func TestDriverPanicFailsTheAttempt(t *testing.T) { 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) + } +} From bf39ff83e5e95d5434744e90e81c46785ff64e8d Mon Sep 17 00:00:00 2001 From: thetapilla Date: Fri, 30 Oct 2026 14:46:40 +0800 Subject: [PATCH 3/4] fix(multipart): report the storage error instead of a closed window - When the storage fails while a chunk is in flight, the chunk used to answer "multipart upload window closed" and the real cause only survived in data.error. Once the attempt's verdict is in, report it as "upload failed (): ", worded like /complete. - Return null data for error responses of unknown sessions (404/403) instead of an all-zero snapshot, matching the published API document. - Add TestChunkReportsStorageError and TestSnapshotDataOmitsUnknownSession. Co-Authored-By: Claude --- internal/multipart/session.go | 8 +++-- internal/multipart/session_test.go | 54 ++++++++++++++++++++++++++++++ server/handles/multipart.go | 13 +++++-- server/handles/multipart_test.go | 20 +++++++++++ 4 files changed, 91 insertions(+), 4 deletions(-) create mode 100644 server/handles/multipart_test.go diff --git a/internal/multipart/session.go b/internal/multipart/session.go index a00c2c395f..f35520918b 100644 --- a/internal/multipart/session.go +++ b/internal/multipart/session.go @@ -551,12 +551,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 6f51176bdc..339a40019c 100644 --- a/internal/multipart/session_test.go +++ b/internal/multipart/session_test.go @@ -760,3 +760,57 @@ func TestSupersedeWaitsForTheOldAttempt(t *testing.T) { 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) + } +} 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)) + } +} From e6ec5fd29a65f1744dbbc0729fdde1c09904bedc Mon Sep 17 00:00:00 2001 From: thetapilla Date: Fri, 30 Oct 2026 14:50:16 +0800 Subject: [PATCH 4/4] fix(multipart): release upload attempts left idle by the client - Track how long the pipeline's reader has been starved: blocked in Read waiting for a chunk while no chunk byte arrives. A slow client that keeps sending is not starved, and neither is a driver busy with the storage (backpressure), because the reader is not waiting then. - During GC, release an attempt starved for AttemptIdleTimeout (10 minutes). It stops holding the destination for the whole 30-minute session TTL, while the session keeps its metadata and resumes through the existing refill protocol (resend from chunk 0). - The session turns failed_retriable only once the released attempt has returned, so a refill never overlaps it. - Add TestWindowStarvedFor, TestIdleAttemptIsReleased, TestSlowClientIsNotReleased and TestBackpressureIsNotIdle. Co-Authored-By: Claude --- internal/multipart/session.go | 47 +++++++- internal/multipart/session_test.go | 173 +++++++++++++++++++++++++++++ internal/multipart/window.go | 49 +++++++- internal/multipart/window_test.go | 57 ++++++++++ 4 files changed, 323 insertions(+), 3 deletions(-) diff --git a/internal/multipart/session.go b/internal/multipart/session.go index f35520918b..ac48b931df 100644 --- a/internal/multipart/session.go +++ b/internal/multipart/session.go @@ -37,6 +37,12 @@ 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. @@ -54,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 @@ -95,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 @@ -257,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. @@ -296,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() @@ -406,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{ @@ -464,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 diff --git a/internal/multipart/session_test.go b/internal/multipart/session_test.go index 339a40019c..2e99b8c7eb 100644 --- a/internal/multipart/session_test.go +++ b/internal/multipart/session_test.go @@ -9,6 +9,7 @@ import ( "os" "strings" "sync" + "sync/atomic" "testing" "time" @@ -814,3 +815,175 @@ func TestChunkReportsStorageError(t *testing.T) { 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) + } +}