From 9815bb4e59ac9d6c7d7d7ff13e72a162882021ef Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 20 Oct 2026 05:36:42 -0700 Subject: [PATCH] master: retry unenforced quarantine marks and restore server writability on release (#11763) Dial the advertised grpc port, return mark errors, and keep a quarantine entry until the writable push succeeds so a released copy does not stay read-only on the server. --- weed/server/master_server.go | 15 ++++++++------- weed/topology/topology.go | 26 ++++++++++++++++++-------- 2 files changed, 26 insertions(+), 15 deletions(-) diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 57088272e2..b36679ebcd 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -165,20 +165,21 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se ms.Topo.SetGrowthFreezeConfig( v.GetBool("master.volume_growth.freeze_on_topology_mismatch"), v.GetDuration("master.volume_growth.state_loss_recovery_wait")) - ms.Topo.QuarantineVolumeFn = func(dn *topology.DataNode, vid needle.VolumeId, readOnly bool) { - if err := operation.WithVolumeServerClient(false, pb.ServerAddress(dn.Url()), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + ms.Topo.QuarantineVolumeFn = func(dn *topology.DataNode, vid needle.VolumeId, readOnly bool) error { + err := operation.WithVolumeServerClient(false, dn.ServerAddress(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - var err error if readOnly { - _, err = client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid), CanDelete: true}) - } else { - _, err = client.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{VolumeId: uint32(vid)}) + _, err := client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid), CanDelete: true}) + return err } + _, err := client.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{VolumeId: uint32(vid)}) return err - }); err != nil { + }) + if err != nil { glog.Warningf("quarantine volume %d on %s readOnly=%v: %v", vid, dn.Id(), readOnly, err) } + return err } ms.vg = topology.NewDefaultVolumeGrowth() glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB") diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 1d500caae9..e4e49cc5f3 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -109,7 +109,7 @@ type Topology struct { // QuarantineVolumeFn pushes the quarantine onto the volume server itself // (VolumeMarkReadonly/VolumeMarkWritable). nil keeps quarantine at the // layout level only. - QuarantineVolumeFn func(dn *DataNode, vid needle.VolumeId, readOnly bool) + QuarantineVolumeFn func(dn *DataNode, vid needle.VolumeId, readOnly bool) error } type quarantinedVolume struct { @@ -138,12 +138,15 @@ func (t *Topology) enforceQuarantine(dn *DataNode, vi storage.VolumeInfo) storag state := v.(quarantinedVolumeState) if !state.enforced { state.reported = vi.ReadOnly - t.quarantinedVolumeIds.Store(q, state) - } else if !vi.ReadOnly && t.QuarantineVolumeFn != nil { - // The server lost the pushed mark (restart, manual unquarantine); - // re-push it while the conflict stands. - t.QuarantineVolumeFn(dn, vi.Id, true) } + if t.QuarantineVolumeFn != nil && (!state.enforced || !vi.ReadOnly) { + // The mark never landed, or the server lost it (restart, manual + // unquarantine); push it while the conflict stands. + if t.QuarantineVolumeFn(dn, vi.Id, true) == nil { + state.enforced = true + } + } + t.quarantinedVolumeIds.Store(q, state) } return vi } @@ -706,8 +709,7 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) { if vl != nil { vl.SetVolumeQuarantined(n, v.Id) } - if t.QuarantineVolumeFn != nil { - t.QuarantineVolumeFn(n, v.Id, true) + if t.QuarantineVolumeFn != nil && t.QuarantineVolumeFn(n, v.Id, true) == nil { if val, ok := t.quarantinedVolumeIds.Load(q); ok { state := val.(quarantinedVolumeState) state.enforced = true @@ -745,6 +747,13 @@ func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) { vl.SetReplicaReadOnlyFlag(dn, vid, true) } else { vl.SetVolumeWritable(dn, vid) + if state.enforced && t.QuarantineVolumeFn != nil { + if err := t.QuarantineVolumeFn(dn, vid, false); err != nil { + // The server still rejects writes; keep the mark so a + // later release pass retries the writable push. + return true + } + } } t.quarantinedVolumeIds.Delete(key) return true @@ -781,6 +790,7 @@ func volumeLayoutKey(collection string, rp *super_block.ReplicaPlacement, ttl *n return fmt.Sprintf("collection=%q rp=%s ttl=%s disk=%s", collection, rp, ttl, diskType) } + // findDuplicateVolumeLayouts reports every layout that already holds v.Id on // data nodes other than dn. Volume ids are global, so the same id held by two // nodes under two layouts means the allocator reused an id.