diff --git a/test/tus/tus_integration_test.go b/test/tus/tus_integration_test.go index 0576caa98..cf9456325 100644 --- a/test/tus/tus_integration_test.go +++ b/test/tus/tus_integration_test.go @@ -1,6 +1,7 @@ package tus import ( + "bufio" "bytes" "context" "encoding/base64" @@ -996,3 +997,109 @@ func TestTusAbortedPatchKeepsStoredChunks(t *testing.T) { require.NoError(t, err) assert.Equal(t, testData, body, "the resumed upload should read back whole") } + +// TestTusConcurrentPatchRefused checks that a PATCH sent while another PATCH +// on the same session is still consuming its body is refused with 423 Locked. +// Before the per-session claim, both were accepted at the same offset, recorded +// the range twice, and completion failed on the duplicate while HEAD reported +// the upload fully received - the file was never created and every stored byte +// became garbage when the session expired. +func TestTusConcurrentPatchRefused(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second) + defer cancel() + + cluster, err := startTestCluster(t, ctx) + require.NoError(t, err) + defer func() { + cluster.Stop() + os.RemoveAll(cluster.dataDir) + }() + + const subChunkSize = 4 * 1024 * 1024 + testData := make([]byte, 3*subChunkSize) + for i := range testData { + testData[i] = byte(i % 251) + } + targetPath := "/raced/video.bin" + client := &http.Client{} + + createReq, err := http.NewRequest(http.MethodPost, cluster.TusURL()+targetPath, nil) + require.NoError(t, err) + createReq.Header.Set("Tus-Resumable", TusVersion) + createReq.Header.Set("Upload-Length", strconv.Itoa(len(testData))) + + createResp, err := client.Do(createReq) + require.NoError(t, err) + createResp.Body.Close() + require.Equal(t, http.StatusCreated, createResp.StatusCode) + uploadLocation := createResp.Header.Get("Location") + + // PATCH A promises one sub-chunk and stalls mid-body, holding the session. + conn, err := net.Dial("tcp", "127.0.0.1:"+testFilerPort) + require.NoError(t, err) + defer conn.Close() + // bound the raw reads below so a filer that never answers fails here + require.NoError(t, conn.SetDeadline(time.Now().Add(60*time.Second))) + _, err = fmt.Fprintf(conn, "PATCH %s HTTP/1.1\r\nHost: 127.0.0.1:%s\r\nTus-Resumable: %s\r\nContent-Type: application/offset+octet-stream\r\nUpload-Offset: 0\r\nContent-Length: %d\r\n\r\n", + uploadLocation, testFilerPort, TusVersion, subChunkSize) + require.NoError(t, err) + _, err = conn.Write(testData[:1024*1024]) + require.NoError(t, err) + time.Sleep(2 * time.Second) + + // PATCH B is the client's retry of the same range while A is in flight. + retryReq, err := http.NewRequest(http.MethodPatch, cluster.FullURL(uploadLocation), bytes.NewReader(testData[:subChunkSize])) + require.NoError(t, err) + retryReq.Header.Set("Tus-Resumable", TusVersion) + retryReq.Header.Set("Upload-Offset", "0") + retryReq.Header.Set("Content-Type", "application/offset+octet-stream") + + retryResp, err := client.Do(retryReq) + require.NoError(t, err) + retryResp.Body.Close() + require.Equal(t, http.StatusLocked, retryResp.StatusCode, "a concurrent PATCH must be refused, not recorded twice") + + // Finish PATCH A and read its response. + _, err = conn.Write(testData[1024*1024 : subChunkSize]) + require.NoError(t, err) + respReader := bufio.NewReader(conn) + respA, err := http.ReadResponse(respReader, nil) + require.NoError(t, err) + respA.Body.Close() + require.Equal(t, http.StatusNoContent, respA.StatusCode) + + headReq, err := http.NewRequest(http.MethodHead, cluster.FullURL(uploadLocation), nil) + require.NoError(t, err) + headReq.Header.Set("Tus-Resumable", TusVersion) + + headResp, err := client.Do(headReq) + require.NoError(t, err) + headResp.Body.Close() + require.Equal(t, http.StatusOK, headResp.StatusCode) + currentOffset, err := strconv.Atoi(headResp.Header.Get("Upload-Offset")) + require.NoError(t, err) + require.Equal(t, subChunkSize, currentOffset, "only PATCH A's sub-chunk should be recorded") + + patchReq, err := http.NewRequest(http.MethodPatch, cluster.FullURL(uploadLocation), bytes.NewReader(testData[currentOffset:])) + require.NoError(t, err) + patchReq.Header.Set("Tus-Resumable", TusVersion) + patchReq.Header.Set("Upload-Offset", strconv.Itoa(currentOffset)) + patchReq.Header.Set("Content-Type", "application/offset+octet-stream") + + patchResp, err := client.Do(patchReq) + require.NoError(t, err) + patchResp.Body.Close() + require.Equal(t, http.StatusNoContent, patchResp.StatusCode) + + getResp, err := client.Get(cluster.FilerURL() + targetPath) + require.NoError(t, err) + defer getResp.Body.Close() + require.Equal(t, http.StatusOK, getResp.StatusCode) + body, err := io.ReadAll(getResp.Body) + require.NoError(t, err) + assert.Equal(t, testData, body, "the raced upload should complete with intact content") +} diff --git a/weed/filer/filer.go b/weed/filer/filer.go index a57f5eb3c..3654bf2b0 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -46,7 +46,7 @@ type Filer struct { UniqueFilerEpoch int32 Store VirtualFilerStore MasterClient *wdclient.MasterClient - fileIdDeletionQueue *util.UnboundedQueue + FileIdDeletionQueue *util.UnboundedQueue GrpcDialOption grpc.DialOption DirBucketsPath string Cipher bool @@ -74,7 +74,7 @@ type Filer struct { func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerHost pb.ServerAddress, filerGroup string, collection string, replication string, dataCenter string, maxFilenameLength uint32, notifyFn func()) *Filer { f := &Filer{ MasterClient: wdclient.NewMasterClient(grpcDialOption, filerGroup, cluster.FilerType, filerHost, dataCenter, "", masters), - fileIdDeletionQueue: util.NewUnboundedQueue(), + FileIdDeletionQueue: util.NewUnboundedQueue(), GrpcDialOption: grpcDialOption, FilerConf: NewFilerConf(), RemoteStorage: NewFilerRemoteStorage(), diff --git a/weed/filer/filer_deletion.go b/weed/filer/filer_deletion.go index 6d472d0a7..63a20251d 100644 --- a/weed/filer/filer_deletion.go +++ b/weed/filer/filer_deletion.go @@ -306,7 +306,7 @@ func (f *Filer) loopProcessingDeletion() { glog.V(0).Infof("deletion processor shutting down") return case <-ticker.C: - f.fileIdDeletionQueue.Consume(func(fileIds []string) { + f.FileIdDeletionQueue.Consume(func(fileIds []string) { for i := 0; i < len(fileIds); i += DeletionBatchSize { end := i + DeletionBatchSize if end > len(fileIds) { @@ -599,7 +599,7 @@ func (f *Filer) DeleteChunks(ctx context.Context, fullpath util.FullPath, chunks func (f *Filer) doDeleteChunks(ctx context.Context, chunks []*filer_pb.FileChunk) { for _, chunk := range chunks { if !chunk.IsChunkManifest { - f.fileIdDeletionQueue.EnQueue(chunk.GetFileIdString()) + f.FileIdDeletionQueue.EnQueue(chunk.GetFileIdString()) continue } dataChunks, manifestResolveErr := ResolveOneChunkManifest(ctx, f.MasterClient.LookupFileId, chunk) @@ -607,15 +607,15 @@ func (f *Filer) doDeleteChunks(ctx context.Context, chunks []*filer_pb.FileChunk glog.V(0).InfofCtx(ctx, "failed to resolve manifest %s: %v", chunk.FileId, manifestResolveErr) } for _, dChunk := range dataChunks { - f.fileIdDeletionQueue.EnQueue(dChunk.GetFileIdString()) + f.FileIdDeletionQueue.EnQueue(dChunk.GetFileIdString()) } - f.fileIdDeletionQueue.EnQueue(chunk.GetFileIdString()) + f.FileIdDeletionQueue.EnQueue(chunk.GetFileIdString()) } } func (f *Filer) DeleteChunksNotRecursive(chunks []*filer_pb.FileChunk) { for _, chunk := range chunks { - f.fileIdDeletionQueue.EnQueue(chunk.GetFileIdString()) + f.FileIdDeletionQueue.EnQueue(chunk.GetFileIdString()) } } diff --git a/weed/filer/filer_lazy_remote_test.go b/weed/filer/filer_lazy_remote_test.go index 673a2eaa2..7ac2f4f4d 100644 --- a/weed/filer/filer_lazy_remote_test.go +++ b/weed/filer/filer_lazy_remote_test.go @@ -277,7 +277,7 @@ func newTestFiler(t *testing.T, store *stubFilerStore, rs *FilerRemoteStorage) * FilerConf: NewFilerConf(), MaxFilenameLength: 255, MasterClient: mc, - fileIdDeletionQueue: util.NewUnboundedQueue(), + FileIdDeletionQueue: util.NewUnboundedQueue(), deletionQuit: make(chan struct{}), LocalMetaLogBuffer: log_buffer.NewLogBuffer("test", time.Minute, func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, func() {}), diff --git a/weed/operation/upload_chunked.go b/weed/operation/upload_chunked.go index 888ce6427..244a553f9 100644 --- a/weed/operation/upload_chunked.go +++ b/weed/operation/upload_chunked.go @@ -214,7 +214,7 @@ uploadLoop: // this fid whether we retry or give up here, and an unreferenced // needle is not garbage vacuum can find, so drop it either way. deleteChunkFromHolders(chunkHolders(assignResult), assignResult.Fid, jwt) - if attempt == chunkAssignAttempts || !shouldReassignUpload(uploadResultErr) || objectFailed() { + if attempt == chunkAssignAttempts || !ShouldReassignUpload(uploadResultErr) || objectFailed() { break } glog.V(2).Infof("re-assigning chunk at offset %d after attempt %d/%d: %v", offset, attempt, chunkAssignAttempts, uploadResultErr) @@ -400,7 +400,7 @@ func uploadChunkToHolders(ctx context.Context, hosts []string, fid string, data if firstErr == nil { firstErr = o.err cancel() - } else if !shouldReassignUpload(firstErr) && shouldReassignUpload(o.err) { + } else if !ShouldReassignUpload(firstErr) && ShouldReassignUpload(o.err) { firstErr = o.err } } else { diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index d65fbbb86..549c36aae 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -130,7 +130,7 @@ type uploadStatusError struct { func (e *uploadStatusError) Error() string { return e.err.Error() } func (e *uploadStatusError) Unwrap() error { return e.err } -// shouldReassignUpload reports whether an upload error means the client should +// ShouldReassignUpload reports whether an upload error means the client should // ask for a fresh volume assignment and retry on another volume. // // On the write path a volume server only 5xxs on a ReplicatedWrite failure @@ -139,7 +139,7 @@ func (e *uploadStatusError) Unwrap() error { return e.err } // assigned target never answered (down/unreachable), also reassignable. A 4xx // is a genuine client error and is surfaced. Errors without a status come from // the AssignVolume RPC or request setup; retry only transient transport ones. -func shouldReassignUpload(err error) bool { +func ShouldReassignUpload(err error) bool { if err == nil { return false } @@ -220,7 +220,7 @@ func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, ho return true }) } else { - err = util.RetryOnError("uploadWithRetry", shouldReassignUpload, doUploadFunc) + err = util.RetryOnError("uploadWithRetry", ShouldReassignUpload, doUploadFunc) } return diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index 6a65668f7..e1f3558ba 100644 --- a/weed/operation/upload_content_test.go +++ b/weed/operation/upload_content_test.go @@ -69,8 +69,8 @@ func TestShouldReassignUpload(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - if got := shouldReassignUpload(tc.err); got != tc.want { - t.Fatalf("shouldReassignUpload(%v) = %v, want %v", tc.err, got, tc.want) + if got := ShouldReassignUpload(tc.err); got != tc.want { + t.Fatalf("ShouldReassignUpload(%v) = %v, want %v", tc.err, got, tc.want) } }) } diff --git a/weed/server/filer_grpc_server_rename_test.go b/weed/server/filer_grpc_server_rename_test.go index be5a4476a..57757ca8d 100644 --- a/weed/server/filer_grpc_server_rename_test.go +++ b/weed/server/filer_grpc_server_rename_test.go @@ -250,12 +250,13 @@ func newRenameTestFiler(t *testing.T, store *renameTestStore) *filer.Filer { t.Cleanup(logBuffer.ShutdownLogBuffer) return &filer.Filer{ - Store: filer.NewFilerStoreWrapper(store), - MasterClient: masterClient, - FilerConf: filer.NewFilerConf(), - RemoteStorage: filer.NewFilerRemoteStorage(), - MaxFilenameLength: 255, - LocalMetaLogBuffer: logBuffer, + Store: filer.NewFilerStoreWrapper(store), + MasterClient: masterClient, + FilerConf: filer.NewFilerConf(), + RemoteStorage: filer.NewFilerRemoteStorage(), + MaxFilenameLength: 255, + LocalMetaLogBuffer: logBuffer, + FileIdDeletionQueue: util.NewUnboundedQueue(), } } diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 11b1c080b..44792feeb 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -127,6 +127,11 @@ type FilerServer struct { // chunk sharing (tier 1). Always populated. mountPeerRegistry *filer.MountPeerRegistry + // tusActiveUploads marks TUS sessions with a mutating request in flight, so + // a concurrent PATCH or DELETE is refused instead of recording duplicate + // chunks behind the first request's back. + tusActiveUploads sync.Map + // entryLockTable serializes mutations to the same entry path on this filer. // CreateEntry takes it today; UpdateEntry and DeleteEntry are intended to take // it too as their callers route a key's writes to this node, making it the diff --git a/weed/server/filer_server_handlers_write_upload.go b/weed/server/filer_server_handlers_write_upload.go index 256adf82c..f2c18043a 100644 --- a/weed/server/filer_server_handlers_write_upload.go +++ b/weed/server/filer_server_handlers_write_upload.go @@ -206,7 +206,12 @@ func (fs *FilerServer) dataToChunkWithSSE(ctx context.Context, r *http.Request, var uploadResult *operation.UploadResult var failedFileChunks []*filer_pb.FileChunk - err := util.Retry("filerDataToChunk", func() error { + // Each attempt assigns anew, so also retry the errors a fresh volume dodges: + // a target gone read-only or full mid-write 5xxs until the master notices. + shouldRetry := func(err error) bool { + return util.IsTransientError(err) || operation.ShouldReassignUpload(err) + } + err := util.RetryOnError("filerDataToChunk", shouldRetry, func() error { // assign one file id for one chunk fileId, urlLocation, auth, uploadErr = fs.assignNewFileInfo(ctx, so, uint64(len(data))) if uploadErr != nil { @@ -237,6 +242,13 @@ func (fs *FilerServer) dataToChunkWithSSE(ctx context.Context, r *http.Request, return failedFileChunks, err } + // A retry that lands elsewhere strands the earlier attempts: a volume server + // 5xxs after storing the needle locally when replication fails, and each + // attempt used its own file id, so nothing references them now. + if len(failedFileChunks) > 0 { + fs.filer.DeleteUncommittedChunks(ctx, failedFileChunks) + } + // if last chunk exhausted the reader exactly at the border if uploadResult.Size == 0 { return nil, nil diff --git a/weed/server/filer_server_tus_complete_test.go b/weed/server/filer_server_tus_complete_test.go new file mode 100644 index 000000000..4eaa5db8b --- /dev/null +++ b/weed/server/filer_server_tus_complete_test.go @@ -0,0 +1,122 @@ +package weed_server + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// TestFilerServer_completeTusUpload_OverlappingRecords verifies a session whose +// chunk records overlap - a PATCH retried while its predecessor was still +// storing a sub-chunk - still completes: coverage is validated the way HEAD +// computes the offset, and only records extending coverage join the entry. +func TestFilerServer_completeTusUpload_OverlappingRecords(t *testing.T) { + fidA, fidB, fidDup := "3,01637037d6", "4,02637037d6", "5,03637037d6" + + tests := []struct { + name string + chunks [][3]int64 // offset, size, fid index into fids + wantChunks [][2]int64 // offset, size expected on the entry + }{ + { + name: "exact duplicate dropped", + chunks: [][3]int64{{0, 8, 0}, {0, 8, 2}, {8, 4, 1}}, + wantChunks: [][2]int64{{0, 8}, {8, 4}}, + }, + { + name: "partial overlap kept", + chunks: [][3]int64{{0, 8, 0}, {6, 6, 2}}, + wantChunks: [][2]int64{{0, 8}, {6, 6}}, + }, + { + // a covered record naming a file id the entry keeps must not free it + name: "duplicate sharing a kept file id", + chunks: [][3]int64{{0, 8, 0}, {0, 4, 0}, {8, 4, 1}}, + wantChunks: [][2]int64{{0, 8}, {8, 4}}, + }, + } + + fids := []string{fidA, fidB, fidDup} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fs, store := newTusTestServer(t, nil) + targetPath := "/buckets/data/raced.bin" + size := int64(12) + seedTusSession(t, fs, store, TusSession{ID: tusTestUploadID, TargetPath: targetPath, Size: size}) + for _, c := range tt.chunks { + seedTusChunk(t, fs, store, tusTestUploadID, c[0], c[1], fids[c[2]]) + } + + // a zero-length PATCH at the reported offset triggers completion + req := tusRequest(http.MethodPatch, "/.tus/.uploads/"+tusTestUploadID, map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, nil, nil), + "Content-Type": "application/offset+octet-stream", + "Upload-Offset": "12", + }, "") + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + + if rec.Code != http.StatusNoContent { + t.Fatalf("PATCH = %d, want %d; body=%q", rec.Code, http.StatusNoContent, rec.Body.String()) + } + entry, err := store.FindEntry(context.Background(), util.FullPath(targetPath)) + if err != nil { + t.Fatalf("final entry not created: %v", err) + } + chunks := entry.GetChunks() + if len(chunks) != len(tt.wantChunks) { + t.Fatalf("entry chunks = %d, want %d", len(chunks), len(tt.wantChunks)) + } + for i, want := range tt.wantChunks { + if chunks[i].Offset != want[0] || int64(chunks[i].Size) != want[1] { + t.Errorf("chunk[%d] = @%d+%d, want @%d+%d", i, chunks[i].Offset, chunks[i].Size, want[0], want[1]) + } + } + + var freed []string + fs.filer.FileIdDeletionQueue.Consume(func(fileIds []string) { + freed = append(freed, fileIds...) + }) + for _, chunk := range chunks { + for _, fileId := range freed { + if fileId == chunk.FileId { + t.Errorf("file id %s is referenced by the entry but was freed", fileId) + } + } + } + if _, err := store.FindEntry(context.Background(), util.FullPath(fs.tusSessionInfoPath(tusTestUploadID))); err == nil { + t.Errorf("session still present after completion") + } + }) + } +} + +// TestFilerServer_completeTusUpload_GapRejected verifies records that do not +// cover the full size still fail completion. +func TestFilerServer_completeTusUpload_GapRejected(t *testing.T) { + fs, store := newTusTestServer(t, nil) + targetPath := "/buckets/data/gap.bin" + seedTusSession(t, fs, store, TusSession{ID: tusTestUploadID, TargetPath: targetPath, Size: 12}) + + session := &TusSession{ + ID: tusTestUploadID, + TargetPath: targetPath, + Size: 12, + Offset: 12, + Chunks: []*TusChunkInfo{ + {Offset: 0, Size: 8, FileId: "3,01637037d6"}, + {Offset: 9, Size: 3, FileId: "4,02637037d6"}, + }, + } + err := fs.completeTusUpload(context.Background(), session) + if err == nil || !strings.Contains(err.Error(), "chunk gap") { + t.Fatalf("completeTusUpload err = %v, want a chunk gap error", err) + } + if _, findErr := store.FindEntry(context.Background(), util.FullPath(targetPath)); findErr == nil { + t.Fatalf("entry created despite gap") + } +} diff --git a/weed/server/filer_server_tus_handlers.go b/weed/server/filer_server_tus_handlers.go index 83f6e6d5c..e2d9aa218 100644 --- a/weed/server/filer_server_tus_handlers.go +++ b/weed/server/filer_server_tus_handlers.go @@ -12,11 +12,9 @@ import ( "path" "strconv" "strings" - "time" "github.com/google/uuid" "github.com/seaweedfs/seaweedfs/weed/glog" - "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/stats" @@ -87,6 +85,18 @@ func (fs *FilerServer) tusHandler(w http.ResponseWriter, r *http.Request) { writeJsonError(w, r, http.StatusUnauthorized, errors.New("wrong jwt")) return } + // One mutating request per session at a time, like tusd: a PATCH retried + // while its predecessor is still storing a sub-chunk would otherwise + // record the same range twice, and a DELETE would race the writer. The + // chunk state is loaded under this claim so the offset check sees every + // record the previous request left behind. 423 tells the client to retry. + if r.Method != http.MethodHead { + if !fs.lockTusUpload(uploadID) { + http.Error(w, "Upload is locked by another request", http.StatusLocked) + return + } + defer fs.unlockTusUpload(uploadID) + } if err := fs.loadTusSessionChunks(ctx, session); err != nil { glog.Errorf("Failed to load TUS session %s chunks: %v", uploadID, err) writeTusSessionNotFound(w, r.Method) @@ -119,6 +129,17 @@ func (fs *FilerServer) tusHandler(w http.ResponseWriter, r *http.Request) { } } +// lockTusUpload claims a session for one mutating request; it reports false +// while another request holds the claim. +func (fs *FilerServer) lockTusUpload(uploadID string) bool { + _, loaded := fs.tusActiveUploads.LoadOrStore(uploadID, struct{}{}) + return !loaded +} + +func (fs *FilerServer) unlockTusUpload(uploadID string) { + fs.tusActiveUploads.Delete(uploadID) +} + // writeTusSessionNotFound answers a request whose session cannot be resolved. // DELETE is idempotent and returns 204 for a missing session; other verbs 404. func writeTusSessionNotFound(w http.ResponseWriter, method string) { @@ -253,7 +274,7 @@ func (fs *FilerServer) tusCreateHandler(w http.ResponseWriter, r *http.Request, } if r.ContentLength > 0 { // Upload data in the creation request - bytesWritten, uploadErr := fs.tusWriteData(ctx, session, 0, r.Body, r.ContentLength) + bytesWritten, uploadErr := fs.tusWriteData(ctx, r, session, 0, r.Body, r.ContentLength) if uploadErr != nil { // Cleanup session on failure fs.deleteTusSession(ctx, uploadID) @@ -522,7 +543,7 @@ func (fs *FilerServer) tusPatchHandler(w http.ResponseWriter, r *http.Request, s } // Write data - bytesWritten, err := fs.tusWriteData(ctx, session, uploadOffset, r.Body, r.ContentLength) + bytesWritten, err := fs.tusWriteData(ctx, r, session, uploadOffset, r.Body, r.ContentLength) if err != nil { if errors.Is(err, ErrContentTooLarge) { http.Error(w, "Content-Length exceeds remaining upload size", http.StatusRequestEntityTooLarge) @@ -580,7 +601,7 @@ var ErrContentTooLarge = fmt.Errorf("content length exceeds remaining upload siz // tusWriteData uploads data to volume servers in streaming chunks and updates session // It reads data in fixed-size sub-chunks to avoid buffering large TUS chunks entirely in memory -func (fs *FilerServer) tusWriteData(ctx context.Context, session *TusSession, offset int64, reader io.Reader, contentLength int64) (int64, error) { +func (fs *FilerServer) tusWriteData(ctx context.Context, r *http.Request, session *TusSession, offset int64, reader io.Reader, contentLength int64) (int64, error) { if contentLength == 0 { return 0, nil } @@ -633,12 +654,6 @@ func (fs *FilerServer) tusWriteData(ctx context.Context, session *TusSession, of var totalWritten int64 var uploadErr error - // Create one uploader for all sub-chunks to reuse HTTP client connections - uploader, uploaderErr := operation.NewUploader() - if uploaderErr != nil { - return 0, fmt.Errorf("create uploader: %w", uploaderErr) - } - chunkBuf := make([]byte, tusChunkSize) currentOffset := offset @@ -658,36 +673,26 @@ func (fs *FilerServer) tusWriteData(ctx context.Context, session *TusSession, of break } - chunkData := chunkBuf[:n] - - // Assign file ID from master for this sub-chunk - fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(ctx, so, uint64(n)) - if assignErr != nil { - uploadErr = fmt.Errorf("assign volume: %w", assignErr) + // Store the sub-chunk through the regular write path's chunk writer, + // which assigns a fresh file id per attempt; a failed attempt's needle + // is returned so it can be freed instead of lingering unreferenced. + chunks, chunkErr := fs.dataToChunkWithSSE(ctx, r, "", mimeType, chunkBuf[:n], currentOffset, so) + if chunkErr != nil { + fs.filer.DeleteUncommittedChunks(ctx, chunks) + uploadErr = fmt.Errorf("upload data: %w", chunkErr) break } - - // Upload to volume server using BytesReader (avoids double buffering in uploader) - uploadResult, uploadResultErr, _ := uploader.Upload(ctx, util.NewBytesReader(chunkData), &operation.UploadOption{ - UploadUrl: urlLocation, - Filename: "", - Cipher: fs.option.Cipher, - IsInputCompressed: false, - MimeType: mimeType, - PairMap: nil, - Jwt: auth, - }) - if uploadResultErr != nil { - uploadErr = fmt.Errorf("upload data: %w", uploadResultErr) + if len(chunks) == 0 { + uploadErr = fmt.Errorf("no chunk stored at offset %d", currentOffset) break } + stored := chunks[0] - // Create chunk info and save it chunk := &TusChunkInfo{ - Offset: currentOffset, - Size: int64(uploadResult.Size), - FileId: fileId, - UploadAt: time.Now().UnixNano(), + Offset: stored.Offset, + Size: int64(stored.Size), + FileId: stored.FileId, + UploadAt: stored.ModifiedTsNs, } if saveErr := fs.saveTusChunk(ctx, session.ID, chunk); saveErr != nil { @@ -696,8 +701,8 @@ func (fs *FilerServer) tusWriteData(ctx context.Context, session *TusSession, of break } - totalWritten += int64(uploadResult.Size) - currentOffset += int64(uploadResult.Size) + totalWritten += chunk.Size + currentOffset += chunk.Size stats.FilerHandlerCounter.WithLabelValues("tusUploadChunk").Inc() } diff --git a/weed/server/filer_server_tus_lock_test.go b/weed/server/filer_server_tus_lock_test.go new file mode 100644 index 000000000..9ca53273a --- /dev/null +++ b/weed/server/filer_server_tus_lock_test.go @@ -0,0 +1,51 @@ +package weed_server + +import ( + "net/http" + "net/http/httptest" + "testing" +) + +// TestFilerServer_tusHandler_ConcurrentMutationLocked verifies a session with a +// mutating request in flight refuses a second PATCH or DELETE with 423 while +// HEAD still answers, and accepts the retry once the first request finishes. +func TestFilerServer_tusHandler_ConcurrentMutationLocked(t *testing.T) { + fs, _ := newTusTestServer(t, map[string]string{tusTestUploadID: "/buckets/data/file.bin"}) + + if !fs.lockTusUpload(tusTestUploadID) { + t.Fatal("lockTusUpload failed on an idle session") + } + + do := func(method string, headers map[string]string) int { + req := tusRequest(method, "/.tus/.uploads/"+tusTestUploadID, headers, "") + rec := httptest.NewRecorder() + fs.tusHandler(rec, req) + return rec.Code + } + patchHeaders := map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, nil, nil), + "Content-Type": "application/offset+octet-stream", + "Upload-Offset": "0", + } + deleteHeaders := map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestWriteKey, nil, nil), + } + headHeaders := map[string]string{ + "Authorization": "Bearer " + signFilerToken(t, tusTestReadKey, nil, nil), + } + + if code := do(http.MethodPatch, patchHeaders); code != http.StatusLocked { + t.Errorf("PATCH while locked = %d, want %d", code, http.StatusLocked) + } + if code := do(http.MethodDelete, deleteHeaders); code != http.StatusLocked { + t.Errorf("DELETE while locked = %d, want %d", code, http.StatusLocked) + } + if code := do(http.MethodHead, headHeaders); code != http.StatusOK { + t.Errorf("HEAD while locked = %d, want %d", code, http.StatusOK) + } + + fs.unlockTusUpload(tusTestUploadID) + if code := do(http.MethodPatch, patchHeaders); code != http.StatusNoContent { + t.Errorf("PATCH after unlock = %d, want %d", code, http.StatusNoContent) + } +} diff --git a/weed/server/filer_server_tus_session.go b/weed/server/filer_server_tus_session.go index 2e6a6d4d5..b5bc6ae1d 100644 --- a/weed/server/filer_server_tus_session.go +++ b/weed/server/filer_server_tus_session.go @@ -490,40 +490,49 @@ func (fs *FilerServer) completeTusUpload(ctx context.Context, session *TusSessio pathLock := fs.entryLockTable.AcquireLock("tusComplete", sessionPath, util.ExclusiveLock) defer fs.entryLockTable.ReleaseLock(sessionPath, pathLock) - // Sort chunks by offset to ensure correct order + // Sort by offset, widest record first among equals, so a duplicate sorts + // behind the record that covers it sort.Slice(session.Chunks, func(i, j int) bool { - return session.Chunks[i].Offset < session.Chunks[j].Offset + if session.Chunks[i].Offset != session.Chunks[j].Offset { + return session.Chunks[i].Offset < session.Chunks[j].Offset + } + return session.Chunks[i].Size > session.Chunks[j].Size }) - // Validate chunks are contiguous with no gaps or overlaps - expectedOffset := int64(0) - for _, chunk := range session.Chunks { - if chunk.Offset != expectedOffset { - return fmt.Errorf("chunk gap or overlap detected: expected offset %d, got %d", expectedOffset, chunk.Offset) - } - expectedOffset = chunk.Offset + chunk.Size - } - if expectedOffset != session.Size { - return fmt.Errorf("chunks do not cover full file: chunks end at %d, expected %d", expectedOffset, session.Size) - } - - // Assemble file chunks in order + // A PATCH retried while its predecessor was still storing a sub-chunk can + // record one range twice. The records must cover the full size with no gap, + // the same watermark HEAD reports the offset from; a record extending + // coverage joins the entry (the read path resolves a partial overlap by + // ModifiedTsNs, and the raced copies carry identical bytes), while a fully + // covered duplicate is freed once the entry lands. var fileChunks []*filer_pb.FileChunk - + var duplicateChunks []*filer_pb.FileChunk + covered := int64(0) for _, chunk := range session.Chunks { + if chunk.Offset > covered { + return fmt.Errorf("chunk gap detected: covered up to %d, next chunk at %d", covered, chunk.Offset) + } + if chunk.Offset+chunk.Size <= covered { + duplicateChunks = append(duplicateChunks, &filer_pb.FileChunk{FileId: chunk.FileId}) + continue + } + covered = chunk.Offset + chunk.Size + fid, fidErr := filer_pb.ToFileIdObject(chunk.FileId) if fidErr != nil { return fmt.Errorf("invalid file ID %s at offset %d: %w", chunk.FileId, chunk.Offset, fidErr) } - fileChunk := &filer_pb.FileChunk{ + fileChunks = append(fileChunks, &filer_pb.FileChunk{ FileId: chunk.FileId, Offset: chunk.Offset, Size: uint64(chunk.Size), ModifiedTsNs: chunk.UploadAt, Fid: fid, - } - fileChunks = append(fileChunks, fileChunk) + }) + } + if covered != session.Size { + return fmt.Errorf("chunks do not cover full file: chunks end at %d, expected %d", covered, session.Size) } // Determine content type from metadata @@ -583,6 +592,25 @@ func (fs *FilerServer) completeTusUpload(ctx context.Context, session *TusSessio return fmt.Errorf("create final file entry: %w", err) } + // Free the duplicates' data; their records go with the session directory + // below. A file id the entry still references is never freed: coverage is + // computed from ranges, so a malformed record could name one. + if len(duplicateChunks) > 0 { + referenced := make(map[string]bool, len(fileChunks)) + for _, chunk := range fileChunks { + referenced[chunk.FileId] = true + } + var chunksToDelete []*filer_pb.FileChunk + for _, chunk := range duplicateChunks { + if !referenced[chunk.FileId] { + chunksToDelete = append(chunksToDelete, chunk) + } + } + if len(chunksToDelete) > 0 { + fs.filer.DeleteChunks(ctx, targetPath, chunksToDelete) + } + } + // Delete the session (but keep the chunks since they're now part of the final file) sessionDirPath := util.FullPath(fs.tusSessionPath(session.ID)) if err := fs.filer.DeleteEntryMetaAndData(ctx, sessionDirPath, true, false, false, false, nil, 0); err != nil {