diff --git a/weed/server/filer_grpc_server_condition_test.go b/weed/server/filer_grpc_server_condition_test.go index c930ec80e1..3673c76b55 100644 --- a/weed/server/filer_grpc_server_condition_test.go +++ b/weed/server/filer_grpc_server_condition_test.go @@ -434,3 +434,35 @@ func TestDeleteCutoff(t *testing.T) { t.Fatal("entry past the guarded far-future second must be treated as newer") } } + +func TestDeleteEntryKeepsAnEntryModifiedAfterTheCutoff(t *testing.T) { + listed := time.Unix(1700000000, 500) + for _, tc := range []struct { + name string + req *filer_pb.DeleteEntryRequest + mtime time.Time + kept bool + }{ + {"ns cutoff, unchanged", &filer_pb.DeleteEntryRequest{IfNotModifiedAfterNs: listed.UnixNano()}, listed, false}, + {"ns cutoff, modified since", &filer_pb.DeleteEntryRequest{IfNotModifiedAfterNs: listed.UnixNano()}, listed.Add(time.Nanosecond), true}, + {"seconds cutoff, same second", &filer_pb.DeleteEntryRequest{IfNotModifiedAfter: listed.Unix()}, listed.Add(time.Millisecond), false}, + {"seconds cutoff, modified since", &filer_pb.DeleteEntryRequest{IfNotModifiedAfter: listed.Unix()}, listed.Add(time.Second), true}, + } { + t.Run(tc.name, func(t *testing.T) { + store := newRenameTestStore() + store.entries["/test/obj"] = &filer.Entry{ + FullPath: "/test/obj", + Attr: filer.Attr{Inode: 1, Mtime: tc.mtime, Mode: 0644}, + } + fs := &FilerServer{filer: newRenameTestFiler(t, store), option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()} + tc.req.Directory, tc.req.Name = "/test", "obj" + resp, err := fs.DeleteEntry(context.Background(), tc.req) + if err != nil || resp.Error != "" { + t.Fatalf("DeleteEntry: %v %q", err, resp.GetError()) + } + if _, kept := store.entries["/test/obj"]; kept != tc.kept { + t.Fatalf("kept = %v, want %v", kept, tc.kept) + } + }) + } +} diff --git a/weed/shell/command_remote_cache.go b/weed/shell/command_remote_cache.go index 69ff9bcfec..d2d74802f5 100644 --- a/weed/shell/command_remote_cache.go +++ b/weed/shell/command_remote_cache.go @@ -2,6 +2,7 @@ package shell import ( "context" + "errors" "flag" "fmt" "io" @@ -260,7 +261,7 @@ func (c *commandRemoteCache) doComprehensiveSync(commandEnv *CommandEnv, writer isDirectory := remoteEntry.RemoteSize == 0 && remoteEntry.RemoteMtime == 0 if lookupErr == filer_pb.ErrNotFound { // Create new entry - _, createErr := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{ + createErr := filer_pb.CreateEntry(ctx, client, &filer_pb.CreateEntryRequest{ Directory: string(localDir), Entry: &filer_pb.Entry{ Name: name, @@ -273,7 +274,12 @@ func (c *commandRemoteCache) doComprehensiveSync(commandEnv *CommandEnv, writer Extended: filer.MergeRemoteContentEncoding(remoteEntry, nil), RemoteEntry: remoteEntry, }, + OExcl: true, }) + if errors.Is(createErr, filer_pb.ErrEntryAlreadyExists) { + fmt.Fprintf(writer, "skipped (created since it was looked up)\n") + continue + } if createErr != nil { fmt.Fprintf(writer, "failed to create: %v\n", createErr) continue @@ -287,18 +293,11 @@ func (c *commandRemoteCache) doComprehensiveSync(commandEnv *CommandEnv, writer continue } - existingEntry.RemoteEntry = remoteEntry - existingEntry.Attributes.FileSize = uint64(remoteEntry.RemoteSize) - existingEntry.Attributes.Mtime = remoteEntry.RemoteMtime - existingEntry.Attributes.Md5 = nil - existingEntry.Extended = filer.MergeRemoteContentEncoding(remoteEntry, existingEntry.Extended) - existingEntry.Chunks = nil - existingEntry.Content = nil - - _, updateErr := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{ - Directory: string(localDir), - Entry: existingEntry, - }) + stale, updateErr := doSaveRemoteEntry(client, string(localDir), existingEntry, remoteEntry) + if stale { + fmt.Fprintf(writer, "skipped (changed since it was looked up)\n") + continue + } if updateErr != nil { fmt.Fprintf(writer, "failed to update: %v\n", updateErr) continue diff --git a/weed/shell/command_remote_meta_sync.go b/weed/shell/command_remote_meta_sync.go index ca9320d3f5..90c9888eae 100644 --- a/weed/shell/command_remote_meta_sync.go +++ b/weed/shell/command_remote_meta_sync.go @@ -2,6 +2,7 @@ package shell import ( "context" + "errors" "flag" "fmt" "io" @@ -214,10 +215,13 @@ func pullMetadataDirectory(ctx context.Context, client filer_pb.SeaweedFilerClie continue } } else { - fmt.Fprintf(writer, "%s (delete)\n", localPath) - if err := deleteLocalEntry(ctx, client, localDir, name); err != nil { + kept, err := deleteLocalEntry(ctx, client, writer, localDir, existingEntry) + if err != nil { return err } + if kept { + continue + } } existingEntry = nil } @@ -240,9 +244,20 @@ func pullMetadataDirectory(ctx context.Context, client filer_pb.SeaweedFilerClie } if existingEntry == nil { - if err := createRemoteEntry(ctx, client, writer, localDir, name, child, remoteLoc.Name); err != nil { + taken, err := createRemoteEntry(ctx, client, writer, localDir, name, child, remoteLoc.Name) + if err != nil { return err } + if taken && child.isDirectory { + // descending into a name a local file took would turn it into a directory + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: string(localDir), Name: name}) + if err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + return err + } + if resp == nil || !resp.Entry.IsDirectory { + continue + } + } } else if !child.isDirectory { if existingEntry.RemoteEntry == nil { // a local change that should not be overwritten @@ -255,9 +270,13 @@ func pullMetadataDirectory(ctx context.Context, client filer_pb.SeaweedFilerClie // persist the stat-derived encoding so the next run skips the stat (statFiles && existingEntry.RemoteEntry.RemoteContentEncoding == nil && child.remoteEntry.RemoteContentEncoding != nil) { - fmt.Fprintf(writer, "%s (update)\n", localPath) - if err := doSaveRemoteEntry(client, string(localDir), existingEntry, child.remoteEntry); err != nil { + stale, err := doSaveRemoteEntry(client, string(localDir), existingEntry, child.remoteEntry) + if stale { + fmt.Fprintf(writer, "%s (skip, changed since listed)\n", localPath) + } else if err != nil { return err + } else { + fmt.Fprintf(writer, "%s (update)\n", localPath) } } else { fmt.Fprintf(writer, "%s (skip)\n", localPath) @@ -297,8 +316,8 @@ func deleteRemoteBackedEntry(ctx context.Context, client filer_pb.SeaweedFilerCl if existingEntry.RemoteEntry == nil { return nil } - fmt.Fprintf(writer, "%s (delete)\n", localPath) - return deleteLocalEntry(ctx, client, localDir, name) + _, err := deleteLocalEntry(ctx, client, writer, localDir, existingEntry) + return err } children, err := listLocalDirectory(ctx, client, localPath) @@ -319,14 +338,35 @@ func deleteRemoteBackedEntry(ctx context.Context, client filer_pb.SeaweedFilerCl return err } if empty { - fmt.Fprintf(writer, "%s (delete)\n", localPath) - return deleteLocalEntry(ctx, client, localDir, name) + _, err = deleteLocalEntry(ctx, client, writer, localDir, existingEntry) } - return nil + return err } -func deleteLocalEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, localDir util.FullPath, name string) error { - return filer_pb.DoRemove(ctx, client, string(localDir), name, true, false, false, false, nil) +// deleteLocalEntry removes the entry as it was listed; the filer keeps an entry +// whose mtime a local write has moved past the listed one, reported as kept. +// An entry listed without an mtime is kept once a write sets one. +func deleteLocalEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, writer io.Writer, localDir util.FullPath, entry *filer_pb.Entry) (kept bool, err error) { + req := &filer_pb.DeleteEntryRequest{Directory: string(localDir), Name: entry.Name, IsDeleteData: true} + var ok bool + req.IfNotModifiedAfter, req.IfNotModifiedAfterNs, ok = deleteEntryCutoff(entry.Attributes.GetMtime(), entry.Attributes.GetMtimeNs()) + if !ok { + req.IfNotModifiedAfterNs = max(1, int64(entry.Attributes.GetMtimeNs())) + } + if _, err = filer_pb.DoRemoveRequest(ctx, client, req); err != nil { + return false, err + } + localPath := localDir.Child(entry.Name) + _, err = filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: string(localDir), Name: entry.Name}) + if err == nil { + fmt.Fprintf(writer, "%s (skip, changed since listed)\n", localPath) + return true, nil + } + if !errors.Is(err, filer_pb.ErrNotFound) { + return false, err + } + fmt.Fprintf(writer, "%s (delete)\n", localPath) + return false, nil } func isLocalDirectoryEmpty(ctx context.Context, client filer_pb.SeaweedFilerClient, dir util.FullPath) (bool, error) { @@ -338,7 +378,9 @@ func isLocalDirectoryEmpty(ctx context.Context, client filer_pb.SeaweedFilerClie return empty, err } -func createRemoteEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, writer io.Writer, localDir util.FullPath, name string, child *remoteChild, storageName string) error { +// createRemoteEntry creates the entry only if the name is still free; taken +// reports that something was created there after the directory was listed. +func createRemoteEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, writer io.Writer, localDir util.FullPath, name string, child *remoteChild, storageName string) (taken bool, err error) { attributes := &filer_pb.FuseAttributes{ FileMode: remoteEntryFileMode(child.isDirectory), TtlSec: 0, // Remote entries should not have TTL @@ -356,7 +398,7 @@ func createRemoteEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, attributes.FileSize = uint64(child.remoteEntry.RemoteSize) attributes.Mtime = child.remoteEntry.RemoteMtime } - _, err := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{ + err = filer_pb.CreateEntry(ctx, client, &filer_pb.CreateEntryRequest{ Directory: string(localDir), Entry: &filer_pb.Entry{ Name: name, @@ -365,9 +407,17 @@ func createRemoteEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, Extended: filer.MergeRemoteContentEncoding(remoteEntry, nil), RemoteEntry: remoteEntry, }, + OExcl: true, }) + if errors.Is(err, filer_pb.ErrEntryAlreadyExists) { + fmt.Fprintf(writer, "%s (skip, created since listed)\n", localDir.Child(name)) + return true, nil + } + if err != nil { + return false, err + } fmt.Fprintf(writer, "%s (create)\n", localDir.Child(name)) - return err + return false, nil } func childRemoteLocation(remoteLoc *remote_pb.RemoteStorageLocation, name string) *remote_pb.RemoteStorageLocation { diff --git a/weed/shell/command_remote_meta_sync_test.go b/weed/shell/command_remote_meta_sync_test.go new file mode 100644 index 0000000000..8b0782899d --- /dev/null +++ b/weed/shell/command_remote_meta_sync_test.go @@ -0,0 +1,245 @@ +package shell + +import ( + "context" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +type exclusiveCreateFiler struct { + filer_pb.SeaweedFilerClient + resp *filer_pb.CreateEntryResponse + req *filer_pb.CreateEntryRequest +} + +func (s *exclusiveCreateFiler) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest, _ ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) { + s.req = req + return s.resp, nil +} + +// meta.sync updates entries from a directory listing taken once per +// directory; a file written locally after that listing must not be reset to +// the remote pointer. +func TestDoSaveRemoteEntryIsConditionedOnTheListedEntry(t *testing.T) { + listed := &filer_pb.Entry{ + Name: "f", + Attributes: &filer_pb.FuseAttributes{FileSize: 3, Mtime: 10}, + Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 1, Cookie: 7}}}, + RemoteEntry: &filer_pb.RemoteEntry{StorageName: "r", RemoteETag: "old", RemoteSize: 3, RemoteMtime: 10}, + } + remote := &filer_pb.RemoteEntry{StorageName: "r", RemoteETag: "new", RemoteSize: 5, RemoteMtime: 20} + + for _, tc := range []struct { + name string + err error + stale bool + }{ + {"unchanged", nil, false}, + {"written since listed", status.Error(codes.FailedPrecondition, "precondition failed"), true}, + {"deleted since listed", status.Error(codes.NotFound, "not found"), true}, + } { + t.Run(tc.name, func(t *testing.T) { + fake := &conditionalUpdateFiler{err: tc.err} + stale, err := doSaveRemoteEntry(fake, "/buckets/b", proto.Clone(listed).(*filer_pb.Entry), remote) + if stale != tc.stale || (err != nil) != (tc.err != nil) { + t.Fatalf("stale=%v err=%v, want stale=%v err=%v", stale, err, tc.stale, tc.err) + } + clauses := fake.req.GetCondition().GetClauses() + if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL || !proto.Equal(clauses[0].ExpectedEntry, listed) { + t.Fatalf("condition %v must expect the listed entry", fake.req.GetCondition()) + } + written := fake.req.Entry + if len(written.Chunks) != 0 || !proto.Equal(written.RemoteEntry, remote) || written.Attributes.FileSize != 5 { + t.Fatalf("written entry %v must point at the new remote copy", written) + } + }) + } +} + +// A name missing from the listing may be created locally before meta.sync +// gets to it; the remote-only entry must not replace it. +func TestCreateRemoteEntryKeepsAnEntryCreatedSinceListed(t *testing.T) { + child := &remoteChild{remoteEntry: &filer_pb.RemoteEntry{StorageName: "r", RemoteSize: 5, RemoteMtime: 20}} + + for _, tc := range []struct { + name string + resp *filer_pb.CreateEntryResponse + output string + err bool + }{ + {"absent", &filer_pb.CreateEntryResponse{}, "/buckets/b/f (create)", false}, + {"created since listed", &filer_pb.CreateEntryResponse{Error: "exists", ErrorCode: filer_pb.FilerError_ENTRY_ALREADY_EXISTS}, "/buckets/b/f (skip, created since listed)", false}, + {"refused", &filer_pb.CreateEntryResponse{Error: "read only", ErrorCode: filer_pb.FilerError_READ_ONLY}, "", true}, + } { + t.Run(tc.name, func(t *testing.T) { + fake := &exclusiveCreateFiler{resp: tc.resp} + var out strings.Builder + taken, err := createRemoteEntry(context.Background(), fake, &out, util.FullPath("/buckets/b"), "f", child, "r") + if (err != nil) != tc.err || taken != (tc.resp.ErrorCode == filer_pb.FilerError_ENTRY_ALREADY_EXISTS) { + t.Fatalf("taken=%v err=%v, want error %v", taken, err, tc.err) + } + if !fake.req.OExcl { + t.Fatalf("create of %s must be exclusive", fake.req.Entry.Name) + } + if got := strings.TrimSpace(out.String()); got != tc.output { + t.Fatalf("output %q, want %q", got, tc.output) + } + }) + } +} + +type takenNameFiler struct { + stubFilerSvc + creates []string +} + +func (s *takenNameFiler) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest, _ ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) { + s.creates = append(s.creates, string(util.FullPath(req.Directory).Child(req.Entry.Name))) + return &filer_pb.CreateEntryResponse{Error: "exists", ErrorCode: filer_pb.FilerError_ENTRY_ALREADY_EXISTS}, nil +} + +type listingRemote struct { + remote_storage.RemoteStorageClient + dirs map[string]map[string]bool +} + +func (r *listingRemote) ListDirectory(_ context.Context, loc *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) error { + for name, isDirectory := range r.dirs[loc.Path] { + if err := visitFn(loc.Path, name, isDirectory, &filer_pb.RemoteEntry{StorageName: "r", RemoteSize: 1, RemoteMtime: 1}); err != nil { + return err + } + } + return nil +} + +// A remote directory whose name is taken locally after the listing is only +// descended into when the name now holds a directory: creating remote +// children under a local file would turn that file into a directory. +func TestPullMetadataDirectoryDescendsOnlyIntoATakenDirectory(t *testing.T) { + remote := &listingRemote{dirs: map[string]map[string]bool{ + "/": {"docs": true}, + "/docs": {"a.txt": false}, + }} + for _, tc := range []struct { + name string + isDir bool + creates []string + }{ + {"taken by a file", false, []string{"/buckets/b/docs"}}, + {"taken by a directory", true, []string{"/buckets/b/docs", "/buckets/b/docs/a.txt"}}, + } { + t.Run(tc.name, func(t *testing.T) { + fake := &takenNameFiler{stubFilerSvc: stubFilerSvc{lookupResp: &filer_pb.LookupDirectoryEntryResponse{ + Entry: &filer_pb.Entry{Name: "docs", IsDirectory: tc.isDir}, + }}} + var out strings.Builder + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "rb", Path: "/"} + if err := pullMetadataDirectory(context.Background(), fake, &out, remote, util.FullPath("/buckets/b"), loc, false, false); err != nil { + t.Fatalf("sync: %v", err) + } + if strings.Join(fake.creates, ",") != strings.Join(tc.creates, ",") { + t.Fatalf("creates %v, want %v", fake.creates, tc.creates) + } + }) + } +} + +type deleteRecordingFiler struct { + takenNameFiler + deletes []*filer_pb.DeleteEntryRequest + listing map[string][]*filer_pb.Entry +} + +func (s *deleteRecordingFiler) ListEntries(_ context.Context, req *filer_pb.ListEntriesRequest, _ ...grpc.CallOption) (filer_pb.SeaweedFiler_ListEntriesClient, error) { + return &stubListStream{entries: s.listing[req.Directory]}, nil +} + +func (s *deleteRecordingFiler) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest, _ ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) { + s.deletes = append(s.deletes, req) + return &filer_pb.DeleteEntryResponse{}, nil +} + +// A remote-backed entry that meta.sync drops may be replaced locally after the +// directory listing; the delete must only remove the listed version. +func TestPullMetadataDirectoryDeletesOnlyTheListedFile(t *testing.T) { + remoteFile := &filer_pb.RemoteEntry{StorageName: "r", RemoteSize: 1, RemoteMtime: 10} + for _, tc := range []struct { + name string + listed *filer_pb.Entry + remote map[string]bool + deleteStale bool + sec, ns int64 + }{ + {"remote turned it into a directory", &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{Mtime: 10, MtimeNs: 5}, RemoteEntry: remoteFile}, map[string]bool{"f": true}, false, 10, 10e9 + 5}, + {"gone from the remote", &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{Mtime: 10, MtimeNs: 5}, RemoteEntry: remoteFile}, map[string]bool{}, true, 10, 10e9 + 5}, + {"listed without an mtime", &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{}, RemoteEntry: remoteFile}, map[string]bool{}, true, 0, 1}, + {"listed at the epoch with nanoseconds", &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{MtimeNs: 500}, RemoteEntry: remoteFile}, map[string]bool{}, true, 0, 500}, + {"empty directory gone from the remote", &filer_pb.Entry{Name: "f", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mtime: 20}, RemoteEntry: &filer_pb.RemoteEntry{StorageName: "r"}}, map[string]bool{}, true, 20, 20e9}, + } { + t.Run(tc.name, func(t *testing.T) { + fake := &deleteRecordingFiler{ + takenNameFiler: takenNameFiler{stubFilerSvc: stubFilerSvc{ + lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: &filer_pb.Entry{Name: "f"}}, + }}, + listing: map[string][]*filer_pb.Entry{"/buckets/b": {tc.listed}}, + } + remote := &listingRemote{dirs: map[string]map[string]bool{"/": tc.remote}} + var out strings.Builder + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "rb", Path: "/"} + if err := pullMetadataDirectory(context.Background(), fake, &out, remote, util.FullPath("/buckets/b"), loc, tc.deleteStale, false); err != nil { + t.Fatalf("sync: %v", err) + } + if len(fake.deletes) != 1 || fake.deletes[0].Name != "f" || fake.deletes[0].IfNotModifiedAfter != tc.sec || fake.deletes[0].IfNotModifiedAfterNs != tc.ns { + t.Fatalf("deletes %v must be conditioned on the listed mtime", fake.deletes) + } + }) + } +} + +// The filer answers a delete it skipped under the cutoff with success, so the +// sync reports what a lookup after the delete finds. +func TestPullMetadataDirectoryReportsAKeptEntry(t *testing.T) { + listed := &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{Mtime: 10}, RemoteEntry: &filer_pb.RemoteEntry{StorageName: "r", RemoteSize: 1, RemoteMtime: 10}} + for _, tc := range []struct { + name string + remote map[string]bool + deleteStale bool + kept bool + want string + creates int + }{ + {"stale file kept", map[string]bool{}, true, true, "/buckets/b/f (skip, changed since listed)\n", 0}, + {"stale file deleted", map[string]bool{}, true, false, "/buckets/b/f (delete)\n", 0}, + {"file replaced by a remote directory kept", map[string]bool{"f": true}, false, true, "/buckets/b/f (skip, changed since listed)\n", 0}, + {"file replaced by a remote directory deleted", map[string]bool{"f": true}, false, false, "/buckets/b/f (delete)\n/buckets/b/f (skip, created since listed)\n", 1}, + } { + t.Run(tc.name, func(t *testing.T) { + stub := stubFilerSvc{lookupErr: filer_pb.ErrNotFound} + if tc.kept { + stub = stubFilerSvc{lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: listed}} + } + fake := &deleteRecordingFiler{ + takenNameFiler: takenNameFiler{stubFilerSvc: stub}, + listing: map[string][]*filer_pb.Entry{"/buckets/b": {listed}}, + } + remote := &listingRemote{dirs: map[string]map[string]bool{"/": tc.remote}} + var out strings.Builder + loc := &remote_pb.RemoteStorageLocation{Name: "r", Bucket: "rb", Path: "/"} + if err := pullMetadataDirectory(context.Background(), fake, &out, remote, util.FullPath("/buckets/b"), loc, tc.deleteStale, false); err != nil { + t.Fatalf("sync: %v", err) + } + if out.String() != tc.want || len(fake.creates) != tc.creates { + t.Fatalf("output %q with %d creates, want %q with %d", out.String(), len(fake.creates), tc.want, tc.creates) + } + }) + } +} diff --git a/weed/shell/command_remote_mount.go b/weed/shell/command_remote_mount.go index debfa27a63..2dcc661098 100644 --- a/weed/shell/command_remote_mount.go +++ b/weed/shell/command_remote_mount.go @@ -212,7 +212,11 @@ func ensureMountDirectory(commandEnv *CommandEnv, dir string, nonEmpty bool, rem // if an entry has synchronized metadata but has new updates // // entry.Attributes.Mtime * 1,000,000,000 > entry.RemoteEntry.LastLocalSyncTsNs -func doSaveRemoteEntry(client filer_pb.SeaweedFilerClient, localDir string, existingEntry *filer_pb.Entry, remoteEntry *filer_pb.RemoteEntry) error { +// +// The entry is written only while the filer still holds existingEntry; stale +// reports that it changed since it was read and nothing was written. +func doSaveRemoteEntry(client filer_pb.SeaweedFilerClient, localDir string, existingEntry *filer_pb.Entry, remoteEntry *filer_pb.RemoteEntry) (stale bool, err error) { + readEntry := proto.Clone(existingEntry).(*filer_pb.Entry) existingEntry.RemoteEntry = remoteEntry existingEntry.Attributes.FileSize = uint64(remoteEntry.RemoteSize) existingEntry.Attributes.Mtime = remoteEntry.RemoteMtime @@ -221,12 +225,5 @@ func doSaveRemoteEntry(client filer_pb.SeaweedFilerClient, localDir string, exis existingEntry.Extended = filer.MergeRemoteContentEncoding(remoteEntry, existingEntry.Extended) existingEntry.Chunks = nil existingEntry.Content = nil - _, updateErr := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{ - Directory: localDir, - Entry: existingEntry, - }) - if updateErr != nil { - return updateErr - } - return nil + return updateEntryIfUnchanged(context.Background(), client, localDir, existingEntry, readEntry) } diff --git a/weed/shell/command_remote_uncache.go b/weed/shell/command_remote_uncache.go index 20383a8f40..e3eb43f735 100644 --- a/weed/shell/command_remote_uncache.go +++ b/weed/shell/command_remote_uncache.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/proto" ) func init() { @@ -106,18 +107,18 @@ func (c *commandRemoteUncache) uncacheContentData(commandEnv *CommandEnv, writer return true // should not uncache an entry that is not synchronized with remote } - entry.RemoteEntry.LastLocalSyncTsNs = 0 - entry.Chunks = nil - fmt.Fprintf(writer, "Uncache %+v ... ", dir.Child(entry.Name)) + var stale bool err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - _, updateErr := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{ - Directory: string(dir), - Entry: entry, - }) + var updateErr error + stale, updateErr = uncacheEntry(client, string(dir), entry) return updateErr }) + if stale { + fmt.Fprintf(writer, "skipped, changed since it was read\n") + return true + } if err != nil { fmt.Fprintf(writer, "uncache %+v: %v\n", dir.Child(entry.Name), err) return false @@ -128,6 +129,15 @@ func (c *commandRemoteUncache) uncacheContentData(commandEnv *CommandEnv, writer }) } +// uncacheEntry drops the local copy of entry only while the filer still holds +// it as read, so a write that landed after the read keeps its data. +func uncacheEntry(client filer_pb.SeaweedFilerClient, dir string, entry *filer_pb.Entry) (stale bool, err error) { + readEntry := proto.Clone(entry).(*filer_pb.Entry) + entry.RemoteEntry.LastLocalSyncTsNs = 0 + entry.Chunks = nil + return updateEntryIfUnchanged(context.Background(), client, dir, entry, readEntry) +} + type FileFilter struct { include *string exclude *string diff --git a/weed/shell/command_remote_uncache_test.go b/weed/shell/command_remote_uncache_test.go index 5de572db31..62442ef042 100644 --- a/weed/shell/command_remote_uncache_test.go +++ b/weed/shell/command_remote_uncache_test.go @@ -5,6 +5,9 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) func TestFileFilter_matches_minCacheAge(t *testing.T) { @@ -102,3 +105,39 @@ func TestFileFilter_matches_minCacheAge(t *testing.T) { }) } } + +// The uncache clears the chunks of the entry its walk read. Conditioning the +// write on that copy keeps a file overwritten since the walk from being reset +// to the remote pointer with its new chunks dropped. +func TestUncacheEntryIsConditionedOnTheReadEntry(t *testing.T) { + read := &filer_pb.Entry{ + Name: "f", + Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 1, Cookie: 7}}}, + RemoteEntry: &filer_pb.RemoteEntry{StorageName: "r", LastLocalSyncTsNs: 5}, + } + for _, tc := range []struct { + name string + err error + stale bool + }{ + {"unchanged", nil, false}, + {"overwritten since the walk", status.Error(codes.FailedPrecondition, "precondition failed"), true}, + } { + t.Run(tc.name, func(t *testing.T) { + fake := &conditionalUpdateFiler{err: tc.err} + entry := proto.Clone(read).(*filer_pb.Entry) + stale, err := uncacheEntry(fake, "/buckets/b", entry) + if stale != tc.stale || (err != nil) != (tc.err != nil) { + t.Fatalf("stale=%v err=%v, want stale=%v err=%v", stale, err, tc.stale, tc.err) + } + clauses := fake.req.GetCondition().GetClauses() + if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL || !proto.Equal(clauses[0].ExpectedEntry, read) { + t.Fatalf("condition %v must expect the entry as read", fake.req.GetCondition()) + } + written := fake.req.Entry + if len(written.Chunks) != 0 || written.RemoteEntry.LastLocalSyncTsNs != 0 { + t.Fatalf("written entry %v must drop the local copy", written) + } + }) + } +}