filer: conditional UpdateEntry with a chunk-set write condition (#10382)

* filer: accept a WriteCondition on UpdateEntry, under the per-path lock

UpdateEntry was a bare read-modify-write: the precondition check, the
chunk garbage diff, and the store write could interleave with a
concurrent update to the same path. Take the per-path lock CreateEntry
already holds, and evaluate an optional CreateEntry-style WriteCondition
under it, failing with FailedPrecondition like expected_extended.

* filer: IF_CHUNKS_EQUAL write condition compares the stored chunk fid set

A chunk-preserving read-modify-write (tagging, setattr, copy-in-place)
races UpdateEntry's garbage diff: if a concurrent update empties the
chunk list first, the stale writer's commit resurrects fids that are
already queued for deletion, stranding the entry on a dead needle once
vacuum reclaims it. The reverse also holds: a writer that read an empty
chunk list can wipe chunks a concurrent update just added.

IF_CHUNKS_EQUAL guards both: the stored chunk fid multiset must still
equal what the caller read, order-independent, with an empty fids list
expecting no chunks. Absent entry counts as no chunks for CreateEntry
overwrites and transactions.

* filer: delete and append serialize on the entry path lock

DeleteEntry queues the entry's chunks for deletion and AppendToEntry
rewrites the chunk list, but neither held the per-path lock, so either
could interleave with a conditional update between its precondition
check and its write — a passed IF_CHUNKS_EQUAL would then resurrect
fids already on the deletion queue, or clobber a freshly appended
chunk. AppendToEntry keeps the cluster lock for cross-filer append
serialization; the path lock covers the local read-modify-write.

* filer: reuse lockPath in UpdateEntry lookup

weed/pb/filer_pb/filer_vtproto.pb.go

@@ -1105,6 +1105,15 @@ func (m *WriteCondition_Clause) MarshalToSizedBufferVT(dAtA []byte) (int, error)
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if len(m.Fids) > 0 {
+		for iNdEx := len(m.Fids) - 1; iNdEx >= 0; iNdEx-- {
+			i -= len(m.Fids[iNdEx])
+			copy(dAtA[i:], m.Fids[iNdEx])
+			i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.Fids[iNdEx])))
+			i--
+			dAtA[i] = 0x4a
+		}
+	}
 	if len(m.GateValue) > 0 {
 		i -= len(m.GateValue)
 		copy(dAtA[i:], m.GateValue)
@@ -2038,6 +2047,16 @@ func (m *UpdateEntryRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.Condition != nil {
+		size, err := m.Condition.MarshalToSizedBufferVT(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = protohelpers.EncodeVarint(dAtA, i, uint64(size))
+		i--
+		dAtA[i] = 0x32
+	}
 	if len(m.ExpectedExtended) > 0 {
 		for k := range m.ExpectedExtended {
 			v := m.ExpectedExtended[k]
@@ -6615,6 +6634,12 @@ func (m *WriteCondition_Clause) SizeVT() (n int) {
 	if l > 0 {
 		n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
 	}
+	if len(m.Fids) > 0 {
+		for _, s := range m.Fids {
+			l = len(s)
+			n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
+		}
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -6979,6 +7004,10 @@ func (m *UpdateEntryRequest) SizeVT() (n int) {
 			n += mapEntrySize + 1 + protohelpers.SizeOfVarint(uint64(mapEntrySize))
 		}
 	}
+	if m.Condition != nil {
+		l = m.Condition.SizeVT()
+		n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -11801,6 +11830,38 @@ func (m *WriteCondition_Clause) UnmarshalVT(dAtA []byte) error {
 			}
 			m.GateValue = string(dAtA[iNdEx:postIndex])
 			iNdEx = postIndex
+		case 9:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Fids", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return protohelpers.ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return protohelpers.ErrInvalidLength
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return protohelpers.ErrInvalidLength
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Fids = append(m.Fids, string(dAtA[iNdEx:postIndex]))
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := protohelpers.Skip(dAtA[iNdEx:])
@@ -14420,6 +14481,42 @@ func (m *UpdateEntryRequest) UnmarshalVT(dAtA []byte) error {
 			}
 			m.ExpectedExtended[mapkey] = mapvalue
 			iNdEx = postIndex
+		case 6:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Condition", 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.Condition == nil {
+				m.Condition = &WriteCondition{}
+			}
+			if err := m.Condition.UnmarshalVT(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := protohelpers.Skip(dAtA[iNdEx:])