diff --git a/weed/mount/README.md b/weed/mount/README.md new file mode 100644 index 0000000000..1daf2b02e2 --- /dev/null +++ b/weed/mount/README.md @@ -0,0 +1 @@ +Concurrent writers to one file from different mounts require `-dlm`; without it, a conflicting write fails with EIO and the file may refuse further appends until reopened after the cache drops. diff --git a/weed/mount/content_gap.go b/weed/mount/content_gap.go new file mode 100644 index 0000000000..b9b5ba8c77 --- /dev/null +++ b/weed/mount/content_gap.go @@ -0,0 +1,81 @@ +package mount + +import ( + "fmt" + "sort" + "sync/atomic" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// A publication owner carries provider refusal evidence through every retry. +// Helpers and the periodic flusher never retire content that is still owned. +type contentPublication struct { + refused bool + failure *atomic.Bool +} + +func (p *contentPublication) retire(err error) { + if err != nil && p.refused { + p.abandon() + } +} + +// An abandoned writeback drain leaves the kernel EOF past the stored one, +// whatever the filer answered. +func (p *contentPublication) abandon() { + if p.failure != nil { + p.failure.Store(true) + } +} + +func (wfs *WFS) checkContentPublication(fh *FileHandle, base *filer_pb.Entry) error { + if wfs.option.WritebackCache && (fh.writebackContentError == nil || fh.writebackContentError.Load()) { + return fmt.Errorf("writeback inode awaits Forget after content refusal: %w", filer_pb.ErrPreconditionFailed) + } + // This guard shares the CAS version with CreateEntry. A rewritten or + // manifest bounding range cannot conceal the original kernel WRITE frontier. + var eof uint64 + if base != nil { + eof = filer.FileSize(base) + } + if !wfs.option.WritebackCache && fh.pendingAppend && fh.pendingAppendOffset > eof { + return fmt.Errorf("append offset %d exceeds committed EOF %d: %w", fh.pendingAppendOffset, eof, filer_pb.ErrPreconditionFailed) + } + return nil +} + +// Only used when a peer shrank the file: a merge must not acknowledge a gap +// between that stored EOF and local pending content. Ordinary sparse pwrite +// without a concurrent truncate is unaffected. Compare copies, never reorder +// a handle's chunks. A manifest's bounding range is not proof of coverage. +func contentGapAfter(entry *filer_pb.Entry, start uint64) bool { + end := filer.FileSize(entry) + if start >= end { + return false + } + chunks := append([]*filer_pb.FileChunk(nil), entry.Chunks...) + sort.SliceStable(chunks, func(i, j int) bool { return chunks[i].Offset < chunks[j].Offset }) + covered := start + if uint64(len(entry.Content)) > covered { + covered = uint64(len(entry.Content)) + } + for _, c := range chunks { + if covered >= end { + return false + } + if c.Offset < 0 || c.Size > ^uint64(0)-uint64(c.Offset) { + return true + } + left, right := uint64(c.Offset), uint64(c.Offset)+c.Size + if right <= covered { + continue + } + if left > covered || c.IsChunkManifest { + return true + } + covered = right + } + return covered < end +} diff --git a/weed/mount/content_gap_merge_test.go b/weed/mount/content_gap_merge_test.go new file mode 100644 index 0000000000..a00b252ebb --- /dev/null +++ b/weed/mount/content_gap_merge_test.go @@ -0,0 +1,212 @@ +//go:build !freebsd + +package mount + +import ( + "bytes" + "context" + "syscall" + "testing" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "google.golang.org/protobuf/proto" +) + +// These caller controls measure published bytes; none imports #12608 internals. +func masterAssertPeerMergeOutcome(t *testing.T, periodic bool) { + for _, scenario := range []string{"peer-append", "peer-replaced-file", "peer-truncate"} { + t.Run(scenario, func(t *testing.T) { + base := attributeAppendEntry(1) + stored := attributeAppendEntry(2) + pending := attributeAppendEntry(2) + pending.Chunks[1].Fid.FileKey = 3 + pending.Chunks[1].Offset = 512 + pending.Attributes.FileSize = 768 + blobs := map[string][]byte{} + for i, c := range stored.Chunks { + blobs[c.GetFileIdString()] = bytes.Repeat([]byte{[]byte{'O', 'P'}[i]}, 256) + } + blobs[pending.Chunks[1].GetFileIdString()] = bytes.Repeat([]byte{'C'}, 256) + switch scenario { + case "peer-replaced-file": + stored = attributeAppendEntry(1) + stored.Attributes.Inode = 99 + stored.Chunks[0].Fid.FileKey = 900 + blobs[stored.Chunks[0].GetFileIdString()] = bytes.Repeat([]byte{'R'}, 256) + case "peer-truncate": + base = attributeAppendEntry(2) + stored = attributeAppendEntry(1) + pending = attributeAppendEntry(3) + blobs[pending.Chunks[2].GetFileIdString()] = bytes.Repeat([]byte{'C'}, 256) + blobs[base.Chunks[1].GetFileIdString()] = bytes.Repeat([]byte{'A'}, 256) + } + stored.Attributes.FileMode, stored.CommitTsNs = 0600, 20000 + original := proto.Clone(stored).(*filer_pb.Entry) + peer := &g6ConditionalPeer{entry: proto.Clone(stored).(*filer_pb.Entry), blobs: blobs} + w := newInvalidateTestWFS(t) + startFakeFiler(t, w, peer) + w.mapPbIdFromFilerToLocal(base) + w.mapPbIdFromFilerToLocal(pending) + fh := newFileHandle(w, 1, 42, pending) + t.Cleanup(fh.dirtyPages.Destroy) + fh.RememberPath("/dir/file") + fh.baseEntry.Store(base) + fh.dirtyMetadata = true + code := fuse.OK + if periodic { + if err := w.flushFileMetadata(fh); err != nil { + code = fuse.EIO + t.Logf("periodic error: %v", err) + } + } else { + code = w.doFlush(context.Background(), fh, 1000, 20, false, 0) + } + peer.mu.Lock() + defer peer.mu.Unlock() + actual := g6StoredBytes(t, peer) + t.Logf("scenario=%s status=%v size=%d rejected=%d gap[256,512)-zero=%v", scenario, code, len(actual), peer.rejected, len(actual) >= 512 && bytes.Equal(actual[256:512], make([]byte, 256))) + if scenario == "peer-replaced-file" { + if code != fuse.EIO || !g6EntryEqual(peer.entry, original) || !bytes.Equal(actual, bytes.Repeat([]byte{'R'}, 256)) { + t.Fatal("replacement must be refused without publishing stale content") + } + return + } + if scenario == "peer-truncate" && code == fuse.EIO { + if !g6EntryEqual(peer.entry, original) || !bytes.Equal(actual, bytes.Repeat([]byte{'O'}, 256)) { + t.Fatal("refusal changed peer truncate") + } + return + } + want := bytes.Repeat([]byte{'O'}, 256) + if scenario == "peer-append" { + want = append(want, bytes.Repeat([]byte{'P'}, 256)...) + } + want = append(want, bytes.Repeat([]byte{'C'}, 256)...) + if code != fuse.OK || peer.entry.Attributes.FileMode != 0600 || peer.entry.Attributes.FileSize != uint64(len(want)) || !bytes.Equal(actual, want) { + t.Fatalf("unsafe merge: status=%v size=%d want-size=%d exact-bytes=%v; peer/local bytes must survive without a gap", code, len(actual), len(want), bytes.Equal(actual, want)) + } + }) + } +} + +func TestMasterCommittedEOFAppendHasNoGap(t *testing.T) { + for _, appendOpen := range []bool{true, false} { + name := "append" + if !appendOpen { + name = "intentional-sparse-pwrite" + } + t.Run(name, func(t *testing.T) { + base := attributeAppendEntry(1) + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + w, h := g6AppendHandle(t, peer, base, 2100) + if !appendOpen { + out := &fuse.OpenOut{} + if code := w.Open(nil, &fuse.OpenIn{InHeader: fuse.InHeader{NodeId: 42}, Flags: syscall.O_WRONLY}, out); code != fuse.OK { + t.Fatal(code) + } + h = out.Fh + } + if n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: h, Offset: 512, Size: 256}, bytes.Repeat([]byte{'B'}, 256)); n != 256 || code != fuse.OK { + t.Fatal(n, code) + } + code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: h}) + peer.mu.Lock() + defer peer.mu.Unlock() + actual := g6StoredBytes(t, peer) + if appendOpen { + if code != fuse.EIO || peer.entry.Attributes.FileSize != 256 || !bytes.Equal(actual, bytes.Repeat([]byte{'O'}, 256)) { + t.Fatalf("stale append frontier published a gap: status=%v size=%d interval[256,512)-zero=%v", code, len(actual), len(actual) >= 512 && bytes.Equal(actual[256:512], make([]byte, 256))) + } + } else { + want := append(bytes.Repeat([]byte{'O'}, 256), make([]byte, 256)...) + want = append(want, bytes.Repeat([]byte{'B'}, 256)...) + if code != fuse.OK || !bytes.Equal(actual, want) || peer.entry.Attributes.FileSize != 768 { + t.Fatal("intentional sparse pwrite must remain allowed", code, len(actual)) + } + } + }) + } +} + +func TestMasterConflictThenAppendCannotPublishGap(t *testing.T) { + for _, mode := range []struct { + name string + writeback, coalesced bool + }{{"normal", false, false}, {"writeback", true, false}, {"writeback-coalesced", true, true}} { + t.Run(mode.name, func(t *testing.T) { + base := attributeAppendEntry(1) + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + w, h := g6AppendHandle(t, peer, base, 2200, mode.writeback) + if n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: h, Offset: 256, Size: 256}, bytes.Repeat([]byte{'A'}, 256)); n != 256 || code != fuse.OK { + t.Fatal(n, code) + } + // Model the native compactor's output, including independently uploaded + // bytes and the rewrite marker: a single chunk containing original O+A. + // No product implementation is supplied or changed by this fixture. + fh := w.GetHandle(FileHandleId(h)) + if err := fh.dirtyPages.FlushData(); err != nil { + t.Fatal(err) + } + compacted := &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 8000, Cookie: 1}, Size: 512, ModifiedTsNs: 100} + peer.mu.Lock() + peer.blobs[compacted.GetFileIdString()] = append(bytes.Repeat([]byte{'O'}, 256), bytes.Repeat([]byte{'A'}, 256)...) + peer.mu.Unlock() + fh.UpdateEntry(func(e *filer_pb.Entry) { + e.Chunks = []*filer_pb.FileChunk{proto.Clone(compacted).(*filer_pb.FileChunk)} + e.Attributes.FileSize = 512 + }) + fh.noteRewrite() + replacement := attributeAppendEntry(1) + replacement.Chunks[0].Fid.FileKey = 900 + replacement.Attributes.FileMode, replacement.CommitTsNs = 0600, 20000 + peer.mu.Lock() + peer.entry = proto.Clone(replacement).(*filer_pb.Entry) + peer.clock = 20000 + peer.blobs[replacement.Chunks[0].GetFileIdString()] = bytes.Repeat([]byte{'P'}, 256) + peer.mu.Unlock() + if code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: h}); code != fuse.EIO { + t.Fatal("setup: native real rewrite conflict must be refused", code) + } + w.Release(nil, &fuse.ReleaseIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: h, ReleaseFlags: fuse.FUSE_RELEASE_FLOCK_UNLOCK}) + if _, kept := w.fhMap.FindFileHandle(42); kept { + t.Fatal("failed synchronous close retained handle") + } + peer.mu.Lock() + unchanged := g6EntryEqual(peer.entry, replacement) + peer.mu.Unlock() + if !unchanged { + t.Fatal("refused close published over peer") + } + if err := w.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", proto.Clone(replacement).(*filer_pb.Entry)), 20000); err != nil { + t.Fatal(err) + } + out := &fuse.OpenOut{} + if code := w.Open(nil, &fuse.OpenIn{InHeader: fuse.InHeader{NodeId: 42}, Flags: syscall.O_WRONLY | syscall.O_APPEND}, out); code != fuse.OK { + t.Fatal(code) + } + g6AppendUploader(t, w.GetHandle(FileHandleId(out.Fh)), 2300, peer) + offset := uint64(512) + payload := bytes.Repeat([]byte{'B'}, 256) + if mode.coalesced { + offset = 0 + payload = append(bytes.Repeat([]byte{'P'}, 256), make([]byte, 256)...) + payload = append(payload, bytes.Repeat([]byte{'B'}, 256)...) + } + if n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh, Offset: offset, Size: uint32(len(payload))}, payload); n != uint32(len(payload)) || code != fuse.OK { + t.Fatal(n, code) + } + code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh}) + peer.mu.Lock() + defer peer.mu.Unlock() + actual := g6StoredBytes(t, peer) + if code != fuse.EIO || peer.entry.Attributes.FileSize != 256 || !bytes.Equal(actual, bytes.Repeat([]byte{'P'}, 256)) { + t.Fatalf("terminal conflict followed by false success/gap: status=%v size=%d interval[256,512)-zero=%v", code, len(actual), len(actual) >= 512 && bytes.Equal(actual[256:512], make([]byte, 256))) + } + }) + } +} + +func TestContentGapPeerMerge(t *testing.T) { masterAssertPeerMergeOutcome(t, false) } +func TestContentGapPeriodicPeerMerge(t *testing.T) { masterAssertPeerMergeOutcome(t, true) } diff --git a/weed/mount/content_gap_test.go b/weed/mount/content_gap_test.go new file mode 100644 index 0000000000..e26f72133d --- /dev/null +++ b/weed/mount/content_gap_test.go @@ -0,0 +1,600 @@ +//go:build !freebsd + +package mount + +import ( + "bytes" + "context" + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" + "io" + "os" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" +) + +func g6AppendHandle(t *testing.T, peer *g6ConditionalPeer, base *filer_pb.Entry, key uint64, writeback ...bool) (*WFS, uint64) { + t.Helper() + w := newInvalidateTestWFS(t) + w.option.CacheDirForWrite = t.TempDir() + w.option.DefaultPermissions = true + w.option.WritebackCache = len(writeback) > 0 && writeback[0] + w.posixLocks = NewPosixLockTable() + w.option.ConcurrentWriters = 1 + w.concurrentWriters = util.NewLimitedConcurrentExecutor(1) + startFakeFiler(t, w, peer) + w.inodeToPath.Lookup("/dir", 1, true, false, 0, false) + w.inodeToPath.Lookup("/dir/file", 1, false, false, 42, true) + w.inodeToPath.MarkChildrenCached("/dir") + if err := w.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", proto.Clone(base).(*filer_pb.Entry)), 1); err != nil { + t.Fatal(err) + } + out := &fuse.OpenOut{} + if code := w.Open(nil, &fuse.OpenIn{InHeader: fuse.InHeader{NodeId: 42}, Flags: syscall.O_WRONLY | syscall.O_APPEND}, out); code != fuse.OK { + t.Fatal(code) + } + fh := w.GetHandle(FileHandleId(out.Fh)) + peer.mu.Lock() + if peer.blobs == nil { + peer.blobs = make(map[string][]byte) + } + peer.blobs[base.Chunks[0].GetFileIdString()] = bytes.Repeat([]byte{'O'}, 256) + peer.mu.Unlock() + g6AppendUploader(t, fh, key, peer) + return w, out.Fh +} + +func g6AppendUploader(t *testing.T, fh *FileHandle, key uint64, peer *g6ConditionalPeer) { + t.Helper() + t.Cleanup(fh.dirtyPages.Destroy) + fh.dirtyPages.randomWriter.(*ChunkedDirtyPages).saveDataAsChunk = func(util.FullPath) filer.SaveDataAsChunkFunctionType { + return func(r io.Reader, _ string, offset, ts int64, _ uint64) (*filer_pb.FileChunk, error) { + body, err := io.ReadAll(r) + if err != nil { + return nil, err + } + key++ + chunk := &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: key, Cookie: 1}, Offset: offset, Size: uint64(len(body)), ModifiedTsNs: ts} + peer.mu.Lock() + peer.blobs[chunk.GetFileIdString()] = append([]byte(nil), body...) + peer.mu.Unlock() + return chunk, nil + } + } +} + +func TestAppendAfterRefusedFlushHasNoHole(t *testing.T) { + base := attributeAppendEntry(1) + base.CommitTsNs = 100 + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + w, handle := g6AppendHandle(t, peer, base, 100) + write := func(w *WFS, handle, offset uint64, value byte) { + t.Helper() + data := bytes.Repeat([]byte{value}, 256) + n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle, Offset: offset, Size: 256}, data) + if n != 256 || code != fuse.OK { + t.Fatalf("write: %d %v", n, code) + } + } + write(w, handle, 256, 'A') + // A peer chmod changes only attributes after this handle observed its base. + peer.mu.Lock() + peer.entry.Attributes.FileMode = 0600 + peer.entry.CommitTsNs++ + peer.mu.Unlock() + first := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle}) + t.Logf("first fsync=%v", first) + w.Release(nil, &fuse.ReleaseIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle, ReleaseFlags: fuse.FUSE_RELEASE_FLOCK_UNLOCK}) + peer.mu.Lock() + latest := proto.Clone(peer.entry).(*filer_pb.Entry) + peer.mu.Unlock() + if err := w.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", latest), 3000); err != nil { + t.Fatal(err) + } + out := &fuse.OpenOut{} + if code := w.Open(nil, &fuse.OpenIn{InHeader: fuse.InHeader{NodeId: 42}, Flags: syscall.O_WRONLY | syscall.O_APPEND}, out); code != fuse.OK { + t.Fatal(code) + } + next, nextHandle := w, out.Fh + g6AppendUploader(t, w.GetHandle(FileHandleId(nextHandle)), 200, peer) + // FUSE WRITE offsets are supplied by the kernel; its cached EOF advanced + // on the first successful WRITE even when that handle's fsync failed. + write(next, nextHandle, 512, 'B') + if code := next.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: nextHandle}); code != fuse.OK { + t.Fatal(code) + } + peer.mu.Lock() + defer peer.mu.Unlock() + want := append(bytes.Repeat([]byte{'O'}, 256), bytes.Repeat([]byte{'A'}, 256)...) + want = append(want, bytes.Repeat([]byte{'B'}, 256)...) + if peer.entry.Attributes.FileSize != 768 || peer.entry.Attributes.FileMode != 0600 || !bytes.Equal(g6StoredBytes(t, peer), want) { + t.Fatal("original/A/B bytes, size or peer mode changed") + } + for _, offset := range []int64{0, 256, 512} { + covered := false + for _, c := range peer.entry.Chunks { + if c.Offset <= offset && c.Offset+int64(c.Size) >= offset+256 { + covered = true + } + } + if !covered { + t.Fatalf("fsync acknowledged a hole [%d,%d): first=%v chunks=%v", offset, offset+256, first, peer.entry.Chunks) + } + } +} + +type g6ConditionalPeer struct { + filer_pb.UnimplementedSeaweedFilerServer + mu sync.Mutex + entry *filer_pb.Entry + rejected int + refuse bool + blobs map[string][]byte + clock int64 + beforeCreate func(context.Context) +} + +func g6EntryEqual(current, expected *filer_pb.Entry) bool { + actual := filer.FromPbEntry("", proto.Clone(current).(*filer_pb.Entry)).ToProtoEntry() + want := proto.Clone(expected).(*filer_pb.Entry) + filer_pb.BeforeEntrySerialization(actual.Chunks) + filer_pb.BeforeEntrySerialization(want.Chunks) + return proto.Equal(actual, filer.FromPbEntry("", want).ToProtoEntry()) +} + +func g6StoredBytes(t *testing.T, s *g6ConditionalPeer) []byte { + t.Helper() + result := make([]byte, s.entry.Attributes.FileSize) + for _, c := range s.entry.Chunks { + body, ok := s.blobs[c.GetFileIdString()] + if !ok || uint64(len(body)) != c.Size || c.Offset < 0 || uint64(c.Offset)+c.Size > uint64(len(result)) { + t.Fatalf("invalid stored reference %v", c) + } + copy(result[c.Offset:], body) + } + return result +} + +func (s *g6ConditionalPeer) LookupDirectoryEntry(context.Context, *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(s.entry).(*filer_pb.Entry)}, nil +} + +func (s *g6ConditionalPeer) UpdateEntry(_ context.Context, r *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + for _, c := range r.GetCondition().GetClauses() { + if c.Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL || !g6EntryEqual(s.entry, c.ExpectedEntry) { + return nil, status.Error(codes.FailedPrecondition, "entry changed") + } + } + s.entry = proto.Clone(r.Entry).(*filer_pb.Entry) + s.clock = max(max(s.clock, s.entry.CommitTsNs), 10000) + 1 + s.entry.CommitTsNs = s.clock + return &filer_pb.UpdateEntryResponse{LogTsNs: s.clock, MetadataEvent: &filer_pb.SubscribeMetadataResponse{Directory: r.Directory, TsNs: s.clock, EventNotification: &filer_pb.EventNotification{NewEntry: proto.Clone(s.entry).(*filer_pb.Entry)}}}, nil +} + +func (s *g6ConditionalPeer) CreateEntry(ctx context.Context, r *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + if s.beforeCreate != nil { + s.beforeCreate(ctx) + } + s.mu.Lock() + defer s.mu.Unlock() + if s.refuse { + s.rejected++ + return &filer_pb.CreateEntryResponse{Error: "precondition failed", ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil + } + for _, c := range r.GetCondition().GetClauses() { + match := c.Kind == filer_pb.WriteCondition_IF_NOT_EXISTS && s.entry == nil || c.Kind == filer_pb.WriteCondition_IF_ENTRY_EQUAL && s.entry != nil && c.ExpectedEntry != nil && g6EntryEqual(s.entry, c.ExpectedEntry) + if !match { + s.rejected++ + return &filer_pb.CreateEntryResponse{Error: "precondition failed", ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil + } + } + s.entry = proto.Clone(r.Entry).(*filer_pb.Entry) + s.clock = max(max(s.clock, s.entry.CommitTsNs), 10000) + 1 + s.entry.CommitTsNs = s.clock + return &filer_pb.CreateEntryResponse{LogTsNs: s.entry.CommitTsNs, MetadataEvent: &filer_pb.SubscribeMetadataResponse{Directory: r.Directory, TsNs: s.entry.CommitTsNs, EventNotification: &filer_pb.EventNotification{NewEntry: proto.Clone(s.entry).(*filer_pb.Entry)}}}, nil +} + +func TestG6AppendFsyncAfterPeerAttribute(t *testing.T) { + for _, op := range []string{"none", "chmod", "chown", "utimes", "setxattr"} { + t.Run(op, func(t *testing.T) { + writer := newInvalidateTestWFS(t) + attr := newInvalidateTestWFS(t) + attr.atimeMap = make(map[uint64]time.Time) + base := attributeAppendEntry(1) + base.CommitTsNs = 100 + s := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + s.blobs = map[string][]byte{base.Chunks[0].GetFileIdString(): bytes.Repeat([]byte{'O'}, 256), attributeAppendEntry(2).Chunks[1].GetFileIdString(): bytes.Repeat([]byte{'A'}, 256)} + startFakeFiler(t, writer, s) + startFakeFiler(t, attr, s) + path := util.FullPath("/dir/file") + local := proto.Clone(base).(*filer_pb.Entry) + writer.mapPbIdFromFilerToLocal(local) + fh := newFileHandle(writer, 1, 42, local) + fh.RememberPath(path) + t.Cleanup(fh.dirtyPages.Destroy) + writer.fhMap.inode2fh[42], writer.fhMap.fh2inode[1] = fh, 42 + pending := attributeAppendEntry(2).Chunks[1] + fh.UpdateEntry(func(e *filer_pb.Entry) { e.Chunks = append(e.Chunks, pending); e.Attributes.FileSize = 512 }) + fh.dirtyMetadata = true + attr.inodeToPath.Lookup("/dir", 1, true, false, 0, false) + attr.inodeToPath.Lookup(path, 1, false, false, 42, true) + attr.inodeToPath.MarkChildrenCached("/dir") + if err := attr.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", base), 1); err != nil { + t.Fatal(err) + } + code := fuse.OK + if op == "setxattr" { + code = attr.SetXAttr(nil, &fuse.SetXAttrIn{InHeader: fuse.InHeader{NodeId: 42}}, "user.test", []byte("peer acknowledged")) + } else if op != "none" { + in := &fuse.SetAttrIn{} + in.NodeId = 42 + switch op { + case "chmod": + in.Valid, in.Mode = fuse.FATTR_MODE, 0600 + case "chown": + in.Valid, in.Owner.Uid, in.Owner.Gid = fuse.FATTR_UID|fuse.FATTR_GID, 1001, 21 + case "utimes": + in.Valid, in.Mtime, in.Mtimensec = fuse.FATTR_MTIME, 123, 456 + } + code = attr.SetAttr(nil, in, &fuse.AttrOut{}) + } + if code != fuse.OK { + t.Fatalf("peer attribute: %v", code) + } + s.mu.Lock() + wantChunk, gotChunk := proto.Clone(base.Chunks[0]).(*filer_pb.FileChunk), proto.Clone(s.entry.Chunks[0]).(*filer_pb.FileChunk) + filer_pb.BeforeEntrySerialization([]*filer_pb.FileChunk{wantChunk, gotChunk}) + contentUnchanged := proto.Equal(wantChunk, gotChunk) && len(s.entry.Chunks) == 1 + ackedPeer := proto.Clone(s.entry).(*filer_pb.Entry) + s.mu.Unlock() + if !contentUnchanged { + t.Fatal("peer attribute changed content before append publication") + } + if code = writer.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: 1}); code != fuse.OK { + t.Fatalf("append after acknowledged peer %s returned %v; pending=%d base=%d rejected=%d", op, code, len(fh.entry.Chunks), len(fh.baseEntry.Load().Chunks), s.rejected) + } + s.mu.Lock() + defer s.mu.Unlock() + want := append(bytes.Repeat([]byte{'O'}, 256), bytes.Repeat([]byte{'A'}, 256)...) + if s.entry.Attributes.FileSize != 512 || !bytes.Equal(g6StoredBytes(t, s), want) { + t.Fatal("append lost original/pending bytes or size") + } + a, b := s.entry.Attributes, ackedPeer.Attributes + if op == "chmod" && a.FileMode != b.FileMode || op == "chown" && (a.Uid != b.Uid || a.Gid != b.Gid) || op == "utimes" && (a.Mtime != b.Mtime || a.MtimeNs != b.MtimeNs) || op == "setxattr" && !bytes.Equal(s.entry.Extended[XATTR_PREFIX+"user.test"], ackedPeer.Extended[XATTR_PREFIX+"user.test"]) { + t.Fatal("append rolled back acknowledged peer attributes") + } + }) + } +} + +func TestAppendCompatibleCommittedPrefix(t *testing.T) { + w := newInvalidateTestWFS(t) + base, current, pending := attributeAppendEntry(1), attributeAppendEntry(2), attributeAppendEntry(3) + current.Attributes.FileMode = 0600 + peer := &g6ConditionalPeer{entry: current, blobs: make(map[string][]byte)} + for i, c := range pending.Chunks { + peer.blobs[c.GetFileIdString()] = bytes.Repeat([]byte{[]byte{'O', 'A', 'B'}[i]}, 256) + } + startFakeFiler(t, w, peer) + w.mapPbIdFromFilerToLocal(base) + w.mapPbIdFromFilerToLocal(pending) + fh := newFileHandle(w, 1, 42, pending) + t.Cleanup(fh.dirtyPages.Destroy) + fh.RememberPath("/dir/file") + fh.baseEntry.Store(base) + fh.dirtyMetadata = true + if code := w.doFlush(context.Background(), fh, 1000, 20, false, 0); code != fuse.OK { + t.Fatal(code) + } + peer.mu.Lock() + defer peer.mu.Unlock() + want := append(bytes.Repeat([]byte{'O'}, 256), bytes.Repeat([]byte{'A'}, 256)...) + want = append(want, bytes.Repeat([]byte{'B'}, 256)...) + if peer.rejected != 1 || peer.entry.Attributes.FileMode != 0600 || peer.entry.Attributes.FileSize != 768 || !bytes.Equal(g6StoredBytes(t, peer), want) { + t.Fatal("prefix rebase lost bytes, size or peer attributes") + } +} + +func TestAsyncRefusalAfterForgetDoesNotPoisonRelookup(t *testing.T) { + testAsyncRefusalAfterForget(t, false) +} + +func TestAsyncPublicationBlockedBeforeCaptureDoesNotPoisonRelookup(t *testing.T) { + testAsyncRefusalAfterForget(t, true) +} + +func testAsyncRefusalAfterForget(t *testing.T, publicationBlocked bool) { + base := attributeAppendEntry(1) + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry), refuse: true} + entered, resume := make(chan struct{}), make(chan struct{}) + var once, resumeOnce sync.Once + release := func() { resumeOnce.Do(func() { close(resume) }) } + peer.beforeCreate = func(ctx context.Context) { + once.Do(func() { + close(entered) + select { + case <-resume: + case <-ctx.Done(): + } + }) + } + w, handle := g6AppendHandle(t, peer, base, 1000, true) + w.pendingAsyncFlush = make(map[uint64]chan struct{}) + w.startAsyncFlushWorkers(1) + t.Cleanup(w.WaitForAsyncFlush) + t.Cleanup(release) + unlockPublication := func() {} + if publicationBlocked { + unlock := w.lockMetadataPublication("/dir/file") + var unlockOnce sync.Once + unlockPublication = func() { unlockOnce.Do(unlock) } + t.Cleanup(unlockPublication) + } + if n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle, Offset: 256, Size: 256}, bytes.Repeat([]byte{'A'}, 256)); n != 256 || code != fuse.OK { + t.Fatal(n, code) + } + w.Release(nil, &fuse.ReleaseIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle}) + w.pendingAsyncFlushMu.Lock() + done := w.pendingAsyncFlush[42] + w.pendingAsyncFlushMu.Unlock() + if done == nil { + t.Fatal("Release did not queue an async publication") + } + waitEntered := func() { + t.Helper() + select { + case <-entered: + case <-time.After(2 * time.Second): + t.Fatal("async publication did not reach the filer") + } + } + if publicationBlocked { + if blocked, _ := waitFlushPublication(t, make(chan fuse.Status)); !blocked { + t.Fatal("worker was not blocked before publication capture") + } + } else { + waitEntered() + } + w.Forget(42, 1) + w.inodeToPath.Lookup("/dir/file", 1, false, false, 42, true) + if publicationBlocked { + unlockPublication() + waitEntered() + } + release() + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatal("async refusal did not finish") + } + peer.mu.Lock() + if peer.rejected != 4*(maxContentFlushMerges+1) { + t.Errorf("refusals = %d, want sixteen (four outer attempts, four merge attempts each)", peer.rejected) + } + peer.refuse = false + peer.mu.Unlock() + if _, found := w.fhMap.FindFileHandle(42); found { + t.Fatal("async close retained a handle") + } + out := &fuse.OpenOut{} + if code := w.Open(nil, &fuse.OpenIn{InHeader: fuse.InHeader{NodeId: 42}, Flags: syscall.O_WRONLY | syscall.O_APPEND}, out); code != fuse.OK { + t.Fatal(code) + } + g6AppendUploader(t, w.GetHandle(FileHandleId(out.Fh)), 1100, peer) + if n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh, Offset: 256, Size: 256}, bytes.Repeat([]byte{'B'}, 256)); n != 256 || code != fuse.OK { + t.Fatal(n, code) + } + if code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh}); code != fuse.OK { + t.Fatalf("new inode lifetime inherited stale EIO: %v", code) + } + w.Release(nil, &fuse.ReleaseIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh, ReleaseFlags: fuse.FUSE_RELEASE_FLOCK_UNLOCK}) + peer.mu.Lock() + defer peer.mu.Unlock() + want := append(bytes.Repeat([]byte{'O'}, 256), bytes.Repeat([]byte{'B'}, 256)...) + if peer.entry.Attributes.FileSize != 512 || !bytes.Equal(g6StoredBytes(t, peer), want) { + t.Fatal("relookup publication changed original/new bytes or size") + } +} + +func TestAppendAfterPendingPositionalExtensionPublishes(t *testing.T) { + base := attributeAppendEntry(1) + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + w, appendHandle := g6AppendHandle(t, peer, base, 1000) + out := &fuse.OpenOut{} + if code := w.Open(nil, &fuse.OpenIn{InHeader: fuse.InHeader{NodeId: 42}, Flags: syscall.O_WRONLY}, out); code != fuse.OK { + t.Fatal(code) + } + write := func(h, offset uint64, value byte) { + t.Helper() + n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: h, Offset: offset, Size: 256}, bytes.Repeat([]byte{value}, 256)) + if n != 256 || code != fuse.OK { + t.Fatal("write", n, code) + } + } + write(out.Fh, 256, 'A') + write(appendHandle, 512, 'B') // the kernel appends at its EOF, which includes A + if code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: appendHandle}); code != fuse.OK { + t.Fatal("append over a pending extension", code) + } + peer.mu.Lock() + defer peer.mu.Unlock() + want := append(bytes.Repeat([]byte{'O'}, 256), bytes.Repeat([]byte{'A'}, 256)...) + want = append(want, bytes.Repeat([]byte{'B'}, 256)...) + if !bytes.Equal(g6StoredBytes(t, peer), want) { + t.Fatal("pending extension or append bytes changed") + } +} + +func TestPositionalSparseWriteStillPublishes(t *testing.T) { + base := attributeAppendEntry(1) + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + w, _ := g6AppendHandle(t, peer, base, 900) + out := &fuse.OpenOut{} + if code := w.Open(nil, &fuse.OpenIn{InHeader: fuse.InHeader{NodeId: 42}, Flags: syscall.O_WRONLY}, out); code != fuse.OK { + t.Fatal(code) + } + n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh, Offset: 512, Size: 256}, bytes.Repeat([]byte{'S'}, 256)) + if n != 256 || code != fuse.OK { + t.Fatal(n, code) + } + if code = w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh}); code != fuse.OK { + t.Fatal("positional sparse write", code) + } + assertSparseWrite(t, peer) +} + +func TestSparseWriteAfterAppendFlagClearedPublishes(t *testing.T) { + base := attributeAppendEntry(1) + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + w, handle := g6AppendHandle(t, peer, base, 950) + n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle, Offset: 512, Size: 256, Flags: syscall.O_WRONLY}, bytes.Repeat([]byte{'S'}, 256)) + if n != 256 || code != fuse.OK { + t.Fatal(n, code) + } + if code = w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle}); code != fuse.OK { + t.Fatal("sparse write after clearing O_APPEND", code) + } + assertSparseWrite(t, peer) +} + +func assertSparseWrite(t *testing.T, peer *g6ConditionalPeer) { + t.Helper() + peer.mu.Lock() + defer peer.mu.Unlock() + want := append(bytes.Repeat([]byte{'O'}, 256), make([]byte, 256)...) + want = append(want, bytes.Repeat([]byte{'S'}, 256)...) + if !bytes.Equal(g6StoredBytes(t, peer), want) { + t.Fatal("sparse positional data changed") + } +} + +func TestTruncatingOpenDiscardsRefusedAppend(t *testing.T) { + base := attributeAppendEntry(1) + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + w, handle := g6AppendHandle(t, peer, base, 1700) + if n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle, Offset: 512, Size: 256}, bytes.Repeat([]byte{'A'}, 256)); n != 256 || code != fuse.OK { + t.Fatal(n, code) + } + if code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle}); code == fuse.OK { + t.Fatal("append past the committed EOF published") + } + dir, _ := w.inodeToPath.GetInode("/dir") + out := &fuse.CreateOut{} + if code := w.Create(nil, &fuse.CreateIn{InHeader: fuse.InHeader{NodeId: dir}, Flags: syscall.O_WRONLY | syscall.O_CREAT | syscall.O_TRUNC, Mode: 0644}, "file", out); code != fuse.OK { + t.Fatal("O_TRUNC create", code) + } + g6AppendUploader(t, w.GetHandle(FileHandleId(out.Fh)), 1800, peer) + payload := bytes.Repeat([]byte{'Z'}, 256) + if n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh, Size: 256}, payload); n != 256 || code != fuse.OK { + t.Fatal(n, code) + } + if code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh}); code != fuse.OK { + t.Fatal("rewrite after an O_TRUNC open", code) + } + peer.mu.Lock() + defer peer.mu.Unlock() + if !bytes.Equal(g6StoredBytes(t, peer), payload) { + t.Fatal("truncate and rewrite did not publish the new bytes") + } +} + +type g6TtlRuleFiler struct{ *g6ConditionalPeer } + +func (s *g6TtlRuleFiler) CreateEntry(ctx context.Context, r *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + if r.Entry.IsDirectory { + return &filer_pb.CreateEntryResponse{}, nil + } + r = proto.Clone(r).(*filer_pb.CreateEntryRequest) + if r.Entry.Attributes.TtlSec == 0 { + r.Entry.Attributes.TtlSec = 3600 + } + return s.g6ConditionalPeer.CreateEntry(ctx, r) +} + +func (s *g6TtlRuleFiler) UpdateEntry(ctx context.Context, r *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + if r.Entry.IsDirectory { + return &filer_pb.UpdateEntryResponse{}, nil + } + return s.g6ConditionalPeer.UpdateEntry(ctx, r) +} + +func TestEagerCreateWriteKeepsFilerDefaults(t *testing.T) { + peer := &g6ConditionalPeer{blobs: make(map[string][]byte)} + w := newInvalidateTestWFS(t) + w.option.EagerFilerCreate = true + w.option.ConcurrentWriters = 1 + w.concurrentWriters = util.NewLimitedConcurrentExecutor(1) + startFakeFiler(t, w, &g6TtlRuleFiler{peer}) + dir := w.inodeToPath.Lookup("/dir", 1, true, false, 0, false) + w.inodeToPath.MarkChildrenCached("/") + w.inodeToPath.MarkChildrenCached("/dir") + dirEntry := &filer_pb.Entry{Name: "dir", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{FileMode: uint32(os.ModeDir | 0777), Inode: dir}} + if err := w.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/", dirEntry), 1); err != nil { + t.Fatal(err) + } + out := &fuse.CreateOut{} + if code := w.Create(nil, &fuse.CreateIn{InHeader: fuse.InHeader{NodeId: dir, Caller: fuse.Caller{Owner: fuse.Owner{Uid: 1000, Gid: 20}}}, Flags: syscall.O_WRONLY, Mode: 0644}, "file", out); code != fuse.OK { + t.Fatal("create", code) + } + g6AppendUploader(t, w.GetHandle(FileHandleId(out.Fh)), 2100, peer) + payload := bytes.Repeat([]byte{'E'}, 256) + if n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: out.NodeId}, Fh: out.Fh, Size: 256}, payload); n != 256 || code != fuse.OK { + t.Fatal(n, code) + } + if code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: out.NodeId}, Fh: out.Fh}); code != fuse.OK { + t.Fatal("write after an eager create", code) + } + peer.mu.Lock() + defer peer.mu.Unlock() + if peer.entry.Attributes.TtlSec != 3600 || !bytes.Equal(g6StoredBytes(t, peer), payload) { + t.Fatal("eager create lost the stored defaults or the written bytes") + } +} + +type g6LostAnswerFiler struct { + *g6ConditionalPeer + lost atomic.Bool +} + +func (s *g6LostAnswerFiler) CreateEntry(ctx context.Context, r *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + resp, err := s.g6ConditionalPeer.CreateEntry(ctx, r) + if err == nil && resp.Error == "" && s.lost.CompareAndSwap(false, true) { + return nil, status.Error(codes.Unavailable, "answer lost after commit") + } + return resp, err +} + +func TestLostAnswerToOwnShrinkingFlushIsNotAConflict(t *testing.T) { + base := attributeAppendEntry(2) + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + w, handle := g6AppendHandle(t, peer, base, 2200) + fake := &g6LostAnswerFiler{g6ConditionalPeer: peer} + startFakeFiler(t, w, fake) + if code := w.SetAttr(nil, &fuse.SetAttrIn{SetAttrInCommon: fuse.SetAttrInCommon{InHeader: fuse.InHeader{NodeId: 42}, Valid: fuse.FATTR_SIZE | fuse.FATTR_FH, Fh: handle}}, &fuse.AttrOut{}); code != fuse.OK { + t.Fatal("ftruncate", code) + } + payload := bytes.Repeat([]byte{'N'}, 128) + if n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle, Size: 128}, payload); n != 128 || code != fuse.OK { + t.Fatal(n, code) + } + if code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle}); code != fuse.OK { + t.Fatal("own committed overwrite reported as a conflict", code) + } + peer.mu.Lock() + defer peer.mu.Unlock() + if !fake.lost.Load() || !bytes.Equal(g6StoredBytes(t, peer), payload) { + t.Fatal("lost answer was not exercised or the overwrite did not land") + } +} diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 233bc27515..736529b2e2 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -23,7 +23,7 @@ var IsDebugFileReadWrite = false type FileHandle struct { fh FileHandleId counter int64 - fhs map[FileHandleId]struct{} // live fh values, one per open; guarded by fhMap's lock + fhs map[FileHandleId]bool // live fh values, one per open; guarded by fhMap's lock entry *LockedEntry // Nested handle lock order: wfs.fhLockTable[fh.fh] -> entryLock -> // LockedEntry.RWMutex. Paths may use a subset, but must never acquire an @@ -34,15 +34,18 @@ type FileHandle struct { wfs *WFS // cache file has been written to - dirtyMetadata bool - dirtyPages *PageWriter - reader *filer.ChunkReadAt - contentType string - asyncFlushPending bool // set in writebackCache mode to defer flush to Release - asyncFlushUid uint32 // saved uid for deferred metadata flush - asyncFlushGid uint32 // saved gid for deferred metadata flush - savedDir string // last known parent path if inode-to-path state is forgotten - savedName string // last known file name if inode-to-path state is forgotten + pendingAppend bool + pendingAppendOffset uint64 + writebackContentError *atomic.Bool + dirtyMetadata bool + dirtyPages *PageWriter + reader *filer.ChunkReadAt + contentType string + asyncFlushPending bool // set in writebackCache mode to defer flush to Release + asyncFlushUid uint32 // saved uid for deferred metadata flush + asyncFlushGid uint32 // saved gid for deferred metadata flush + savedDir string // last known parent path if inode-to-path state is forgotten + savedName string // last known file name if inode-to-path state is forgotten isDeleted bool // deleteEpoch counts the times isDeleted was raised, all of them under the @@ -96,7 +99,7 @@ func newFileHandle(wfs *WFS, handleId FileHandleId, inode uint64, entry *filer_p fh := &FileHandle{ fh: handleId, counter: 1, - fhs: map[FileHandleId]struct{}{handleId: {}}, + fhs: map[FileHandleId]bool{handleId: false}, inode: inode, wfs: wfs, } @@ -337,6 +340,7 @@ func (fh *FileHandle) ResetDirtyPages() { fh.dirtyPages.Destroy() fh.dirtyPages = newPageWriter(fh, fh.wfs.option.ChunkSizeLimit) fh.dirtyMetadata = false + fh.pendingAppend = false fh.contentType = "" } diff --git a/weed/mount/filehandle_map.go b/weed/mount/filehandle_map.go index bf0c71e919..99159bf3cc 100644 --- a/weed/mount/filehandle_map.go +++ b/weed/mount/filehandle_map.go @@ -2,6 +2,7 @@ package mount import ( "sync" + "sync/atomic" "github.com/seaweedfs/seaweedfs/weed/util" @@ -65,12 +66,26 @@ func (i *FileHandleToInode) MarkInodeRenamed(inode uint64) { // open — reads on it return ENOENT, which the kernel reports to read() as EIO. // Callers must treat a draining report as "wait for the pending flush, then // retry the whole acquire", not as success. -func (i *FileHandleToInode) AcquireFileHandle(wfs *WFS, inode uint64, entry *filer_pb.Entry, versionTsNs int64, signature int32) (*FileHandle, FileHandleId, bool, bool) { +func (i *FileHandleToInode) AcquireFileHandle(wfs *WFS, inode uint64, entry *filer_pb.Entry, versionTsNs int64, signature int32, appendMode ...bool) (*FileHandle, FileHandleId, bool, bool) { + // Bind before publishing the handle or queuing any async close. Capture + // under the inode-table lock, but release it before taking the fhMap lock. + var writebackError *atomic.Bool + if wfs.option.WritebackCache { + wfs.inodeToPath.RLock() + state, found := wfs.writebackContentErrors.Load(inode) + if !found { + state, _ = wfs.writebackContentErrors.LoadOrStore(inode, new(atomic.Bool)) + } + writebackError = state.(*atomic.Bool) + wfs.inodeToPath.RUnlock() + } i.Lock() defer i.Unlock() fh, found := i.inode2fh[inode] if !found { fh = newFileHandle(wfs, FileHandleId(util.RandomUint64()), inode, entry) + fh.writebackContentError = writebackError + fh.fhs[fh.fh] = len(appendMode) > 0 && appendMode[0] fh.entryVersionTsNs.Store(versionTsNs) fh.entryVersionSignature.Store(signature) i.inode2fh[inode] = fh @@ -82,11 +97,20 @@ func (i *FileHandleToInode) AcquireFileHandle(wfs *WFS, inode uint64, entry *fil } fh.counter++ openFh := FileHandleId(util.RandomUint64()) - fh.fhs[openFh] = struct{}{} + fh.fhs[openFh] = len(appendMode) > 0 && appendMode[0] i.fh2inode[openFh] = inode return fh, openFh, true, false } +func (i *FileHandleToInode) isAppendHandle(openFh FileHandleId) bool { + i.RLock() + defer i.RUnlock() + if fh := i.inode2fh[i.fh2inode[openFh]]; fh != nil { + return fh.fhs[openFh] + } + return false +} + func (i *FileHandleToInode) ReleaseByHandle(fh FileHandleId) *FileHandle { i.Lock() defer i.Unlock() diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 2455ea5e49..82f7f77cc3 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -133,6 +133,8 @@ type Option struct { } type WFS struct { + // Metadata-only refusal token, detached at final Forget; no live resources retained. + writebackContentErrors sync.Map // https://dl.acm.org/doi/fullHtml/10.1145/3310148 // follow https://github.com/hanwen/go-fuse/blob/master/fuse/api.go fuse.RawFileSystem diff --git a/weed/mount/weedfs_async_flush.go b/weed/mount/weedfs_async_flush.go index d3e7cbfa54..1ed0d54beb 100644 --- a/weed/mount/weedfs_async_flush.go +++ b/weed/mount/weedfs_async_flush.go @@ -61,6 +61,7 @@ func (wfs *WFS) completeAsyncFlush(fh *FileHandle) { // is a permanent one and the chunk data has been freed. if err := fh.dirtyPages.FlushData(); err != nil { glog.Errorf("completeAsyncFlush inode %d: data flush failed: %v", fh.inode, err) + (&contentPublication{failure: fh.writebackContentError}).abandon() // Data is lost at this point. Proceed to cleanup to avoid resource leaks and unmount hangs. } else if fh.dirtyMetadata { // Phase 2: Flush metadata unless the file was explicitly unlinked. @@ -139,10 +140,11 @@ func isFilerUnreachable(err error) bool { // nothing else holds these references: like the writeback chunk uploads, it // keeps retrying for as long as the filer is unreachable. func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFullPath util.FullPath) error { + publication := &contentPublication{failure: fh.writebackContentError} attempts := 0 err := retryMetadataFlushUntil(context.Background(), func() error { attempts++ - return wfs.flushMetadataToFiler(context.Background(), fh, dir, name, fh.asyncFlushUid, fh.asyncFlushGid, nil) + return wfs.flushMetadataToFiler(context.Background(), fh, dir, name, fh.asyncFlushUid, fh.asyncFlushGid, nil, publication) }, isFilerUnreachable, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { if totalAttempts == 0 { glog.Warningf("completeAsyncFlush %s: filer unreachable, retrying metadata flush (attempt %d) after %v: %v", @@ -153,6 +155,7 @@ func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFul fileFullPath, nextAttempt, totalAttempts, backoff, err) }) if err != nil { + publication.abandon() glog.Errorf("completeAsyncFlush %s: metadata flush failed after %d attempts: %v - "+ "chunks are uploaded but NOT referenced in filer metadata; "+ "they will appear as orphans in volume.fsck", diff --git a/weed/mount/weedfs_async_flush_test.go b/weed/mount/weedfs_async_flush_test.go index c1578b2da9..79772343ab 100644 --- a/weed/mount/weedfs_async_flush_test.go +++ b/weed/mount/weedfs_async_flush_test.go @@ -80,6 +80,94 @@ func TestFlushMetadataWithRetryStopsOnFilerRefusal(t *testing.T) { } } +// Under writeback the kernel already moved EOF past the stored entry, so a +// drain that gives up must fence the inode whatever the filer answered. +func TestAbandonedWritebackFlushFencesInode(t *testing.T) { + originalSleep := metadataFlushSleep + t.Cleanup(func() { + metadataFlushSleep = originalSleep + }) + metadataFlushSleep = func(ctx context.Context, d time.Duration) {} + + const name = "abandoned-writeback.log" + wfs, _ := newCreateTestWFS(t) + fake := &rejectingCreateFiler{rejections: 1 << 20, refuse: func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + return nil, status.Error(codes.PermissionDenied, "denied") + }} + startFakeFiler(t, wfs, fake) + + out := &fuse.CreateOut{} + if code := wfs.Create(make(chan struct{}), &fuse.CreateIn{ + InHeader: fuse.InHeader{NodeId: 1}, + Flags: syscall.O_WRONLY | syscall.O_CREAT, + Mode: 0o644, + }, name, out); code != fuse.OK { + t.Fatalf("Create status = %v, want OK", code) + } + fh := wfs.GetHandle(FileHandleId(out.Fh)) + if fh == nil { + t.Fatal("Create did not register an open file handle") + } + wfs.option.WritebackCache = true + fh.writebackContentError = new(atomic.Bool) + fh.dirtyMetadata = true + + if err := wfs.flushMetadataWithRetry(fh, "/", name, util.FullPath("/").Child(name)); err == nil { + t.Fatal("flush against a refusing filer succeeded") + } + if !fh.writebackContentError.Load() { + t.Fatal("abandoned writeback flush left the inode unfenced") + } + if err := wfs.checkContentPublication(fh, nil); !errors.Is(err, filer_pb.ErrPreconditionFailed) { + t.Fatalf("checkContentPublication after abandonment = %v, want a precondition refusal", err) + } +} + +func TestFailedFlockReleaseFencesInode(t *testing.T) { + originalSleep := metadataFlushSleep + t.Cleanup(func() { + metadataFlushSleep = originalSleep + }) + metadataFlushSleep = func(ctx context.Context, d time.Duration) {} + + const name = "flock-release.log" + wfs, _ := newCreateTestWFS(t) + wfs.posixLocks = NewPosixLockTable() + fake := &rejectingCreateFiler{rejections: 1 << 20, refuse: func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + return nil, status.Error(codes.PermissionDenied, "denied") + }} + startFakeFiler(t, wfs, fake) + + out := &fuse.CreateOut{} + if code := wfs.Create(make(chan struct{}), &fuse.CreateIn{ + InHeader: fuse.InHeader{NodeId: 1}, + Flags: syscall.O_WRONLY | syscall.O_CREAT, + Mode: 0o644, + }, name, out); code != fuse.OK { + t.Fatalf("Create status = %v, want OK", code) + } + fh := wfs.GetHandle(FileHandleId(out.Fh)) + if fh == nil { + t.Fatal("Create did not register an open file handle") + } + wfs.option.WritebackCache = true + fenced := new(atomic.Bool) + fh.writebackContentError = fenced + fh.dirtyMetadata = true + + wfs.Release(make(chan struct{}), &fuse.ReleaseIn{ + InHeader: fuse.InHeader{NodeId: out.NodeId}, + Fh: out.Fh, + ReleaseFlags: fuse.FUSE_RELEASE_FLOCK_UNLOCK, + }) + if fake.calls.Load() == 0 { + t.Fatal("synchronous release did not attempt the flush") + } + if !fenced.Load() { + t.Fatal("release dropped uncommitted writeback content without fencing the inode") + } +} + func TestIsFilerUnreachable(t *testing.T) { path := "/unavailable/transport.log" cases := []struct { diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go index f55c0473ff..58268ddce2 100644 --- a/weed/mount/weedfs_file_mkrm.go +++ b/weed/mount/weedfs_file_mkrm.go @@ -102,7 +102,7 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o var openFh FileHandleId var existed, draining bool for { - fileHandle, openFh, existed, draining = wfs.fhMap.AcquireFileHandle(wfs, inode, newEntry, 0, 0) + fileHandle, openFh, existed, draining = wfs.fhMap.AcquireFileHandle(wfs, inode, newEntry, 0, 0, in.Flags&syscall.O_APPEND != 0) if !draining { break } diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index a69016a102..8fbd26ef3a 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -222,14 +222,16 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al commit = &commitBudget{left: metadataTimeout} } + publication := &contentPublication{failure: fh.writebackContentError} if err := retryMetadataFlushIf(ctx, func() error { - return wfs.flushMetadataToFiler(ctx, fh, dir, name, uid, gid, commit) + return wfs.flushMetadataToFiler(ctx, fh, dir, name, uid, gid, commit, publication) }, func(err error) bool { return !commit.spent() && isRetryableFilerError(err) }, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { glog.Warningf("%v fh %d flush: retrying metadata flush (attempt %d/%d) after %v: %v", fileFullPath, fh.fh, nextAttempt, totalAttempts, backoff, err) }); err != nil { + publication.retire(err) glog.Errorf("%v fh %d flush: %v", fileFullPath, fh.fh, err) return grpcErrorToFuseStatus(err) } @@ -250,7 +252,11 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al // // commit bounds the filer calls made here, over all attempts of one flush; nil // leaves them to the per-call filer timeouts. -func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, name string, uid, gid uint32, commit *commitBudget) error { +func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, name string, uid, gid uint32, commit *commitBudget, publications ...*contentPublication) error { + publication := &contentPublication{} + if len(publications) > 0 { + publication = publications[0] + } fileFullPath, dir, name := wfs.flushTarget(fh, dir, name) var unlock func() var fhActiveLock *util.ActiveLock @@ -355,7 +361,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n // The chunk merge and the manifest above read and upload chunks, which can // take longer than the whole commit budget. It is charged only from here. commitCtx, end := commit.begin(ctx) - resp, merged, err := wfs.createContentEntry(commitCtx, fh, request, baseSnapshot, false) + resp, merged, err := wfs.createContentEntry(commitCtx, fh, request, baseSnapshot, false, publication) end() if err != nil { glog.Errorf("fh flush create %s: %v", fileFullPath, err) @@ -384,6 +390,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n } if err == nil { + fh.pendingAppend = false fh.dirtyMetadata = false } @@ -413,7 +420,7 @@ func (wfs *WFS) flushTarget(fh *FileHandle, dir, name string) (util.FullPath, st // replayed onto it and published against it, so a peer's acknowledged chunks // survive. merged is that entry in local ids; nil when the request went as is. // With writes still buffered, the merge must keep room for them. -func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request *filer_pb.CreateEntryRequest, local *filer_pb.Entry, buffered bool) (resp *filer_pb.CreateEntryResponse, merged *filer_pb.Entry, err error) { +func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request *filer_pb.CreateEntryRequest, local *filer_pb.Entry, buffered bool, publication *contentPublication) (resp *filer_pb.CreateEntryResponse, merged *filer_pb.Entry, err error) { base := fh.baseEntry.Load() guard := &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS} if base != nil { @@ -423,10 +430,16 @@ func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request } sent := request.Entry for attempt := 0; ; attempt++ { + if guardErr := wfs.checkContentPublication(fh, guard.ExpectedEntry); guardErr != nil { + return nil, nil, guardErr + } request.Condition = &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{guard}} err = wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { var callErr error resp, callErr = filer_pb.CreateEntryWithResponse(ctx, client, request) + if errors.Is(callErr, filer_pb.ErrPreconditionFailed) { + publication.refused = true + } return callErr }) if err == nil && request.Entry != sent { @@ -583,6 +596,10 @@ func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo in size = int64(localSize) } merged.Attributes.FileSize = uint64(size) + // A peer truncate must not turn local pending data into a successful hole. + if filer.FileSize(stored) < filer.FileSize(base) && contentGapAfter(merged, filer.FileSize(stored)) { + return nil, false + } return merged, true } diff --git a/weed/mount/weedfs_file_write.go b/weed/mount/weedfs_file_write.go index 9c10a0df64..38866dc803 100644 --- a/weed/mount/weedfs_file_write.go +++ b/weed/mount/weedfs_file_write.go @@ -10,6 +10,14 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" ) +// isAppendWrite honors current F_SETFL flags; adapters without flags use the open. +func (wfs *WFS) isAppendWrite(in *fuse.WriteIn) bool { + if in.Flags != 0 { + return in.Flags&syscall.O_APPEND != 0 + } + return wfs.fhMap.isAppendHandle(FileHandleId(in.Fh)) +} + /** * Write data * @@ -60,6 +68,13 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr return 0, fuse.OK } + // Only an append past all locally owned bytes needs the stored EOF guard. + if in.Offset > entry.Attributes.FileSize && wfs.isAppendWrite(in) { + if !fh.pendingAppend || in.Offset < fh.pendingAppendOffset { + fh.pendingAppendOffset = in.Offset + } + fh.pendingAppend = true + } entry.Content = nil offset := int64(in.Offset) oldFileSize := int64(entry.Attributes.FileSize) diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go index 1809f53b88..16af2b9e40 100644 --- a/weed/mount/weedfs_filehandle.go +++ b/weed/mount/weedfs_filehandle.go @@ -70,7 +70,7 @@ func (wfs *WFS) AcquireHandle(ctx context.Context, inode uint64, flags, uid, gid } // need to AcquireFileHandle again to ensure correct handle counter var existed, draining bool - fileHandle, openFh, existed, draining = wfs.fhMap.AcquireFileHandle(wfs, inode, entry, entryVersionTsNs.tsNs, entryVersionTsNs.signature) + fileHandle, openFh, existed, draining = wfs.fhMap.AcquireFileHandle(wfs, inode, entry, entryVersionTsNs.tsNs, entryVersionTsNs.signature, flags&syscall.O_APPEND != 0) if draining { // A RELEASE slipped between waitForPendingAsyncFlush and this // attach: the handle's last open closed and its deferred flush @@ -183,6 +183,9 @@ func (wfs *WFS) ReleaseHandle(handleId FileHandleId) { wfs.pendingAsyncFlushMu.Unlock() if fhToRelease != nil { + if fhToRelease.dirtyMetadata && !fhToRelease.isDeleted { + (&contentPublication{failure: fhToRelease.writebackContentError}).abandon() + } fhToRelease.ReleaseHandle() } } diff --git a/weed/mount/weedfs_forget.go b/weed/mount/weedfs_forget.go index ab9d1ac594..a8d10643dc 100644 --- a/weed/mount/weedfs_forget.go +++ b/weed/mount/weedfs_forget.go @@ -68,7 +68,10 @@ func (wfs *WFS) Forget(nodeid, nlookup uint64) { wfs.inodeToPath.Forget(nodeid, nlookup, // Runs at the release, under the table's lock, so a lookup that // rebuilds the same inode number afterwards keeps the times it sets. - wfs.forgetInMemoryTimes, + func(inode uint64) { + wfs.forgetInMemoryTimes(inode) + wfs.writebackContentErrors.Delete(inode) + }, func(dir util.FullPath) { // Runs after Forget releases its lock; a concurrent lookup+rebuild can // re-cache the directory in that window, so purge through the apply loop diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go index e9ebc38e3b..56ba8a36c9 100644 --- a/weed/mount/weedfs_metadata_flush.go +++ b/weed/mount/weedfs_metadata_flush.go @@ -188,7 +188,7 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error { baseSnapshot := proto.Clone(requestEntry).(*filer_pb.Entry) wfs.mapPbIdFromLocalToFiler(request.Entry) - resp, merged, err := wfs.createContentEntry(context.Background(), fh, request, baseSnapshot, true) + resp, merged, err := wfs.createContentEntry(context.Background(), fh, request, baseSnapshot, true, &contentPublication{}) if err != nil { return err }