filer: route conditional UpdateEntry to the entry's owner filer

Two filers locking the same path locally could still pass a stale
condition on the non-owner while the owner's entry had moved on. When a
condition or expected_extended precondition is set, forward the request
to the entry's owner the same way conditional CreateEntry does, with
is_moved bounding the hop.

weed/pb/filer.proto

@@ -479,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;

weed/pb/filer_pb/filer.pb.go

@@ -2378,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
 }
@@ -2454,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"`
@@ -7429,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" +
@@ -7438,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" +

weed/pb/filer_pb/filer_vtproto.pb.go

@@ -2097,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 {
@@ -7346,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
 }
@@ -15117,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:])

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