diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index bdc6bc1ad..b8051e2cf 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -20,6 +20,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/replication/source" "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" ) @@ -41,14 +43,6 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour var lastLogTsNs = time.Now().UnixNano() processEventFnWithOffset := pb.AddOffsetFunc(func(resp *filer_pb.SubscribeMetadataResponse) error { - if resp.EventNotification.NewEntry != nil { - if *option.storageClass == "" { - delete(resp.EventNotification.NewEntry.Extended, s3_constants.AmzStorageClass) - } else { - resp.EventNotification.NewEntry.Extended[s3_constants.AmzStorageClass] = []byte(*option.storageClass) - } - } - processor.AddSyncJob(resp) return nil }, 3*time.Second, func(counter int64, lastTsNs int64) error { @@ -175,10 +169,10 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(parentPath, entryName), remoteStorageMountLocation) if message.NewEntry.IsDirectory { glog.V(0).Infof("mkdir %s", remote_storage.FormatLocation(dest)) - return client.WriteDirectory(dest, message.NewEntry) + return client.WriteDirectory(dest, remoteWriteEntry(message.NewEntry, *option.storageClass)) } glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest)) - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, remoteWriteEntry(message.NewEntry, *option.storageClass), dest) if errors.Is(writeErr, errSuperseded) { glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) return nil @@ -213,7 +207,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem return client.DeleteFile(dest) } if message.OldEntry != nil && message.NewEntry != nil { - return processUpdateEvent(option, filerSource, client, mountedDir, remoteStorageMountLocation, resp) + return processUpdateEvent(option, filerSource, *option.storageClass, client, mountedDir, remoteStorageMountLocation, resp) } return nil @@ -224,6 +218,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem func processUpdateEvent( filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, + storageClass string, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, @@ -244,7 +239,7 @@ func processUpdateEvent( return nil } if message.NewEntry.IsDirectory { - return client.WriteDirectory(dest, message.NewEntry) + return client.WriteDirectory(dest, remoteWriteEntry(message.NewEntry, storageClass)) } if isMetadataOnlyUpdate(resp.Directory, message) { remoteEntry, err := liveRemoteEntry(filerClient, message.NewParentPath, message.NewEntry) @@ -257,7 +252,7 @@ func processUpdateEvent( } if remoteEntry != nil { glog.V(2).Infof("update meta: %+v", resp) - return client.UpdateFileMetadata(dest, message.OldEntry, message.NewEntry) + return client.UpdateFileMetadata(dest, message.OldEntry, remoteWriteEntry(message.NewEntry, storageClass)) } glog.V(0).Infof("never replicated, uploading %s", remote_storage.FormatLocation(dest)) } @@ -277,7 +272,7 @@ func processUpdateEvent( } } } - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, remoteWriteEntry(message.NewEntry, storageClass), dest) if errors.Is(writeErr, errSuperseded) { glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) return nil @@ -417,16 +412,66 @@ func shouldSendToRemote(entry *filer_pb.Entry) bool { return false } +// remoteWriteEntry returns the entry as remote storage should see it: the +// storage class attribute is dropped, or overridden by -storageClass. The +// event entry is left untouched so updateLocalEntry still compares the entry +// the filer stored. +func remoteWriteEntry(entry *filer_pb.Entry, storageClass string) *filer_pb.Entry { + clone := proto.Clone(entry).(*filer_pb.Entry) + if storageClass == "" { + delete(clone.Extended, s3_constants.AmzStorageClass) + } else { + if clone.Extended == nil { + clone.Extended = map[string][]byte{} + } + clone.Extended[s3_constants.AmzStorageClass] = []byte(storageClass) + } + return clone +} + +// updateLocalEntry stamps the entry an event described with its RemoteEntry. +// The write carries IF_ENTRY_EQUAL over the event's entry: the filer deletes +// every stored chunk absent from an updated entry, so a snapshot older than +// the live entry (the file was rewritten while its upload was in flight, or +// the event is a replay) would delete the live chunks. A failed precondition +// means the filer moved past this event; the event that superseded it follows +// in the log and stamps the current entry, so the stale stamp is skipped the +// same way a superseded upload is. func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry, remoteEntry *filer_pb.RemoteEntry) error { remoteEntry.LastLocalSyncTsNs = time.Now().UnixNano() + expected := proto.Clone(entry).(*filer_pb.Entry) entry.RemoteEntry = remoteEntry - return filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{ Directory: dir, Entry: entry, + Condition: ifEntryEqual(expected), }) return err }) + if isFailedPrecondition(err) { + glog.Errorf("skipping stale stamp of %s: %v", util.NewFullPath(dir, entry.Name), err) + return nil + } + return err +} + +// ifEntryEqual builds the precondition that the stored entry still equals the +// one the event described: chunk fids, inline content, and metadata alike. +func ifEntryEqual(entry *filer_pb.Entry) *filer_pb.WriteCondition { + return &filer_pb.WriteCondition{ + Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: entry}}, + } +} + +// isFailedPrecondition reports a write condition the filer refused, through +// any wrapping WithFilerClient added. +func isFailedPrecondition(err error) bool { + if err == nil { + return false + } + st, ok := status.FromError(err) + return ok && st.Code() == codes.FailedPrecondition } func isMultipartUploadFile(dir string, name string) bool { diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index b2f682cc2..7141c2a2c 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -16,6 +16,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "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" ) @@ -684,7 +686,7 @@ func TestRenameWithInheritedRemoteEntryWritesNewKey(t *testing.T) { remote := &recordingRemote{} filerClient := &stubFilerClient{} - if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil { + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil { t.Fatal(err) } @@ -731,7 +733,7 @@ func TestRenameRemoteOnlyEntrySkipsEmptyUpload(t *testing.T) { remote := &recordingRemote{} filerClient := &stubFilerClient{} - if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil { + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil { t.Fatal(err) } if len(remote.writes) != 0 { @@ -764,7 +766,7 @@ func TestRenameDeleteOldKeyFailureReturnsError(t *testing.T) { deleteErr := errors.New("AccessDenied: Access Denied") remote := &recordingRemote{deleteErr: deleteErr} filerClient := &stubFilerClient{} - err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp) + err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp) if !errors.Is(err, deleteErr) { t.Errorf("err = %v, want the delete failure returned so the event is retried", err) } @@ -800,7 +802,7 @@ func TestRenameDeleteOldKeyNotFoundStillWrites(t *testing.T) { remote := &recordingRemote{deleteErr: remote_storage.ErrRemoteObjectNotFound} filerClient := &stubFilerClient{} - if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil { + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil { t.Fatalf("err = %v, want nil: an already-deleted old key must not block the write", err) } wantWrite := &remote_pb.RemoteStorageLocation{Name: "gcs", Bucket: "bucket", Path: "/b/dst/probe.bin"} @@ -808,3 +810,73 @@ func TestRenameDeleteOldKeyNotFoundStillWrites(t *testing.T) { t.Errorf("writes = %+v, want the new key %s written", remote.writes, remote_storage.FormatLocation(wantWrite)) } } + +func TestIfEntryEqualCarriesTheEventEntry(t *testing.T) { + entry := &filer_pb.Entry{ + Name: "f", + Content: []byte("inline"), + Chunks: []*filer_pb.FileChunk{ + {FileId: "3,01a"}, + {Fid: &filer_pb.FileId{VolumeId: 4, FileKey: 0x2b, Cookie: 0x0c}}, + }, + } + cond := ifEntryEqual(entry) + if len(cond.Clauses) != 1 || cond.Clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL { + t.Fatalf("condition = %v, want one IF_ENTRY_EQUAL clause", cond) + } + if got := cond.Clauses[0].ExpectedEntry; !proto.Equal(got, entry) { + t.Fatalf("expected_entry = %v, want %v", got, entry) + } +} + +// The remote-bound entry loses (or gains) the storage class attribute while +// the event entry keeps it, so IF_ENTRY_EQUAL still sees the entry the filer +// stored — otherwise every stamp on an S3-written object reads as stale. +func TestRemoteWriteEntryLeavesEventEntryUntouched(t *testing.T) { + entry := &filer_pb.Entry{ + Name: "f", + Attributes: &filer_pb.FuseAttributes{Mtime: 1}, + Extended: map[string][]byte{s3_constants.AmzStorageClass: []byte("STANDARD")}, + } + + stripped := remoteWriteEntry(entry, "") + if _, ok := stripped.Extended[s3_constants.AmzStorageClass]; ok { + t.Fatalf("remote entry still carries %s", s3_constants.AmzStorageClass) + } + if string(entry.Extended[s3_constants.AmzStorageClass]) != "STANDARD" { + t.Fatalf("event entry Extended mutated: %v", entry.Extended) + } + + overridden := remoteWriteEntry(entry, "GLACIER") + if got := string(overridden.Extended[s3_constants.AmzStorageClass]); got != "GLACIER" { + t.Fatalf("override = %q, want GLACIER", got) + } + if string(entry.Extended[s3_constants.AmzStorageClass]) != "STANDARD" { + t.Fatalf("event entry Extended mutated: %v", entry.Extended) + } + + bare := &filer_pb.Entry{Name: "g", Attributes: &filer_pb.FuseAttributes{Mtime: 1}} + if got := string(remoteWriteEntry(bare, "GLACIER").Extended[s3_constants.AmzStorageClass]); got != "GLACIER" { + t.Fatalf("override on nil Extended = %q, want GLACIER", got) + } +} + +func TestIsFailedPrecondition(t *testing.T) { + refused := status.Error(codes.FailedPrecondition, "precondition failed: /buckets/b/f") + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"status", refused, true}, + {"wrapped status", fmt.Errorf("update entry: %w", refused), true}, + {"message only", errors.New("rpc error: code = FailedPrecondition desc = precondition failed: /f"), false}, + {"other", status.Error(codes.Unavailable, "filer down"), false}, + } + for _, c := range cases { + if got := isFailedPrecondition(c.err); got != c.want { + t.Errorf("%s: isFailedPrecondition = %v, want %v", c.name, got, c.want) + } + } +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 1515c169a..50de94dfc 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -281,6 +281,7 @@ message WriteCondition { IF_EXTENDED_NOT_EQUAL = 7; // fail if present and extended[ext_key] == ext_value IF_EXTENDED_TIME_ELAPSED = 8; // fail if present and extended[ext_key] (unix seconds) is in the future IF_CHUNKS_EQUAL = 9; // fail unless the stored chunk fid multiset equals fids (absent entry = no chunks) + IF_ENTRY_EQUAL = 10; // fail unless the stored entry equals expected_entry } // Clause is one primitive comparison. IF_ETAG_MATCH holds when the current // entry's ETag equals any value in etags; IF_ETAG_NOT_MATCH holds when it @@ -309,6 +310,7 @@ message WriteCondition { string gate_key = 7; // IF_EXTENDED_TIME_ELAPSED: only enforce when extended[gate_key] == gate_value string gate_value = 8; // gate value (e.g. retention mode COMPLIANCE for governance bypass) repeated string fids = 9; // chunk fid strings for IF_CHUNKS_EQUAL + Entry expected_entry = 10; // whole-entry comparison for IF_ENTRY_EQUAL } repeated Clause clauses = 1; // all must hold (logical AND) } @@ -477,6 +479,7 @@ message UpdateEntryRequest { // the write, under the filer's per-path lock. The caller must route the // key's writes to this entry's owner filer for the check to be authoritative. WriteCondition condition = 6; + bool is_moved = 7; // set on a forwarded request so the owner applies it locally } message UpdateEntryResponse { SubscribeMetadataResponse metadata_event = 1; diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index ab71038df..f9bd74c4c 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v7.35.0 +// protoc v3.21.12 // source: filer.proto package filer_pb @@ -197,31 +197,33 @@ func (PosixLockOp) EnumDescriptor() ([]byte, []int) { type WriteCondition_Kind int32 const ( - WriteCondition_NONE WriteCondition_Kind = 0 // unconditional - WriteCondition_IF_NOT_EXISTS WriteCondition_Kind = 1 // fail if the entry exists (If-None-Match: *) - WriteCondition_IF_EXISTS WriteCondition_Kind = 2 // fail if the entry is absent (If-Match: *) - WriteCondition_IF_ETAG_MATCH WriteCondition_Kind = 3 // fail if absent or etag matches none of the set (If-Match) - WriteCondition_IF_ETAG_NOT_MATCH WriteCondition_Kind = 4 // fail if present and etag matches any of the set (If-None-Match) - WriteCondition_IF_UNMODIFIED_SINCE WriteCondition_Kind = 5 // fail if present and mtime > unix_time - WriteCondition_IF_MODIFIED_SINCE WriteCondition_Kind = 6 // fail if present and mtime <= unix_time - WriteCondition_IF_EXTENDED_NOT_EQUAL WriteCondition_Kind = 7 // fail if present and extended[ext_key] == ext_value - WriteCondition_IF_EXTENDED_TIME_ELAPSED WriteCondition_Kind = 8 // fail if present and extended[ext_key] (unix seconds) is in the future - WriteCondition_IF_CHUNKS_EQUAL WriteCondition_Kind = 9 // fail unless the stored chunk fid multiset equals fids (absent entry = no chunks) + WriteCondition_NONE WriteCondition_Kind = 0 // unconditional + WriteCondition_IF_NOT_EXISTS WriteCondition_Kind = 1 // fail if the entry exists (If-None-Match: *) + WriteCondition_IF_EXISTS WriteCondition_Kind = 2 // fail if the entry is absent (If-Match: *) + WriteCondition_IF_ETAG_MATCH WriteCondition_Kind = 3 // fail if absent or etag matches none of the set (If-Match) + WriteCondition_IF_ETAG_NOT_MATCH WriteCondition_Kind = 4 // fail if present and etag matches any of the set (If-None-Match) + WriteCondition_IF_UNMODIFIED_SINCE WriteCondition_Kind = 5 // fail if present and mtime > unix_time + WriteCondition_IF_MODIFIED_SINCE WriteCondition_Kind = 6 // fail if present and mtime <= unix_time + WriteCondition_IF_EXTENDED_NOT_EQUAL WriteCondition_Kind = 7 // fail if present and extended[ext_key] == ext_value + WriteCondition_IF_EXTENDED_TIME_ELAPSED WriteCondition_Kind = 8 // fail if present and extended[ext_key] (unix seconds) is in the future + WriteCondition_IF_CHUNKS_EQUAL WriteCondition_Kind = 9 // fail unless the stored chunk fid multiset equals fids (absent entry = no chunks) + WriteCondition_IF_ENTRY_EQUAL WriteCondition_Kind = 10 // fail unless the stored entry equals expected_entry ) // Enum value maps for WriteCondition_Kind. var ( WriteCondition_Kind_name = map[int32]string{ - 0: "NONE", - 1: "IF_NOT_EXISTS", - 2: "IF_EXISTS", - 3: "IF_ETAG_MATCH", - 4: "IF_ETAG_NOT_MATCH", - 5: "IF_UNMODIFIED_SINCE", - 6: "IF_MODIFIED_SINCE", - 7: "IF_EXTENDED_NOT_EQUAL", - 8: "IF_EXTENDED_TIME_ELAPSED", - 9: "IF_CHUNKS_EQUAL", + 0: "NONE", + 1: "IF_NOT_EXISTS", + 2: "IF_EXISTS", + 3: "IF_ETAG_MATCH", + 4: "IF_ETAG_NOT_MATCH", + 5: "IF_UNMODIFIED_SINCE", + 6: "IF_MODIFIED_SINCE", + 7: "IF_EXTENDED_NOT_EQUAL", + 8: "IF_EXTENDED_TIME_ELAPSED", + 9: "IF_CHUNKS_EQUAL", + 10: "IF_ENTRY_EQUAL", } WriteCondition_Kind_value = map[string]int32{ "NONE": 0, @@ -234,6 +236,7 @@ var ( "IF_EXTENDED_NOT_EQUAL": 7, "IF_EXTENDED_TIME_ELAPSED": 8, "IF_CHUNKS_EQUAL": 9, + "IF_ENTRY_EQUAL": 10, } ) @@ -2375,6 +2378,7 @@ type UpdateEntryRequest struct { // the write, under the filer's per-path lock. The caller must route the // key's writes to this entry's owner filer for the check to be authoritative. Condition *WriteCondition `protobuf:"bytes,6,opt,name=condition,proto3" json:"condition,omitempty"` + IsMoved bool `protobuf:"varint,7,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"` // set on a forwarded request so the owner applies it locally unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -2451,6 +2455,13 @@ func (x *UpdateEntryRequest) GetCondition() *WriteCondition { return nil } +func (x *UpdateEntryRequest) GetIsMoved() bool { + if x != nil { + return x.IsMoved + } + return false +} + type UpdateEntryResponse struct { state protoimpl.MessageState `protogen:"open.v1"` MetadataEvent *SubscribeMetadataResponse `protobuf:"bytes,1,opt,name=metadata_event,json=metadataEvent,proto3" json:"metadata_event,omitempty"` @@ -6856,14 +6867,15 @@ func (x *LookupDirectoryEntriesResponse) GetReadAuth() map[string]string { type WriteCondition_Clause struct { state protoimpl.MessageState `protogen:"open.v1"` Kind WriteCondition_Kind `protobuf:"varint,1,opt,name=kind,proto3,enum=filer_pb.WriteCondition_Kind" json:"kind,omitempty"` - Etags []string `protobuf:"bytes,2,rep,name=etags,proto3" json:"etags,omitempty"` // ETag set for IF_ETAG_* kinds - UnixTime int64 `protobuf:"varint,3,opt,name=unix_time,json=unixTime,proto3" json:"unix_time,omitempty"` // bound (unix seconds) for IF_*_SINCE kinds - AllowWeak bool `protobuf:"varint,4,opt,name=allow_weak,json=allowWeak,proto3" json:"allow_weak,omitempty"` // compare ETags ignoring the weak (W/) marker - ExtKey string `protobuf:"bytes,5,opt,name=ext_key,json=extKey,proto3" json:"ext_key,omitempty"` // extended attribute name for IF_EXTENDED_* kinds - ExtValue string `protobuf:"bytes,6,opt,name=ext_value,json=extValue,proto3" json:"ext_value,omitempty"` // blocking value for IF_EXTENDED_NOT_EQUAL - GateKey string `protobuf:"bytes,7,opt,name=gate_key,json=gateKey,proto3" json:"gate_key,omitempty"` // IF_EXTENDED_TIME_ELAPSED: only enforce when extended[gate_key] == gate_value - GateValue string `protobuf:"bytes,8,opt,name=gate_value,json=gateValue,proto3" json:"gate_value,omitempty"` // gate value (e.g. retention mode COMPLIANCE for governance bypass) - Fids []string `protobuf:"bytes,9,rep,name=fids,proto3" json:"fids,omitempty"` // chunk fid strings for IF_CHUNKS_EQUAL + Etags []string `protobuf:"bytes,2,rep,name=etags,proto3" json:"etags,omitempty"` // ETag set for IF_ETAG_* kinds + UnixTime int64 `protobuf:"varint,3,opt,name=unix_time,json=unixTime,proto3" json:"unix_time,omitempty"` // bound (unix seconds) for IF_*_SINCE kinds + AllowWeak bool `protobuf:"varint,4,opt,name=allow_weak,json=allowWeak,proto3" json:"allow_weak,omitempty"` // compare ETags ignoring the weak (W/) marker + ExtKey string `protobuf:"bytes,5,opt,name=ext_key,json=extKey,proto3" json:"ext_key,omitempty"` // extended attribute name for IF_EXTENDED_* kinds + ExtValue string `protobuf:"bytes,6,opt,name=ext_value,json=extValue,proto3" json:"ext_value,omitempty"` // blocking value for IF_EXTENDED_NOT_EQUAL + GateKey string `protobuf:"bytes,7,opt,name=gate_key,json=gateKey,proto3" json:"gate_key,omitempty"` // IF_EXTENDED_TIME_ELAPSED: only enforce when extended[gate_key] == gate_value + GateValue string `protobuf:"bytes,8,opt,name=gate_value,json=gateValue,proto3" json:"gate_value,omitempty"` // gate value (e.g. retention mode COMPLIANCE for governance bypass) + Fids []string `protobuf:"bytes,9,rep,name=fids,proto3" json:"fids,omitempty"` // chunk fid strings for IF_CHUNKS_EQUAL + ExpectedEntry *Entry `protobuf:"bytes,10,opt,name=expected_entry,json=expectedEntry,proto3" json:"expected_entry,omitempty"` // whole-entry comparison for IF_ENTRY_EQUAL unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -6961,6 +6973,13 @@ func (x *WriteCondition_Clause) GetFids() []string { return nil } +func (x *WriteCondition_Clause) GetExpectedEntry() *Entry { + if x != nil { + return x.ExpectedEntry + } + return nil +} + // if found, send the exact address // if not found, send the full list of existing brokers type LocateBrokerResponse_Resource struct { @@ -7301,9 +7320,9 @@ const file_filer_proto_rawDesc = "" + "signatures\x12=\n" + "\x1bskip_check_parent_directory\x18\x06 \x01(\bR\x18skipCheckParentDirectory\x126\n" + "\tcondition\x18\a \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x12\x19\n" + - "\bis_moved\x18\b \x01(\bR\aisMoved\"\xbc\x04\n" + + "\bis_moved\x18\b \x01(\bR\aisMoved\"\x88\x05\n" + "\x0eWriteCondition\x129\n" + - "\aclauses\x18\x01 \x03(\v2\x1f.filer_pb.WriteCondition.ClauseR\aclauses\x1a\x91\x02\n" + + "\aclauses\x18\x01 \x03(\v2\x1f.filer_pb.WriteCondition.ClauseR\aclauses\x1a\xc9\x02\n" + "\x06Clause\x121\n" + "\x04kind\x18\x01 \x01(\x0e2\x1d.filer_pb.WriteCondition.KindR\x04kind\x12\x14\n" + "\x05etags\x18\x02 \x03(\tR\x05etags\x12\x1b\n" + @@ -7315,7 +7334,9 @@ const file_filer_proto_rawDesc = "" + "\bgate_key\x18\a \x01(\tR\agateKey\x12\x1d\n" + "\n" + "gate_value\x18\b \x01(\tR\tgateValue\x12\x12\n" + - "\x04fids\x18\t \x03(\tR\x04fids\"\xda\x01\n" + + "\x04fids\x18\t \x03(\tR\x04fids\x126\n" + + "\x0eexpected_entry\x18\n" + + " \x01(\v2\x0f.filer_pb.EntryR\rexpectedEntry\"\xee\x01\n" + "\x04Kind\x12\b\n" + "\x04NONE\x10\x00\x12\x11\n" + "\rIF_NOT_EXISTS\x10\x01\x12\r\n" + @@ -7326,7 +7347,9 @@ const file_filer_proto_rawDesc = "" + "\x11IF_MODIFIED_SINCE\x10\x06\x12\x19\n" + "\x15IF_EXTENDED_NOT_EQUAL\x10\a\x12\x1c\n" + "\x18IF_EXTENDED_TIME_ELAPSED\x10\b\x12\x13\n" + - "\x0fIF_CHUNKS_EQUAL\x10\t\"\xa2\x05\n" + + "\x0fIF_CHUNKS_EQUAL\x10\t\x12\x12\n" + + "\x0eIF_ENTRY_EQUAL\x10\n" + + "\"\xa2\x05\n" + "\x0eObjectMutation\x121\n" + "\x04type\x18\x01 \x01(\x0e2\x1d.filer_pb.ObjectMutation.TypeR\x04type\x12\x1c\n" + "\tdirectory\x18\x02 \x01(\tR\tdirectory\x12\x12\n" + @@ -7414,7 +7437,7 @@ const file_filer_proto_rawDesc = "" + "\n" + "error_code\x18\x03 \x01(\x0e2\x14.filer_pb.FilerErrorR\terrorCode\x12\x1a\n" + "\tlog_ts_ns\x18\x04 \x01(\x03R\alogTsNs\x12#\n" + - "\rlog_signature\x18\x05 \x01(\x05R\flogSignature\"\x8a\x03\n" + + "\rlog_signature\x18\x05 \x01(\x05R\flogSignature\"\xa5\x03\n" + "\x12UpdateEntryRequest\x12\x1c\n" + "\tdirectory\x18\x01 \x01(\tR\tdirectory\x12%\n" + "\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x121\n" + @@ -7423,7 +7446,8 @@ const file_filer_proto_rawDesc = "" + "signatures\x18\x04 \x03(\x05R\n" + "signatures\x12_\n" + "\x11expected_extended\x18\x05 \x03(\v22.filer_pb.UpdateEntryRequest.ExpectedExtendedEntryR\x10expectedExtended\x126\n" + - "\tcondition\x18\x06 \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x1aC\n" + + "\tcondition\x18\x06 \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x12\x19\n" + + "\bis_moved\x18\a \x01(\bR\aisMoved\x1aC\n" + "\x15ExpectedExtendedEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\fR\x05value:\x028\x01\"\xa2\x01\n" + @@ -8072,83 +8096,84 @@ var file_filer_proto_depIdxs = []int32{ 107, // 67: filer_pb.LookupDirectoryEntriesResponse.locations_map:type_name -> filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry 108, // 68: filer_pb.LookupDirectoryEntriesResponse.read_auth:type_name -> filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntry 3, // 69: filer_pb.WriteCondition.Clause.kind:type_name -> filer_pb.WriteCondition.Kind - 44, // 70: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations - 44, // 71: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations - 5, // 72: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest - 96, // 73: filer_pb.SeaweedFiler.LookupDirectoryEntries:input_type -> filer_pb.LookupDirectoryEntriesRequest - 7, // 74: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest - 17, // 75: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest - 29, // 76: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest - 31, // 77: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest - 33, // 78: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest - 35, // 79: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest - 21, // 80: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest - 26, // 81: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest - 24, // 82: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest - 37, // 83: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest - 39, // 84: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest - 89, // 85: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest - 41, // 86: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest - 43, // 87: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest - 48, // 88: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest - 50, // 89: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest - 52, // 90: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest - 54, // 91: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest - 56, // 92: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest - 64, // 93: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest - 58, // 94: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest - 58, // 95: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest - 60, // 96: filer_pb.SeaweedFiler.ListMetadataSubscribers:input_type -> filer_pb.ListMetadataSubscribersRequest - 71, // 97: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest - 73, // 98: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest - 76, // 99: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest - 78, // 100: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest - 80, // 101: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest - 82, // 102: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest - 85, // 103: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest - 87, // 104: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest - 91, // 105: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest - 93, // 106: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest - 6, // 107: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse - 98, // 108: filer_pb.SeaweedFiler.LookupDirectoryEntries:output_type -> filer_pb.LookupDirectoryEntriesResponse - 8, // 109: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse - 28, // 110: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse - 30, // 111: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse - 32, // 112: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse - 34, // 113: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse - 36, // 114: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse - 22, // 115: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse - 27, // 116: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse - 25, // 117: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse - 38, // 118: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse - 40, // 119: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse - 90, // 120: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse - 42, // 121: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse - 46, // 122: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse - 49, // 123: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse - 51, // 124: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse - 53, // 125: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse - 55, // 126: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse - 57, // 127: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse - 65, // 128: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse - 59, // 129: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse - 59, // 130: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse - 61, // 131: filer_pb.SeaweedFiler.ListMetadataSubscribers:output_type -> filer_pb.ListMetadataSubscribersResponse - 72, // 132: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse - 74, // 133: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse - 77, // 134: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse - 79, // 135: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse - 81, // 136: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse - 83, // 137: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse - 86, // 138: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse - 88, // 139: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse - 92, // 140: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse - 94, // 141: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse - 107, // [107:142] is the sub-list for method output_type - 72, // [72:107] is the sub-list for method input_type - 72, // [72:72] is the sub-list for extension type_name - 72, // [72:72] is the sub-list for extension extendee - 0, // [0:72] is the sub-list for field type_name + 10, // 70: filer_pb.WriteCondition.Clause.expected_entry:type_name -> filer_pb.Entry + 44, // 71: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations + 44, // 72: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations + 5, // 73: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest + 96, // 74: filer_pb.SeaweedFiler.LookupDirectoryEntries:input_type -> filer_pb.LookupDirectoryEntriesRequest + 7, // 75: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest + 17, // 76: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest + 29, // 77: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest + 31, // 78: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest + 33, // 79: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest + 35, // 80: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest + 21, // 81: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest + 26, // 82: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest + 24, // 83: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest + 37, // 84: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest + 39, // 85: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest + 89, // 86: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest + 41, // 87: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest + 43, // 88: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest + 48, // 89: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest + 50, // 90: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest + 52, // 91: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest + 54, // 92: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest + 56, // 93: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest + 64, // 94: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest + 58, // 95: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest + 58, // 96: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest + 60, // 97: filer_pb.SeaweedFiler.ListMetadataSubscribers:input_type -> filer_pb.ListMetadataSubscribersRequest + 71, // 98: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest + 73, // 99: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest + 76, // 100: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest + 78, // 101: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest + 80, // 102: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest + 82, // 103: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest + 85, // 104: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest + 87, // 105: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest + 91, // 106: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest + 93, // 107: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest + 6, // 108: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse + 98, // 109: filer_pb.SeaweedFiler.LookupDirectoryEntries:output_type -> filer_pb.LookupDirectoryEntriesResponse + 8, // 110: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse + 28, // 111: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse + 30, // 112: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse + 32, // 113: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse + 34, // 114: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse + 36, // 115: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse + 22, // 116: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse + 27, // 117: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse + 25, // 118: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse + 38, // 119: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse + 40, // 120: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse + 90, // 121: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse + 42, // 122: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse + 46, // 123: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse + 49, // 124: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse + 51, // 125: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse + 53, // 126: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse + 55, // 127: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse + 57, // 128: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse + 65, // 129: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse + 59, // 130: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse + 59, // 131: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse + 61, // 132: filer_pb.SeaweedFiler.ListMetadataSubscribers:output_type -> filer_pb.ListMetadataSubscribersResponse + 72, // 133: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse + 74, // 134: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse + 77, // 135: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse + 79, // 136: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse + 81, // 137: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse + 83, // 138: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse + 86, // 139: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse + 88, // 140: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse + 92, // 141: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse + 94, // 142: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse + 108, // [108:143] is the sub-list for method output_type + 73, // [73:108] is the sub-list for method input_type + 73, // [73:73] is the sub-list for extension type_name + 73, // [73:73] is the sub-list for extension extendee + 0, // [0:73] is the sub-list for field type_name } func init() { file_filer_proto_init() } diff --git a/weed/pb/filer_pb/filer_grpc.pb.go b/weed/pb/filer_pb/filer_grpc.pb.go index 6a2b494a3..8a330ce28 100644 --- a/weed/pb/filer_pb/filer_grpc.pb.go +++ b/weed/pb/filer_pb/filer_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.2 -// - protoc v7.35.0 +// - protoc v3.21.12 // source: filer.proto package filer_pb diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 17e958767..249f1d14c 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -1135,6 +1135,16 @@ func (m *WriteCondition_Clause) MarshalToSizedBufferVT(dAtA []byte) (int, error) i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.ExpectedEntry != nil { + size, err := m.ExpectedEntry.MarshalToSizedBufferVT(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = protohelpers.EncodeVarint(dAtA, i, uint64(size)) + i-- + dAtA[i] = 0x52 + } if len(m.Fids) > 0 { for iNdEx := len(m.Fids) - 1; iNdEx >= 0; iNdEx-- { i -= len(m.Fids[iNdEx]) @@ -2087,6 +2097,16 @@ func (m *UpdateEntryRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.IsMoved { + i-- + if m.IsMoved { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x38 + } if m.Condition != nil { size, err := m.Condition.MarshalToSizedBufferVT(dAtA[:i]) if err != nil { @@ -6958,6 +6978,10 @@ func (m *WriteCondition_Clause) SizeVT() (n int) { n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } } + if m.ExpectedEntry != nil { + l = m.ExpectedEntry.SizeVT() + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } n += len(m.unknownFields) return n } @@ -7332,6 +7356,9 @@ func (m *UpdateEntryRequest) SizeVT() (n int) { l = m.Condition.SizeVT() n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } + if m.IsMoved { + n += 2 + } n += len(m.unknownFields) return n } @@ -12374,6 +12401,42 @@ func (m *WriteCondition_Clause) UnmarshalVT(dAtA []byte) error { } m.Fids = append(m.Fids, string(dAtA[iNdEx:postIndex])) iNdEx = postIndex + case 10: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ExpectedEntry", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.ExpectedEntry == nil { + m.ExpectedEntry = &Entry{} + } + if err := m.ExpectedEntry.UnmarshalVT(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) @@ -15067,6 +15130,26 @@ func (m *UpdateEntryRequest) UnmarshalVT(dAtA []byte) error { return err } iNdEx = postIndex + case 7: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field IsMoved", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.IsMoved = bool(v != 0) default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index b0cf88345..e99c9abe2 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -662,6 +662,32 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr fullpath := util.Join(req.Directory, req.Entry.Name) + // A conditional or preconditioned update is a read-then-write that the + // per-path lock below only makes atomic on this filer. Route it to the + // entry's ring owner so one filer's lock arbitrates every writer + // cluster-wide; is_moved bounds this to one hop. + if !req.IsMoved && (conditionIsSet(req.Condition) || len(req.ExpectedExtended) > 0) { + var ownerResp *filer_pb.UpdateEntryResponse + handled, forwardErr := fs.forwardToWriteOwner(ctx, entryRouteKey(util.FullPath(fullpath)), func(owner pb.ServerAddress) error { + glog.V(2).InfofCtx(ctx, "UpdateEntry %s: forwarding to owner %s", fullpath, owner) + req.IsMoved = true + return pb.WithFilerClient(false, 0, owner, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + forwarded, e := client.UpdateEntry(ctx, req) + if e != nil { + return e + } + ownerResp = forwarded + return nil + }) + }) + if handled { + if forwardErr != nil { + return &filer_pb.UpdateEntryResponse{}, forwardErr + } + return ownerResp, nil + } + } + // Serialize concurrent mutations to the same path on this filer so the // read (preconditions, garbage diff) and the write are atomic. Callers // route a key's writes to this owner filer, making this local lock diff --git a/weed/server/filer_grpc_server_condition.go b/weed/server/filer_grpc_server_condition.go index 82406c367..f51b0dcef 100644 --- a/weed/server/filer_grpc_server_condition.go +++ b/weed/server/filer_grpc_server_condition.go @@ -8,6 +8,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "google.golang.org/protobuf/proto" ) // conditionIsSet reports whether a condition asks for any check at all. @@ -82,6 +83,14 @@ func clauseSatisfied(c *filer_pb.WriteCondition_Clause, current *filer.Entry) bo return deadline <= time.Now().Unix() case filer_pb.WriteCondition_IF_CHUNKS_EQUAL: return chunkFidsEqual(current, c.Fids) + case filer_pb.WriteCondition_IF_ENTRY_EQUAL: + if !exists || c.ExpectedEntry == nil { + return !exists && c.ExpectedEntry == nil + } + // Normalize the expected entry the way FindEntry normalizes the stored + // one (e.g. FileSize grows to the chunk extent), or an unchanged entry + // can compare unequal. + return proto.Equal(current.ToProtoEntry(), filer.FromPbEntry("", c.ExpectedEntry).ToProtoEntry()) default: // An unrecognized clause kind (e.g. from a newer client) must not be // treated as satisfied, which would silently bypass the guard. Fail diff --git a/weed/server/filer_grpc_server_condition_test.go b/weed/server/filer_grpc_server_condition_test.go index 66906fafb..2946e9614 100644 --- a/weed/server/filer_grpc_server_condition_test.go +++ b/weed/server/filer_grpc_server_condition_test.go @@ -193,6 +193,34 @@ func TestWriteConditionUnknownKindFailsClosed(t *testing.T) { } } +// IF_ENTRY_EQUAL compares the expected entry after the same normalization +// FindEntry applied to the stored one: a raw event entry whose FileSize is +// still zero must match a stored entry grown to its chunk extent. +func TestIfEntryEqualNormalizesExpected(t *testing.T) { + raw := &filer_pb.Entry{ + Name: "f", + Attributes: &filer_pb.FuseAttributes{Mtime: 42}, + Chunks: []*filer_pb.FileChunk{ + {Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 1, Cookie: 2}, Size: 100}, + }, + } + stored := filer.FromPbEntry("/d", raw) + if stored.FileSize != 100 { + t.Fatalf("stored FileSize = %d, want chunk extent 100", stored.FileSize) + } + cond := one(&filer_pb.WriteCondition_Clause{ + Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, + ExpectedEntry: raw, + }) + if !writeConditionSatisfied(cond, stored) { + t.Error("raw expected entry must equal its normalized stored form") + } + raw.Attributes.Mtime = 43 + if writeConditionSatisfied(cond, stored) { + t.Error("changed expected entry must not equal the stored entry") + } +} + // storedEntryETag prefers the stored Seaweed ETag attribute and falls back to // the Md5-derived ETag, matching the S3 gateway. func TestStoredEntryETag(t *testing.T) {