diff --git a/.github/workflows/java_integration_tests.yml b/.github/workflows/java_integration_tests.yml index 363c424768..cdb61b3faf 100644 --- a/.github/workflows/java_integration_tests.yml +++ b/.github/workflows/java_integration_tests.yml @@ -60,6 +60,7 @@ jobs: # Start SeaweedFS with optimized settings for CI # Include S3 API for s3copier integration tests weed server -dir="$WEED_DATA_DIR" \ + -ip=127.0.0.1 -ip.bind=127.0.0.1 \ -master.raftHashicorp \ -master.electionTimeout=1s \ -master.volumeSizeLimitMB=100 \ diff --git a/.github/workflows/s3-go-tests.yml b/.github/workflows/s3-go-tests.yml index 6f5de41974..7660ce1d71 100644 --- a/.github/workflows/s3-go-tests.yml +++ b/.github/workflows/s3-go-tests.yml @@ -54,7 +54,7 @@ jobs: echo "=== System Information ===" uname -a free -h - df -h + df -h . /tmp echo "=== Starting Tests ===" # Run tests with automatic server management @@ -220,7 +220,7 @@ jobs: echo "=== System Information ===" uname -a free -h - df -h + df -h . /tmp echo "=== Starting Tests ===" # Run tests with automatic server management @@ -358,7 +358,7 @@ jobs: echo "=== System Information ===" uname -a free -h - df -h + df -h . /tmp echo "=== Starting Tests ===" make test-with-server @@ -417,7 +417,7 @@ jobs: echo "=== System Information ===" uname -a free -h - df -h + df -h . /tmp echo "=== Starting Tests ===" # Run tests with automatic server management diff --git a/test/s3/versioning/Makefile b/test/s3/versioning/Makefile index f41cfe3905..26ec27704f 100644 --- a/test/s3/versioning/Makefile +++ b/test/s3/versioning/Makefile @@ -94,6 +94,7 @@ start-server: check-deps @echo "🔍 DEBUG: Launching SeaweedFS server in background..." @echo "🔍 DEBUG: Command: $(WEED_BINARY) mini -dir=./test-volume-data -s3.port=$(S3_PORT) -s3.config=../../../docker/compose/s3.json" @$(WEED_BINARY) mini \ + -ip=127.0.0.1 -ip.bind=127.0.0.1 \ -dir=./test-volume-data \ -s3.port=$(S3_PORT) \ -s3.config=../../../docker/compose/s3.json \ @@ -229,13 +230,13 @@ test-with-server: start-server test-versioning-with-configs: check-deps @echo "Testing with different S3 configurations..." @echo "Testing with delete bucket not empty allowed..." - @$(WEED_BINARY) mini -s3.port=$(S3_PORT) -s3.allowDeleteBucketNotEmpty=true > weed-test-config1.log 2>&1 & echo $$! > weed-config1.pid + @$(WEED_BINARY) mini -ip=127.0.0.1 -ip.bind=127.0.0.1 -s3.port=$(S3_PORT) -s3.allowDeleteBucketNotEmpty=true > weed-test-config1.log 2>&1 & echo $$! > weed-config1.pid @sleep 5 @go test -v -timeout=5m -run "TestVersioningBasicWorkflow" . || true @if [ -f weed-config1.pid ]; then kill -TERM $$(cat weed-config1.pid) 2>/dev/null || true; rm -f weed-config1.pid; fi @sleep 2 @echo "Testing with delete bucket not empty disabled..." - @$(WEED_BINARY) mini -s3.port=$(S3_PORT) -s3.allowDeleteBucketNotEmpty=false > weed-test-config2.log 2>&1 & echo $$! > weed-config2.pid + @$(WEED_BINARY) mini -ip=127.0.0.1 -ip.bind=127.0.0.1 -s3.port=$(S3_PORT) -s3.allowDeleteBucketNotEmpty=false > weed-test-config2.log 2>&1 & echo $$! > weed-config2.pid @sleep 5 @go test -v -timeout=5m -run "TestVersioningBasicWorkflow" . || true @if [ -f weed-config2.pid ]; then kill -TERM $$(cat weed-config2.pid) 2>/dev/null || true; rm -f weed-config2.pid; fi @@ -279,6 +280,7 @@ debug-server: @$(MAKE) stop-server @mkdir -p ./test-volume-data @$(WEED_BINARY) mini \ + -ip=127.0.0.1 -ip.bind=127.0.0.1 \ -debug \ -s3.port=$(S3_PORT) \ -s3.allowDeleteBucketNotEmpty=true \ @@ -323,6 +325,7 @@ health-check: start-server-simple: check-deps @echo "Starting SeaweedFS server (simple mode)..." @$(WEED_BINARY) mini \ + -ip=127.0.0.1 -ip.bind=127.0.0.1 \ -dir=$(SERVER_DIR) \ -s3 \ -s3.port=$(S3_PORT) \ diff --git a/weed/server/master_grpc_server_changed_volumes_test.go b/weed/server/master_grpc_server_changed_volumes_test.go index 5436418fb1..0ed15aa132 100644 --- a/weed/server/master_grpc_server_changed_volumes_test.go +++ b/weed/server/master_grpc_server_changed_volumes_test.go @@ -220,7 +220,7 @@ func TestRegistrationRacingCollectionDeleteDoesNotLeak(t *testing.T) { topo.DeleteCollection("c") // the interleaved registration must refuse the dropped layout - if stale.RegisterVolume(&vi, dn) { + if stale.RegisterVolume(&vi, dn, nil) { t.Fatal("registered into a layout dropped with its collection") } diff --git a/weed/topology/duplicate_volume_id_test.go b/weed/topology/duplicate_volume_id_test.go index 42cb6fd97a..ec0ed4aeda 100644 --- a/weed/topology/duplicate_volume_id_test.go +++ b/weed/topology/duplicate_volume_id_test.go @@ -1,7 +1,12 @@ package topology import ( + "errors" + "sync/atomic" "testing" + "time" + + "github.com/stretchr/testify/require" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage" @@ -104,8 +109,102 @@ func TestQuarantineReleasedWhenConflictResolves(t *testing.T) { topo.UnRegisterVolumeLayout(v2, dn2) + // Registration flags duplicates asynchronously, so a flag that was still + // in flight when the conflict resolved releases the surviving copy only + // once it settles. Wait for that release instead of reading once. + deadline := time.Now().Add(5 * time.Second) + for !isWritable(t, topo, v1) { + if time.Now().After(deadline) { + t.Fatal("the surviving copy must be released once the conflict is gone") + } + time.Sleep(10 * time.Millisecond) + } +} + +func TestFailedReleasePushRestoresQuarantineMark(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("000") + + dn1 := NewDataNode("server1") + dn2 := NewDataNode("server2") + + var pushFails atomic.Bool + topo.QuarantineVolumeFn = func(dn *DataNode, vid needle.VolumeId, readOnly bool) error { + if !readOnly && pushFails.Load() { + return errors.New("volume server unreachable") + } + return nil + } + + v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + v2 := storage.VolumeInfo{Id: 7, Collection: "b", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + dn1.UpdateVolumes([]storage.VolumeInfo{v1}) + dn2.UpdateVolumes([]storage.VolumeInfo{v2}) + topo.RegisterVolumeLayout(v1, dn1) + topo.RegisterVolumeLayout(v2, dn2) + // Registrations flag duplicates asynchronously; let them settle so no + // late mark can land after the release assertions below. + topo.duplicateFlagChecks.Wait() + topo.flagDuplicateVolumeId(v2, dn2) + + // the conflict resolves while the surviving copy is unreachable: the + // writable push fails, so the record and the layout read-only mark must + // both come back instead of advertising a writable replica the server + // still holds read-only + pushFails.Store(true) + topo.UnRegisterVolumeLayout(v2, dn2) + + if !topo.isQuarantined(dn1, v1.Id) { + t.Fatal("a failed writable push must keep the quarantine record for the retry") + } + if isWritable(t, topo, v1) { + t.Fatal("the layout must not advertise a replica the server still holds read-only") + } + + pushFails.Store(false) + // An idle replica drops out of the delta reports once its stored flag + // matches the server's, so the release pass retries off the leader pulse + // rather than waiting for the volume to be reported again. + topo.retryPendingQuarantineReleases() + require.Eventually(t, func() bool { + return !topo.isQuarantined(dn1, v1.Id) && isWritable(t, topo, v1) + }, 5*time.Second, time.Millisecond, "the pending release must retry without another volume report") +} + +func TestQuarantineKeptWhenReplicaUnreadable(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("000") + + dn1 := NewDataNode("server1") + dn2 := NewDataNode("server2") + + v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + v2 := storage.VolumeInfo{Id: 7, Collection: "b", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + dn1.UpdateVolumes([]storage.VolumeInfo{v1}) + dn2.UpdateVolumes([]storage.VolumeInfo{v2}) + topo.RegisterVolumeLayout(v1, dn1) + topo.RegisterVolumeLayout(v2, dn2) + topo.duplicateFlagChecks.Wait() + topo.flagDuplicateVolumeId(v2, dn2) + + // the surviving copy drops out of its node's disk map while the lookup + // index still holds it; its layout key is then unknown, so the release + // pass must keep the mark rather than assume the conflict resolved + dn1.UpdateVolumes(nil) + topo.UnRegisterVolumeLayout(v2, dn2) + + if !topo.isQuarantined(dn1, v1.Id) { + t.Fatal("an unreadable replica must not lose its quarantine record") + } + if isWritable(t, topo, v1) { + t.Fatal("a replica whose conflict cannot be ruled out must stay read-only") + } + + dn1.UpdateVolumes([]storage.VolumeInfo{v1}) + topo.RegisterVolumeLayout(v1, dn1) + topo.duplicateFlagChecks.Wait() if !isWritable(t, topo, v1) { - t.Fatal("the surviving copy must be released once the conflict is gone") + t.Fatal("once readable again and sole, the copy must be released to writable") } } @@ -134,6 +233,47 @@ func TestQuarantineSurvivesResync(t *testing.T) { } } +func TestQuarantineSurvivesReplicaRejoin(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + rp, _ := super_block.NewReplicaPlacementFromString("001") + + dn1 := NewDataNode("server1") + dn1.Ip = "10.0.0.1" + dn1.Port = 8080 + dn2 := NewDataNode("server2") + dn2.Ip = "10.0.0.2" + dn2.Port = 8080 + dn3 := NewDataNode("server3") + dn3.Ip = "10.0.0.3" + dn3.Port = 8080 + + v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + v2 := storage.VolumeInfo{Id: 7, Collection: "b", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL} + dn1.UpdateVolumes([]storage.VolumeInfo{v1}) + dn2.UpdateVolumes([]storage.VolumeInfo{v1}) + dn3.UpdateVolumes([]storage.VolumeInfo{v2}) + topo.RegisterVolumeLayout(v1, dn1) + topo.RegisterVolumeLayout(v1, dn2) + topo.RegisterVolumeLayout(v2, dn3) + topo.flagDuplicateVolumeId(v2, dn3) + + // a replica dropped from the lookup index re-registers through the + // self-heal path; rebuilding location flags from reported state must + // not lift the quarantine on any copy + vl := topo.FindVolumeLayout(v1.Collection, v1.ReplicaPlacement, v1.Ttl, types.ToDiskType(v1.DiskType)) + vl.SetVolumeUnavailable(dn1, v1.Id) + topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{ + Id: 7, Collection: "a", ReplicaPlacement: uint32(rp.Byte()), Version: 3, + }}, dn1) + + if isWritable(t, topo, v1) { + t.Fatal("a re-registering replica must not clear the quarantine on the surviving copies") + } + if isWritable(t, topo, v2) { + t.Fatal("the conflicting copy must stay quarantined too") + } +} + func TestReplicationChangeIsNotDuplicate(t *testing.T) { topo := NewTopology("topo", nil, 128, 3, false) rpOld, _ := super_block.NewReplicaPlacementFromString("000") diff --git a/weed/topology/maintenance_mode_test.go b/weed/topology/maintenance_mode_test.go index 27c3f580ab..596ece558f 100644 --- a/weed/topology/maintenance_mode_test.go +++ b/weed/topology/maintenance_mode_test.go @@ -231,7 +231,7 @@ func TestMaintenanceModeHoldsThroughVacuumAndMarkWritable(t *testing.T) { vl.DrainAndRemoveFromWritable(1) topo.SetDataNodeMaintenanceMode(dn1, true) for _, dn := range []*DataNode{dn1, dn2} { - if vl.SetVolumeAvailable(dn, 1, false, false) { + if vl.SetVolumeAvailable(dn, 1, false, false, false) { t.Errorf("vacuum commit on %s made the volume writable with a replica in maintenance", dn.Id()) } } diff --git a/weed/topology/topology.go b/weed/topology/topology.go index cee0018584..07f1ffa765 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -106,6 +106,21 @@ type Topology struct { // vid collided across layouts. Kept separate from the reported flag so // later heartbeats from the same server cannot quietly undo it. quarantinedVolumeIds sync.Map + // quarantineLock serializes each quarantine record update with the layout + // mark it justifies: a release pass and a heartbeat read-only flag update + // each observe record and mark as one state, never a mark whose record + // was already deleted or a cleared mark whose record is not yet visible. + quarantineLock sync.Mutex + // quarantinePushLocks serialize the remote mark/unmark calls per replica + // so a push decided under an older record state cannot land on the + // volume server after a newer one. Keyed by replica so a slow server + // does not stall pushes to the others. + quarantinePushLocks map[quarantinedVolume]*quarantinePushLock + quarantinePushLocksMu sync.Mutex + + // duplicateFlagChecks tracks in-flight flagDuplicateVolumeId passes so + // tests can wait for the async checks a registration started. + duplicateFlagChecks sync.WaitGroup // QuarantineVolumeFn pushes the quarantine onto the volume server itself // (VolumeMarkReadonly/VolumeMarkWritable). nil keeps quarantine at the @@ -121,6 +136,12 @@ type quarantinedVolume struct { type quarantinedVolumeState struct { reported bool enforced bool + retry bool +} + +type quarantinePushLock struct { + sync.Mutex + refs int } func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool { @@ -128,6 +149,102 @@ func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool { return ok } +// pushQuarantine sends the remote mark only while it still matches the +// record: a read-only push requires the pair to exist, a writable push +// requires it to be gone. The bool reports whether the remote call was +// actually made, so callers can tell a skipped stale push from a failed one. +func (t *Topology) pushQuarantine(dn *DataNode, vid needle.VolumeId, readOnly bool) (bool, error) { + unlock := t.lockQuarantinePush(quarantinedVolume{dn.Id(), vid}) + defer unlock() + return t.sendQuarantine(dn, vid, readOnly) +} + +// sendQuarantine runs the record check and remote mark; the caller holds +// the replica's push lock. +func (t *Topology) sendQuarantine(dn *DataNode, vid needle.VolumeId, readOnly bool) (bool, error) { + t.quarantineLock.Lock() + _, exists := t.quarantinedVolumeIds.Load(quarantinedVolume{dn.Id(), vid}) + t.quarantineLock.Unlock() + if exists != readOnly { + return false, nil + } + return true, t.QuarantineVolumeFn(dn, vid, readOnly) +} + +// releaseQuarantineWritable pushes the writable mark and re-arms the record +// on failure, all under the replica's push lock so a queued stale push cannot +// slip between the failure and the restore and undo the re-armed record. +func (t *Topology) releaseQuarantineWritable(dn *DataNode, vid needle.VolumeId) { + unlock := t.lockQuarantinePush(quarantinedVolume{dn.Id(), vid}) + defer unlock() + if _, err := t.sendQuarantine(dn, vid, false); err != nil { + t.restoreQuarantineRecord(dn, vid) + } +} + +// lockQuarantinePush returns the release for q's push lock, creating the lock +// on first use and dropping it once the last holder lets go. +func (t *Topology) lockQuarantinePush(q quarantinedVolume) func() { + t.quarantinePushLocksMu.Lock() + if t.quarantinePushLocks == nil { + t.quarantinePushLocks = make(map[quarantinedVolume]*quarantinePushLock) + } + l, ok := t.quarantinePushLocks[q] + if !ok { + l = &quarantinePushLock{} + t.quarantinePushLocks[q] = l + } + l.refs++ + t.quarantinePushLocksMu.Unlock() + l.Lock() + return func() { + t.quarantinePushLocksMu.Lock() + l.refs-- + if l.refs == 0 { + delete(t.quarantinePushLocks, q) + } + t.quarantinePushLocksMu.Unlock() + l.Unlock() + } +} + +// takeQuarantineRetryLocked disarms a pending release retry and reports +// whether one was armed. The caller holds quarantineLock. +func (t *Topology) takeQuarantineRetryLocked(q quarantinedVolume) bool { + v, ok := t.quarantinedVolumeIds.Load(q) + if !ok { + return false + } + state := v.(quarantinedVolumeState) + if !state.retry { + return false + } + state.retry = false + t.quarantinedVolumeIds.Store(q, state) + return true +} + +// retryPendingQuarantineReleases re-runs the release pass for every record a +// failed writable push re-armed. The retry in enforceQuarantine only fires +// while a heartbeat keeps carrying the volume, and an idle replica drops out +// of the delta reports once its stored flag matches the server's, so pending +// retries also run off the leader pulse. +func (t *Topology) retryPendingQuarantineReleases() { + vids := make(map[needle.VolumeId]struct{}) + t.quarantineLock.Lock() + t.quarantinedVolumeIds.Range(func(key, _ any) bool { + q := key.(quarantinedVolume) + if t.takeQuarantineRetryLocked(q) { + vids[q.vid] = struct{}{} + } + return true + }) + t.quarantineLock.Unlock() + for vid := range vids { + go t.releaseResolvedQuarantine(vid) + } +} + // enforceQuarantine keeps the latest reported read-only flag for a quarantined // copy so a later release restores the server's true state. The stored report // keeps the server's flag: quarantine lives at the layout level, and folding @@ -135,19 +252,55 @@ func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool { // disagree with the server's and force a full volume list on every heartbeat. func (t *Topology) enforceQuarantine(dn *DataNode, vi storage.VolumeInfo) storage.VolumeInfo { q := quarantinedVolume{dn.Id(), vi.Id} - if v, ok := t.quarantinedVolumeIds.Load(q); ok { - state := v.(quarantinedVolumeState) - if !state.enforced { - state.reported = vi.ReadOnly - } - 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.quarantineLock.Lock() + v, ok := t.quarantinedVolumeIds.Load(q) + if !ok { + t.quarantineLock.Unlock() + return vi + } + if t.takeQuarantineRetryLocked(q) { + // A failed writable push re-armed this record; hand the release pass + // one retry per report. Clearing it first keeps a surviving conflict + // on the normal enforce path afterwards. + t.quarantineLock.Unlock() + go t.releaseResolvedQuarantine(vi.Id) + return vi + } + state := v.(quarantinedVolumeState) + if !state.enforced { + state.reported = vi.ReadOnly + } + if t.QuarantineVolumeFn == nil || (state.enforced && vi.ReadOnly) { t.quarantinedVolumeIds.Store(q, state) + t.quarantineLock.Unlock() + return vi + } + t.quarantineLock.Unlock() + + // The mark never landed on the server, or the server lost it (restart, + // manual unquarantine); push it while the conflict stands. The call is + // remote, so it runs outside the lock. + pushed, err := t.pushQuarantine(dn, vi.Id, true) + if pushed && err == nil { + state.enforced = true + } + t.quarantineLock.Lock() + cur, still := t.quarantinedVolumeIds.Load(q) + if still { + latest := cur.(quarantinedVolumeState) + if !latest.enforced { + latest.reported = state.reported + } + latest.enforced = latest.enforced || state.enforced + t.quarantinedVolumeIds.Store(q, latest) + t.quarantineLock.Unlock() + return vi + } + t.quarantineLock.Unlock() + if state.enforced && !state.reported { + // The conflict resolved while the push was in flight; leave the + // server consistent with the released layout. + t.releaseQuarantineWritable(dn, vi.Id) } return vi } @@ -173,6 +326,7 @@ func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, puls t.Configuration = &Configuration{} t.dataNodeIndex = make(map[string]*DataNode) + t.quarantinePushLocks = make(map[quarantinedVolume]*quarantinePushLock) return t } @@ -678,18 +832,31 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) { diskType := types.ToDiskType(v.DiskType) for { vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType) - if vl.RegisterVolume(&v, dn) { - if t.isQuarantined(dn, v.Id) { - vl.SetVolumeQuarantined(dn, v.Id) - } + // RegisterVolume reads the quarantine record and marks under + // accessLock; holding quarantineLock keeps that pair serialized with + // releaseResolvedQuarantine's clear-and-delete. + t.quarantineLock.Lock() + registered := vl.RegisterVolume(&v, dn, func(n *DataNode) bool { + return t.isQuarantined(n, v.Id) + }) + t.quarantineLock.Unlock() + if registered { vl.EnsureCorrectWritables(&v) - go t.flagDuplicateVolumeId(v, dn) + t.flagDuplicateVolumeIdAsync(v, dn) return } // Dropped with its collection; the next lookup creates a fresh one. } } +func (t *Topology) flagDuplicateVolumeIdAsync(v storage.VolumeInfo, dn *DataNode) { + t.duplicateFlagChecks.Add(1) + go func() { + defer t.duplicateFlagChecks.Done() + t.flagDuplicateVolumeId(v, dn) + }() +} + // flagDuplicateVolumeId reports and quarantines v.Id when another data node // already serves it under a different layout key. It runs after registration // so a legitimate same-node move (disk type or collection change) settles @@ -697,13 +864,14 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) { // quarantined: either could be the impostor, and arrival order must not // decide which copy stays writable. func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) { + // Every pass ends with a release: a conflict could resolve between the + // scan below and the marks, and a check finding no hits is also the retry + // a kept record waits for once its replica reports a readable layout. + defer t.releaseResolvedQuarantine(v.Id) hits := t.findDuplicateVolumeLayouts(v, dn) if len(hits) == 0 { return } - // A conflict could resolve between the scan above and the marks below; - // re-run the release so a stale quarantine is cleared instead of kept. - defer t.releaseResolvedQuarantine(v.Id) stats.MasterDuplicateVolumeIds.WithLabelValues(v.Collection, v.Id.String()).Inc() var otherKeys []string for _, h := range hits { @@ -716,16 +884,33 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) { if vi, err := n.GetVolumesById(v.Id); err == nil { reported = vi.ReadOnly } + t.quarantineLock.Lock() t.quarantinedVolumeIds.LoadOrStore(q, quarantinedVolumeState{reported: reported}) if vl != nil { vl.SetVolumeQuarantined(n, v.Id) } - 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 - t.quarantinedVolumeIds.Store(q, state) - } + t.quarantineLock.Unlock() + if t.QuarantineVolumeFn == nil { + return + } + pushed, err := t.pushQuarantine(n, v.Id, true) + if err != nil { + return + } + t.quarantineLock.Lock() + val, ok := t.quarantinedVolumeIds.Load(q) + if ok { + state := val.(quarantinedVolumeState) + state.enforced = state.enforced || pushed + t.quarantinedVolumeIds.Store(q, state) + t.quarantineLock.Unlock() + return + } + t.quarantineLock.Unlock() + if pushed && !reported { + // The conflict resolved while the push was in flight; leave the + // server consistent with the released layout. + t.releaseQuarantineWritable(n, v.Id) } } quarantine(dn, t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType))) @@ -738,7 +923,14 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) { // releaseResolvedQuarantine lifts read-only marks once the conflicting copy // is gone, so a finished move does not leave the surviving copy quarantined. +// The writable push is a remote call, so the locked pass only applies the +// layout marks and records which replicas still need it; the pushes run +// after the lock is dropped, a skipped push means a newer record re-armed +// the pair so its quarantine stands, and a failed push re-adds the record +// so a later release pass retries it. func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) { + var writablePushes []*DataNode + t.quarantineLock.Lock() t.quarantinedVolumeIds.Range(func(key, val any) bool { q := key.(quarantinedVolume) if q.vid != vid { @@ -749,6 +941,12 @@ func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) { t.quarantinedVolumeIds.Delete(key) return true } + if vi.ReplicaPlacement == nil { + // The replica is in the lookup index but unreadable on its node, + // so its layout key is unknown and the conflict cannot be ruled + // out; keep the mark for a later pass. + return true + } if hits := t.findDuplicateVolumeLayouts(vi, dn); len(hits) > 0 { return true } @@ -759,16 +957,38 @@ func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) { } 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 - } + writablePushes = append(writablePushes, dn) } } t.quarantinedVolumeIds.Delete(key) return true }) + t.quarantineLock.Unlock() + for _, dn := range writablePushes { + t.releaseQuarantineWritable(dn, vid) + } +} + +// restoreQuarantineRecord re-adds a released record after a failed writable +// push so the next volume report retries the release. An existing record keeps its +// reported flag; the server is known read-only, so enforced and retry are set. +// A record restored from scratch also gets its layout read-only mark back: +// the release pass cleared it with the record, and leaving it cleared would +// advertise a writable replica the server still holds read-only. +func (t *Topology) restoreQuarantineRecord(dn *DataNode, vid needle.VolumeId) { + t.quarantineLock.Lock() + defer t.quarantineLock.Unlock() + key := quarantinedVolume{dn.Id(), vid} + if cur, ok := t.quarantinedVolumeIds.LoadOrStore(key, quarantinedVolumeState{enforced: true, retry: true}); ok { + state := cur.(quarantinedVolumeState) + state.enforced = true + state.retry = true + t.quarantinedVolumeIds.Store(key, state) + return + } + if _, vl, _ := t.findQuarantinedVolume(dn.Id(), vid); vl != nil { + vl.SetVolumeReadOnly(dn, vid) + } } func (t *Topology) findQuarantinedVolume(dnId NodeId, vid needle.VolumeId) (*DataNode, *VolumeLayout, storage.VolumeInfo) { @@ -983,7 +1203,9 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati if v.ReplicaPlacement == nil { continue } + t.quarantineLock.Lock() t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)).SetReplicaReadOnlyFlag(dn, v.Id, v.ReadOnly || t.isQuarantined(dn, v.Id)) + t.quarantineLock.Unlock() } // Update effective sizes for all reported volumes (decay pending estimates). // If decay brings a volume eagerly removed by RecordAssign back under the @@ -1002,12 +1224,20 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati // Without this, the volume stays visible in volume.list/admin UI yet // LookupVolume returns "volume id not found". if !vl.HasDataNode(v.Id, dn) { - for !vl.RegisterVolume(&v, dn) { + for { + t.quarantineLock.Lock() + registered := vl.RegisterVolume(&v, dn, func(n *DataNode) bool { + return t.isQuarantined(n, v.Id) + }) + t.quarantineLock.Unlock() + if registered { + break + } // Dropped with its collection; the next lookup creates a fresh // one, which the calls below must use too. vl = t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType) } - go t.flagDuplicateVolumeId(v, dn) + t.flagDuplicateVolumeIdAsync(v, dn) // Volumes new to the disk map were registered above, so reaching // here means only the lookup index had lost it. Clients were told // it went when the node dropped out, so the repair has to tell them @@ -1108,18 +1338,28 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess // volume only that index had lost is an arrival as far as clients are // concerned: they were told it went when the node dropped out. becameServable := !vl.HasDataNode(vi.Id, dn) - for becameServable && !vl.RegisterVolume(&vi, dn) { + for becameServable { + t.quarantineLock.Lock() + registered := vl.RegisterVolume(&vi, dn, func(n *DataNode) bool { + return t.isQuarantined(n, vi.Id) + }) + t.quarantineLock.Unlock() + if registered { + break + } // Dropped with its collection; the next lookup creates a fresh one. vl = t.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, vi.Ttl, types.ToDiskType(vi.DiskType)) } if becameServable { - go t.flagDuplicateVolumeId(vi, dn) + t.flagDuplicateVolumeIdAsync(vi, dn) } if isNew || becameServable || tierTransition || isChanged { newVolumes = append(newVolumes, vi) } if isChanged { + t.quarantineLock.Lock() vl.SetReplicaReadOnlyFlag(dn, vi.Id, vi.ReadOnly || t.isQuarantined(dn, vi.Id)) + t.quarantineLock.Unlock() } vl.UpdateOversizedState(&vi, dn) if vl.UpdateVolumeSize(vi.Id, vi.Size, vi.CompactRevision, true) { diff --git a/weed/topology/topology_event_handling.go b/weed/topology/topology_event_handling.go index 9dcd6f93e3..1793ff0e93 100644 --- a/weed/topology/topology_event_handling.go +++ b/weed/topology/topology_event_handling.go @@ -23,6 +23,7 @@ func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, g freshThreshHold := time.Now().Unix() - 3*t.pulse //3 times of sleep interval t.CollectDeadNodeAndFullVolumes(freshThreshHold, t.volumeSizeLimit, growThreshold) t.DecayQuietVolumeSizes() + t.retryPendingQuarantineReleases() } time.Sleep(time.Duration(float32(t.pulse*1e3)*(1+rand.Float32())) * time.Millisecond) } diff --git a/weed/topology/topology_test.go b/weed/topology/topology_test.go index 3a89ce14f5..e761ede3c5 100644 --- a/weed/topology/topology_test.go +++ b/weed/topology/topology_test.go @@ -851,7 +851,7 @@ func TestSetVolumeAvailableRepairsMissingVolume(t *testing.T) { } // The vacuum commit re-marks the volume available; it must re-register it. - vl.SetVolumeAvailable(dn, vid, false, false) + vl.SetVolumeAvailable(dn, vid, false, false, false) if got := topo.Lookup("", vid); len(got) != 1 { t.Fatalf("after SetVolumeAvailable: lookup %d got %v, want 1 location", vid, got) } diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go index 0cc779d0ea..ea6de330f4 100644 --- a/weed/topology/topology_vacuum.go +++ b/weed/topology/topology_vacuum.go @@ -226,7 +226,9 @@ func (t *Topology) batchVacuumVolumeCommit(grpcDialOption grpc.DialOption, vl *V vl.accessLock.Unlock() for _, dn := range vacuumLocationList.list { - vl.SetVolumeAvailable(dn, vid, isReadOnly, isFullCapacity) + t.quarantineLock.Lock() + vl.SetVolumeAvailable(dn, vid, isReadOnly, isFullCapacity, t.isQuarantined(dn, vid)) + t.quarantineLock.Unlock() } } return isCommitSuccess diff --git a/weed/topology/volume_growth_test.go b/weed/topology/volume_growth_test.go index 11e3ba0829..da9cf5affd 100644 --- a/weed/topology/volume_growth_test.go +++ b/weed/topology/volume_growth_test.go @@ -116,7 +116,7 @@ func setup(topologyLayout string) *Topology { } if vi.ReplicaPlacement != nil { vl := topo.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, needle.EMPTY_TTL, types.HardDriveType) - vl.RegisterVolume(&vi, server) + vl.RegisterVolume(&vi, server, nil) vl.setVolumeWritable(vi.Id) } server.AddOrUpdateVolume(vi) diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index b0a48e750b..c74fefccbc 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -132,7 +132,11 @@ func (vl *VolumeLayout) initSizeTracking(vid needle.VolumeId, size uint64, compa // RegisterVolume records a volume location. It refuses a layout dropped with // its collection — the caller must re-resolve, or the bits it sets would leak. -func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) bool { +// Each replica's read-only flag is rebuilt from what its node last reported; +// isQuarantined folds the master's force-read-only marks into that rebuild, +// since reported flags deliberately exclude them. nil restores reported state +// verbatim. +func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode, isQuarantined func(*DataNode) bool) bool { vl.accessLock.Lock() defer vl.accessLock.Unlock() @@ -140,27 +144,27 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) bool return false } - moveLookupOwnership(v.Id, vl.getOrCreateLocationList(v.Id).Set(dn), dn) + location := vl.getOrCreateLocationList(v.Id) + moveLookupOwnership(v.Id, location.Set(dn), dn) if !v.ReadOnly { vl.initSizeTracking(v.Id, v.Size, v.CompactRevision) } + if isQuarantined != nil && isQuarantined(dn) { + location.SetReadOnly(dn, true) + vl.removeFromWritable(v.Id) + } // glog.V(4).Infof("volume %d added to %s len %d copy %d", v.Id, dn.Id(), vl.vid2location[v.Id].Length(), v.ReplicaPlacement.GetCopyCount()) - location := vl.vid2location[v.Id] - for _, dn := range location.list { - if vInfo, err := dn.GetVolumesById(v.Id); err == nil { - if vInfo.ReadOnly { - glog.V(1).Infof("vid %d removed from writable", v.Id) - vl.removeFromWritable(v.Id) - location.SetReadOnly(dn, true) - return true - } - location.SetReadOnly(dn, false) - } else { - glog.V(1).Infof("vid %d removed from writable", v.Id) - vl.removeFromWritable(v.Id) - location.SetReadOnly(dn, false) - return true + for _, n := range location.list { + marked := isQuarantined != nil && isQuarantined(n) + vInfo, err := n.GetVolumesById(v.Id) + if err == nil && !vInfo.ReadOnly && !marked { + location.SetReadOnly(n, false) + continue } + glog.V(1).Infof("vid %d removed from writable", v.Id) + vl.removeFromWritable(v.Id) + location.SetReadOnly(n, marked || (err == nil && vInfo.ReadOnly)) + return true } return true } @@ -1146,7 +1150,7 @@ func (vl *VolumeLayout) SetVolumeUnavailable(dn *DataNode, vid needle.VolumeId) } return false } -func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, isReadOnly, isFullCapacity bool) (becameWritable bool) { +func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, isReadOnly, isFullCapacity, quarantined bool) (becameWritable bool) { restoreActiveVolumeCount := false vl.accessLock.Lock() defer func() { @@ -1167,8 +1171,13 @@ func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, is // A disconnect during a long vacuum can drop the entry while the volume is // still on the node; re-create it instead of dereferencing a nil location, // so the commit also repairs the split. - moveLookupOwnership(vid, vl.getOrCreateLocationList(vid).Set(dn), dn) + location := vl.getOrCreateLocationList(vid) + moveLookupOwnership(vid, location.Set(dn), dn) + if quarantined { + location.SetReadOnly(dn, true) + return false + } if vInfo.ReadOnly || isReadOnly || isFullCapacity { return false } diff --git a/weed/topology/volume_layout_capacity_recovery_test.go b/weed/topology/volume_layout_capacity_recovery_test.go index 8edb9eead3..686c67e4ba 100644 --- a/weed/topology/volume_layout_capacity_recovery_test.go +++ b/weed/topology/volume_layout_capacity_recovery_test.go @@ -98,7 +98,7 @@ func TestSetVolumeAvailableRestoresActiveCountForCapacityFullVolume(t *testing.T dn := vl.vid2location[1].list[0] vl.accessLock.RUnlock() - if !vl.SetVolumeAvailable(dn, 1, false, false) { + if !vl.SetVolumeAvailable(dn, 1, false, false, false) { t.Fatalf("SetVolumeAvailable should report the volume became writable") } if got := topo.diskUsages.usages[types.HardDriveType].activeVolumeCount; got != initialActive { diff --git a/weed/topology/volume_layout_pick_test.go b/weed/topology/volume_layout_pick_test.go index 4fff615a65..aa8cc99c1f 100644 --- a/weed/topology/volume_layout_pick_test.go +++ b/weed/topology/volume_layout_pick_test.go @@ -61,7 +61,7 @@ func setupWithLimit(t testing.TB, topologyLayout string, volumeSizeLimit uint64) } if vi.ReplicaPlacement != nil { vl := topo.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, needle.EMPTY_TTL, types.HardDriveType) - vl.RegisterVolume(&vi, server) + vl.RegisterVolume(&vi, server, nil) vl.setVolumeWritable(vi.Id) } server.AddOrUpdateVolume(vi)