diff --git a/weed/filer/filer_notify_append.go b/weed/filer/filer_notify_append.go index 27bda98fdc..2b69b755ac 100644 --- a/weed/filer/filer_notify_append.go +++ b/weed/filer/filer_notify_append.go @@ -2,10 +2,13 @@ package filer import ( "context" + "errors" "fmt" "os" + "slices" "time" + "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/util" @@ -22,10 +25,14 @@ func (f *Filer) appendToFile(ctx context.Context, targetFile string, data []byte // expired shutdown deadline does not strand the chunk. ctx = context.WithoutCancel(ctx) + return f.commitAppendedChunk(ctx, util.FullPath(targetFile), uploadResult.ToPbFileChunk(assignResult.Fid, 0, time.Now().UnixNano())) +} + +// commitAppendedChunk appends an uploaded chunk to the file, releasing the +// chunk when it cannot be committed: the flush retries with a new upload. +func (f *Filer) commitAppendedChunk(ctx context.Context, fullpath util.FullPath, chunk *filer_pb.FileChunk) error { // find out existing entry - fullpath := util.FullPath(targetFile) entry, err := f.FindEntry(ctx, fullpath) - var offset int64 = 0 if err == filer_pb.ErrNotFound { entry = &Entry{ FullPath: fullpath, @@ -38,18 +45,32 @@ func (f *Filer) appendToFile(ctx context.Context, targetFile string, data []byte }, } } else if err != nil { + f.DeleteUncommittedChunks(ctx, []*filer_pb.FileChunk{chunk}) return fmt.Errorf("find %s: %v", fullpath, err) } else { - offset = int64(TotalSize(entry.GetChunks())) + chunk.Offset = int64(TotalSize(entry.GetChunks())) } // append to existing chunks - entry.Chunks = append(entry.GetChunks(), uploadResult.ToPbFileChunk(assignResult.Fid, offset, time.Now().UnixNano())) + entry.Chunks = append(entry.GetChunks(), chunk) // update the entry - err = f.CreateEntry(ctx, entry, nil, false, false, nil, false, f.MaxFilenameLength) - - return err + if err := f.CreateEntry(ctx, entry, nil, false, false, nil, false, f.MaxFilenameLength); err != nil { + // A create can store the entry and then fail on its parent folders; + // a retry would only append the piece again. + stored, findErr := f.FindEntry(ctx, fullpath) + if findErr == nil && slices.ContainsFunc(stored.GetChunks(), func(c *filer_pb.FileChunk) bool { + return c.GetFileIdString() == chunk.GetFileIdString() + }) { + glog.V(0).Infof("append to %s landed despite: %v", fullpath, err) + return nil + } + if findErr == nil || errors.Is(findErr, filer_pb.ErrNotFound) { + f.DeleteUncommittedChunks(ctx, []*filer_pb.FileChunk{chunk}) + } + return err + } + return nil } // resolveMetadataLogAssignDiskType returns the disk type and matched path rule for diff --git a/weed/filer/filer_notify_append_orphan_test.go b/weed/filer/filer_notify_append_orphan_test.go new file mode 100644 index 0000000000..e503497905 --- /dev/null +++ b/weed/filer/filer_notify_append_orphan_test.go @@ -0,0 +1,81 @@ +package filer + +import ( + "context" + "errors" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const appendTestLog = util.FullPath("/topics/.system/log/2026-11-05/10-00.00000001") + +func newAppendTestFiler(store *stubFilerStore) *Filer { + f := NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil) + f.Store = NewFilerStoreWrapper(store) + return f +} + +func queuedForDeletion(f *Filer, fid string) bool { + f.deletionLedgerLock.Lock() + defer f.deletionLedgerLock.Unlock() + _, queued := f.pendingDeletions[fid] + return queued +} + +func TestCommitAppendedChunkReleasesChunkWhenCommitFails(t *testing.T) { + store := newStubFilerStore() + store.insertErr = errors.New("store unavailable") + f := newAppendTestFiler(store) + + err := f.commitAppendedChunk(context.Background(), appendTestLog, &filer_pb.FileChunk{FileId: "7,0abc", Size: 5}) + + require.Error(t, err) + assert.True(t, queuedForDeletion(f, "7,0abc"), "a chunk no entry references must be released") +} + +func TestCommitAppendedChunkReleasesChunkWhenLookupFails(t *testing.T) { + store := newStubFilerStore() + store.findErrByPath = map[string]error{string(appendTestLog): errors.New("store unavailable")} + f := newAppendTestFiler(store) + + err := f.commitAppendedChunk(context.Background(), appendTestLog, &filer_pb.FileChunk{FileId: "7,0abc", Size: 5}) + + require.Error(t, err) + assert.True(t, queuedForDeletion(f, "7,0abc"), "a chunk no entry references must be released") +} + +func TestCommitAppendedChunkKeepsChunkTheFailedCreateStored(t *testing.T) { + store := newStubFilerStore() + parent, _ := appendTestLog.DirAndName() + store.findErrByPath = map[string]error{parent: errors.New("store unavailable")} + f := newAppendTestFiler(store) + + err := f.commitAppendedChunk(context.Background(), appendTestLog, &filer_pb.FileChunk{FileId: "7,0abc", Size: 5}) + + require.NoError(t, err, "a stored piece must not be appended again by the flush retry") + stored, findErr := store.FindEntry(context.Background(), appendTestLog) + require.NoError(t, findErr, "the entry is stored before its parent folders are created") + require.Len(t, stored.Chunks, 1) + assert.False(t, queuedForDeletion(f, "7,0abc"), "a chunk the stored entry references must stay") +} + +func TestCommitAppendedChunkAppendsAfterExistingChunks(t *testing.T) { + store := newStubFilerStore() + f := newAppendTestFiler(store) + ctx := context.Background() + + require.NoError(t, f.commitAppendedChunk(ctx, appendTestLog, &filer_pb.FileChunk{FileId: "7,0abc", Size: 5})) + require.NoError(t, f.commitAppendedChunk(ctx, appendTestLog, &filer_pb.FileChunk{FileId: "7,0abd", Size: 3})) + + stored, err := store.FindEntry(ctx, appendTestLog) + require.NoError(t, err) + require.Len(t, stored.Chunks, 2) + assert.Equal(t, int64(5), stored.Chunks[1].Offset) + assert.False(t, queuedForDeletion(f, "7,0abc")) + assert.False(t, queuedForDeletion(f, "7,0abd")) +}