filer: stop TUS uploads from turning into garbage (#10945)

* filer: store TUS sub-chunks through the regular chunk writer

A TUS sub-chunk was written with one assigned file id, retried up to
three times against that same id, and abandoned on failure: an attempt
that had landed on some replicas left a needle no session record and no
entry ever references, unreclaimable by vacuum.

dataToChunkWithSSE, which the regular write path uses per chunk, assigns
a fresh file id per attempt and hands back the file ids of failed
attempts, which are now freed the way the regular write path frees them.

* filer: retry a chunk write on a fresh volume when the server 5xxs

The filer's chunk writer assigns a fresh file id per attempt but only
retried transient network errors, so a volume filling up and turning
read-only mid-write failed the whole request even though the very next
assignment would have landed elsewhere. Every other write client already
routes this through ShouldReassignUpload; the filer's own write path now
does the same, for regular uploads and TUS sub-chunks alike.

* filer: export the chunk deletion queue

The filer test harness in weed/server builds filer.Filer as a struct
literal, so any code path reaching DeleteChunks dereferenced a nil
queue. Exported like the neighboring DeletionRetryQueue so the harness
can arm it.

* filer: complete a TUS upload whose chunk records overlap

A PATCH retried while its predecessor was still storing a sub-chunk -
a proxy timeout with an immediate retry is enough - records the same
range twice. HEAD computes Upload-Offset as the covered watermark and
reported the upload fully received, but completion demanded exactly
adjacent records and failed every attempt: the client concluded success
from offset == length, no entry was created, and the session eventually
expired, turning the entire upload into deleted needles for the vacuum
to chew through.

Completion now validates gapless coverage with the same watermark HEAD
uses. A record extending coverage joins the entry - the read path
resolves partial overlaps by ModifiedTsNs, and the raced copies carry
identical bytes - while a fully covered duplicate is freed once the
entry lands.

* filer: allow one mutating TUS request per session at a time

Nothing stopped two PATCHes from writing the same range concurrently:
both loaded the same offset, both passed the conflict check, and both
recorded their sub-chunks. A client whose request timed out in a proxy
retries immediately while the server side is still storing the buffered
sub-chunk, which is exactly that race.

A session now accepts one PATCH or DELETE at a time, the way tusd locks
uploads; a concurrent one is refused with 423 Locked, which TUS clients
retry, and HEAD keeps answering so progress polling is unaffected. The
chunk state is loaded under the claim, so a retried PATCH sees every
record its predecessor left and conflicts cleanly instead of duplicating
data.

* test: cover a TUS PATCH raced by its own retry

Stalls a PATCH mid-body over a raw connection, retries the same range
while it is in flight, and expects the retry refused with 423 Locked;
the upload then resumes from the reported offset and the final content
must be intact.

* filer: never free a TUS duplicate the entry still references

Coverage is computed from ranges, so a record fully covered by another
is treated as a duplicate no matter which needle it names. A malformed
record naming a file id the entry keeps would have had that needle freed
right after the entry landed - the corruption this change set exists to
stop. The duplicates are now freed in one batch, skipping any file id
the entry references; their records go with the session directory.

* test: bound the raw TUS connection reads

http.ReadResponse on the stalled PATCH's connection blocked until the
whole go test timeout if the filer never answered.

* filer: free the needles of chunk write attempts a retry replaced

A volume server stores the needle locally and only then fans out to the
replicas, so a replication failure 5xxs with the data already written.
Each attempt assigns its own file id, so once a later attempt lands
elsewhere nothing references the earlier ones: the caller only sees the
chunk that succeeded, and the failed ids were dropped.

They are now freed the way the caller frees them when the whole write
fails. Retrying on a 5xx makes this reachable on every read-only or full
volume, which is exactly the condition that filled the reporter's
volumes.

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")
+}

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(),

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())
 	}
 }
 

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() {}),

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 {

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

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)
 			}
 		})
 	}

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(),
 	}
 }
 

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

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

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")
+	}
+}

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()
 	}
 

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)
+	}
+}

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 {