mini: lift the post-mint allocation hold on first inventory report (#12061)

* mini: lift the post-mint allocation hold on first inventory report

DeferAllocation armed a fixed warmup window whenever the leader minted a
fresh topology id, so the first write on a brand-new weed mini blocked
for the whole window even though no prior cluster state could exist.
The window still has to cover a rebuilt master's veteran servers, which
report on their first post-reconnect heartbeat.

Mini has exactly one possible reporter — the embedded volume server,
whose first heartbeat carries the full inventory — so hold allocation
only until that report is applied. The bound raise and topology-id
freeze checks run while the heartbeat is processed, so a wiped master
state still raises the bound or trips the freeze before the hold lifts.

Standalone weed master and weed server keep the fixed window: volume
servers outside the process can still be on their way.

* master: apply queued topology-id reports before opening the allocation gate

SetTopologyInitPending(false) cleared the pending flag before replaying
topology ids stashed while the local id was still being minted. A
NextVolumeId landing in the drain window saw no gate and no freeze yet.
The post-mint allocation hold used to cover the gap by delaying the
winner past the freeze; once the hold can release on first inventory,
the gate must stay closed until the queued checks have run.

* mini: gate the first-report release on single-master mode

releaseAllocOnFirstReport assumes the embedded volume server is the only
possible reporter of prior cluster state. A mini started with
-master.peers joins other masters, whose own volume servers can still be
on their way when the first report lands, so keep the regular
quiet-pulse hold there.

* master: check reported topology id on empty full reports

A HasNoVolumes heartbeat carries the same LastSeenTopologyId a populated
list does, but the mismatch freeze only ran for non-empty reports. An
EC-only veteran could therefore release the post-mint allocation hold
through NoteInventoryApplied while its stale topology id never reached
CheckReportedTopologyId.

* master: drain the topology-id stash inside the init-gate lock

CheckReportedTopologyId only held reportedTopologyIdsLock while stashing,
so a report in flight during SetTopologyInitPending(false) could land its
freeze after topologyInitPending cleared, letting a NextVolumeId slip
through the gap. The whole check now runs under the lock, and the gate is
released inside the same critical section as the drain.

* mini, master: pin and enforce the single volume-server reporter

isSingleMasterMode only checks master peers; SendHeartbeat still accepted
any volume server, so a late foreign report could carry prior cluster
state after the first-report release dropped the hold. Mini now pins the
embedded volume server's address and the master refuses registration from
any other reporter while ReleaseAllocOnFirstReport is set.

* master: admit the volume-server reporter before applying heartbeat state

* test: cover SendHeartbeat admission ordering

A refused reporter must not freeze growth or raise the id bound, and the
pinned reporter's stale topology id still must.

* mini: re-derive the single-reporter release on every run

The package-level master options survive in-process reruns, so a
multi-master mini following a single-master one must not inherit the
first-report release or a stale reporter pin.

* master: admit addressless first heartbeats under the reporter pin

An unregistered heartbeat with no Ip skipped the single-reporter
admission check but still reached the seen-bound raise and the
topology-id freeze before the delta-heartbeat skip. Admission now runs
for every unregistered heartbeat; an empty address can never equal the
pin.

weed/command/master.go

@@ -82,6 +82,15 @@ type MasterOptions struct {
 	// is cancelled. Used by integration tests and by weed mini; nil for
 	// standalone weed master.
 	shutdownCtx context.Context
+	// releaseAllocOnFirstReport lifts the post-mint allocation hold as soon as
+	// one full volume-server inventory has been applied. weed mini sets it in
+	// single-master mode, with singleReporterAddr pinning the embedded volume
+	// server as the only admitted reporter: any other volume server is refused,
+	// so the warmup window after a fresh topology id is dead time.
+	// Standalone weed master / weed server can face late-arriving veterans and
+	// must keep the window.
+	releaseAllocOnFirstReport bool
+	singleReporterAddr        string
 }
 
 func init() {
@@ -488,14 +497,16 @@ func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOp
 		VacuumIntervalSeconds:         *m.vacuumIntervalSeconds,
 		VacuumDeleteEmptyAfterSeconds: *m.vacuumDeleteEmptyAfterSeconds,
 		// PulseSeconds:            *m.pulseSeconds,
-		DefaultReplicaPlacement: *m.defaultReplication,
-		GarbageThreshold:        *m.garbageThreshold,
-		WhiteList:               whiteList,
-		DisableHttp:             *m.disableHttp,
-		MetricsAddress:          *m.metricsAddress,
-		MetricsIntervalSec:      *m.metricsIntervalSec,
-		TelemetryUrl:            *m.telemetryUrl,
-		TelemetryEnabled:        *m.telemetryEnabled,
-		SingleMaster:            isSingleMasterMode(*m.peers),
+		DefaultReplicaPlacement:   *m.defaultReplication,
+		GarbageThreshold:          *m.garbageThreshold,
+		WhiteList:                 whiteList,
+		DisableHttp:               *m.disableHttp,
+		MetricsAddress:            *m.metricsAddress,
+		MetricsIntervalSec:        *m.metricsIntervalSec,
+		TelemetryUrl:              *m.telemetryUrl,
+		TelemetryEnabled:          *m.telemetryEnabled,
+		SingleMaster:              isSingleMasterMode(*m.peers),
+		ReleaseAllocOnFirstReport: m.releaseAllocOnFirstReport,
+		SingleVolumeReporter:      m.singleReporterAddr,
 	}
 }

weed/command/mini.go

@@ -1564,6 +1564,22 @@ func runMini(cmd *Command, args []string) bool {
 	// tear those services down too. On Ctrl+C they rely on their own
 	// OnInterrupt hooks (see grace's LIFO ordering).
 	miniMasterOptions.shutdownCtx = MiniClusterCtx
+	// In single-master mode mini's embedded volume server is the only reporter
+	// of prior cluster state; its first heartbeat already carries the full
+	// inventory, so the post-mint allocation hold ends there instead of
+	// running the whole warmup window on every fresh start. The master pins
+	// the embedded address and refuses any other volume server, keeping the
+	// single-reporter assumption true. Multi-master minis can field
+	// late-reporting volume servers and keep the regular hold.
+	// Both fields are re-derived on every run: the package-level options
+	// survive in-process reruns, so a multi-master mini after a
+	// single-master one must not inherit the release or the stale pin.
+	singleMaster := isSingleMasterMode(*miniMasterOptions.peers)
+	miniMasterOptions.releaseAllocOnFirstReport = singleMaster
+	miniMasterOptions.singleReporterAddr = ""
+	if singleMaster {
+		miniMasterOptions.singleReporterAddr = util.JoinHostPort(*miniIp, *miniOptions.v.port)
+	}
 	miniOptions.v.shutdownCtx = MiniClusterCtx
 	miniFilerOptions.shutdownCtx = MiniClusterCtx
 	// Mini is a small/dev setup with short-lived RPCs; cap the filer's

weed/server/master_grpc_server.go

@@ -231,12 +231,30 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 			continue
 		}
 
-		if (dn == nil && heartbeat.Ip != "") || reportsPriorClusterState(heartbeat, ms.Topo.GetTopologyId(), ms.Topo.GetMaxVolumeId()) {
+		if dn == nil {
+			// An unregistered heartbeat carries no address to admit; under
+			// the pinned single reporter it can never be the embedded one,
+			// so it must not move allocation or topology state either.
+			addr := ""
+			if heartbeat.Ip != "" {
+				addr = util.JoinHostPort(heartbeat.Ip, int(heartbeat.Port))
+			}
+			if !ms.Topo.AdmitVolumeReporter(addr) {
+				err := fmt.Errorf("volume server %s rejected: this master admits a single volume-server reporter", addr)
+				glog.Warningf("SendHeartbeat: %v", err)
+				return err
+			}
+		}
+		priorState := reportsPriorClusterState(heartbeat, ms.Topo.GetTopologyId(), ms.Topo.GetMaxVolumeId())
+		if (dn == nil && heartbeat.Ip != "") || priorState {
 			ms.Topo.NoteAllocationEvidence()
 		}
 		ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)
 		ms.Topo.UpAdjustMaxVolumeId(needle.VolumeId(heartbeat.MaxVolumeIdSeen))
-		if len(heartbeat.Volumes) > 0 {
+		// A HasNoVolumes report still carries the reporter's LastSeenTopologyId;
+		// skipping it would let an EC-only veteran's stale id miss the mismatch
+		// freeze before the hold release below.
+		if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
 			ms.Topo.CheckReportedTopologyId(heartbeat.LastSeenTopologyId)
 		}
 		if dn == nil {
@@ -356,6 +374,15 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 			}
 		}
 
+		// A full volume list is the heartbeat section that settles what a
+		// reporter holds; once applied, a post-mint allocation hold kept for
+		// prior cluster state has nothing left to wait for. A report carrying
+		// such state is itself the evidence the hold exists for, so it keeps
+		// the window for other veterans still due.
+		if (len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes) && !priorState {
+			ms.Topo.NoteInventoryApplied()
+		}
+
 		if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 {
 			stats.MasterReceivedHeartbeatCounter.WithLabelValues("newEcShards").Inc()
 			// update master internal volume layouts

weed/server/master_grpc_server_heartbeat_admit_test.go

@@ -0,0 +1,110 @@
+package weed_server
+
+import (
+	"io"
+	"testing"
+
+	"github.com/seaweedfs/raft"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+)
+
+// admitTestRaftServer answers as leader and swallows Do so a growth freeze
+// lands its local flag without needing a real raft log.
+type admitTestRaftServer struct {
+	raft.Server
+}
+
+func (admitTestRaftServer) State() string                        { return raft.Leader }
+func (admitTestRaftServer) Do(raft.Command) (interface{}, error) { return nil, nil }
+
+type fakeHeartbeatStream struct {
+	grpc.ServerStream
+	heartbeats []*master_pb.Heartbeat
+	sent       []*master_pb.HeartbeatResponse
+}
+
+func (s *fakeHeartbeatStream) Recv() (*master_pb.Heartbeat, error) {
+	if len(s.heartbeats) == 0 {
+		return nil, io.EOF
+	}
+	hb := s.heartbeats[0]
+	s.heartbeats = s.heartbeats[1:]
+	return hb, nil
+}
+
+func (s *fakeHeartbeatStream) Send(resp *master_pb.HeartbeatResponse) error {
+	s.sent = append(s.sent, resp)
+	return nil
+}
+
+// A server the single-reporter admission refuses must not leave any applied
+// state behind: its stale topology id cannot freeze growth and its seen bound
+// cannot raise the master's.
+func TestSendHeartbeatRejectsForeignReporterBeforeApplyingState(t *testing.T) {
+	ms := newLeaderMaster()
+	ms.Topo.RaftServer = admitTestRaftServer{}
+	ms.Topo.SetGrowthFreezeConfig(true, 0)
+	ms.Topo.SetTopologyId("cluster-b")
+	ms.Topo.ReleaseAllocationOnFirstReport("10.0.0.1:8080")
+
+	stream := &fakeHeartbeatStream{heartbeats: []*master_pb.Heartbeat{{
+		Ip:                 "10.0.0.2",
+		Port:               8080,
+		MaxVolumeIdSeen:    7,
+		LastSeenTopologyId: "cluster-a",
+		Volumes:            []*master_pb.VolumeInformationMessage{{Id: 3}},
+	}}}
+
+	err := ms.SendHeartbeat(stream)
+	require.ErrorContains(t, err, "rejected")
+	assert.False(t, ms.Topo.IsGrowthFrozen())
+	assert.Zero(t, uint32(ms.Topo.GetMaxVolumeId()))
+	assert.Empty(t, stream.sent)
+}
+
+// The pinned reporter is still subject to the mismatch freeze: admission gates
+// who may report, not what a legitimate report can do.
+func TestSendHeartbeatAdmittedReporterStillFreezesOnMismatch(t *testing.T) {
+	ms := newLeaderMaster()
+	ms.Topo.RaftServer = admitTestRaftServer{}
+	ms.Topo.SetGrowthFreezeConfig(true, 0)
+	ms.Topo.SetTopologyId("cluster-b")
+	ms.Topo.ReleaseAllocationOnFirstReport("10.0.0.1:8080")
+
+	stream := &fakeHeartbeatStream{heartbeats: []*master_pb.Heartbeat{{
+		Ip:                 "10.0.0.1",
+		Port:               8080,
+		MaxVolumeIdSeen:    7,
+		LastSeenTopologyId: "cluster-a",
+		Volumes:            []*master_pb.VolumeInformationMessage{{Id: 3}},
+	}}}
+
+	_ = ms.SendHeartbeat(stream)
+	assert.True(t, ms.Topo.IsGrowthFrozen())
+}
+
+// An unregistered heartbeat with no address can never be the pinned reporter:
+// its stale topology id and seen bound must not reach the topology on the way
+// to the delta-heartbeat skip.
+func TestSendHeartbeatRejectsAddresslessFirstHeartbeat(t *testing.T) {
+	ms := newLeaderMaster()
+	ms.Topo.RaftServer = admitTestRaftServer{}
+	ms.Topo.SetGrowthFreezeConfig(true, 0)
+	ms.Topo.SetTopologyId("cluster-b")
+	ms.Topo.ReleaseAllocationOnFirstReport("10.0.0.1:8080")
+
+	stream := &fakeHeartbeatStream{heartbeats: []*master_pb.Heartbeat{{
+		MaxVolumeIdSeen:    7,
+		LastSeenTopologyId: "cluster-a",
+		Volumes:            []*master_pb.VolumeInformationMessage{{Id: 3}},
+	}}}
+
+	err := ms.SendHeartbeat(stream)
+	require.ErrorContains(t, err, "rejected")
+	assert.False(t, ms.Topo.IsGrowthFrozen())
+	assert.Zero(t, uint32(ms.Topo.GetMaxVolumeId()))
+	assert.Empty(t, stream.sent)
+}

weed/server/master_server.go

@@ -70,6 +70,13 @@ type MasterOption struct {
 	TelemetryUrl            string
 	TelemetryEnabled        bool
 	VolumeGrowthDisabled    bool
+	// ReleaseAllocOnFirstReport lifts the post-mint allocation hold as soon as
+	// one full volume-server inventory has been applied, for setups whose
+	// embedded volume server is the only possible reporter (weed mini).
+	// SingleVolumeReporter pins that reporter's ip:port; any other volume
+	// server trying to register is rejected.
+	ReleaseAllocOnFirstReport bool
+	SingleVolumeReporter      string
 }
 
 type MasterServer struct {
@@ -182,6 +189,9 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 			ms.Topo.SetAllocationDeferral(d)
 		}
 	}
+	if option.ReleaseAllocOnFirstReport {
+		ms.Topo.ReleaseAllocationOnFirstReport(option.SingleVolumeReporter)
+	}
 	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)

weed/topology/growth_freeze_test.go

@@ -191,3 +191,61 @@ func TestAllocationDeferralDuration(t *testing.T) {
 		t.Fatalf("negative deferral should restore the default, got %v", d)
 	}
 }
+
+func TestAllocationDeferralReleasedByInventory(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	topo.ReleaseAllocationOnFirstReport()
+
+	topo.DeferAllocation(time.Minute)
+	topo.NoteInventoryApplied()
+	if wait := topo.allocationDelay(); wait > 0 {
+		t.Fatalf("applied inventory should lift the deferral, still waiting %v", wait)
+	}
+
+	fresh := NewTopology("topo", nil, 128, 3, false)
+	fresh.ReleaseAllocationOnFirstReport()
+	fresh.NoteInventoryApplied()
+	fresh.DeferAllocation(time.Minute)
+	if wait := fresh.allocationDelay(); wait > 0 {
+		t.Fatalf("a report applied before the mint should keep the hold unarmed, got %v", wait)
+	}
+
+	walled := NewTopology("topo", nil, 128, 3, false)
+	walled.DeferAllocation(time.Minute)
+	walled.NoteInventoryApplied()
+	if wait := walled.allocationDelay(); wait <= 0 {
+		t.Fatal("without opt-in an applied inventory must not lift the deferral")
+	}
+}
+
+func TestQueuedTopologyReportFreezesBeforeGateOpens(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	topo.SetGrowthFreezeConfig(true, 0)
+	topo.SetTopologyInitPending(true)
+
+	// A report arriving while the id is still being minted is stashed.
+	topo.CheckReportedTopologyId("cluster-a")
+	topo.SetTopologyId("cluster-b")
+
+	topo.SetTopologyInitPending(false)
+	if topo.topologyInitPending.Load() {
+		t.Fatal("init gate should be open after pending=false")
+	}
+	if !topo.IsGrowthFrozen() {
+		t.Fatal("a stashed mismatch must freeze growth before the gate opens")
+	}
+}
+
+func TestSingleReporterAdmission(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	if !topo.AdmitVolumeReporter("10.0.0.1:8080") {
+		t.Fatal("without opt-in any reporter is admitted")
+	}
+	topo.ReleaseAllocationOnFirstReport("10.0.0.1:8080")
+	if !topo.AdmitVolumeReporter("10.0.0.1:8080") {
+		t.Fatal("the pinned reporter should be admitted")
+	}
+	if topo.AdmitVolumeReporter("10.0.0.2:8080") {
+		t.Fatal("a foreign reporter should be refused")
+	}
+}

weed/topology/topology.go

@@ -101,6 +101,15 @@ type Topology struct {
 	allocationDeferral      atomic.Int64
 	topologyInitPending     atomic.Bool
 
+	// releaseAllocOnReport lifts a post-mint allocation hold as soon as a full
+	// volume-server inventory has been applied, for single-reporter setups
+	// (weed mini) whose embedded volume server is the only possible source of
+	// prior cluster state. inventoryApplied records that at least one full
+	// volume list has been synced into this topology.
+	releaseAllocOnReport atomic.Bool
+	inventoryApplied     atomic.Bool
+	singleReporter       atomic.Value // string
+
 	reportedTopologyIdsLock    sync.Mutex
 	pendingReportedTopologyIds map[string]struct{}
 
@@ -1477,17 +1486,25 @@ func (t *Topology) CheckReportedTopologyId(reported string) {
 	if reported == "" || !t.freezeOnMismatch.Load() {
 		return
 	}
+	t.reportedTopologyIdsLock.Lock()
+	defer t.reportedTopologyIdsLock.Unlock()
+	t.checkReportedTopologyIdLocked(reported)
+}
+
+// checkReportedTopologyIdLocked runs the mismatch check inside
+// reportedTopologyIdsLock, the same critical section the init gate drains
+// under: a report received before the gate opens is either stashed or frozen
+// before the gate can release.
+func (t *Topology) checkReportedTopologyIdLocked(reported string) {
 	own := t.GetTopologyId()
 	if own == "" {
 		// The local topology id is still being minted. Keep the report so a
 		// later heartbeat echo cannot erase the mismatch on the server side
 		// before the freeze check ever ran.
-		t.reportedTopologyIdsLock.Lock()
 		if t.pendingReportedTopologyIds == nil {
 			t.pendingReportedTopologyIds = make(map[string]struct{})
 		}
 		t.pendingReportedTopologyIds[reported] = struct{}{}
-		t.reportedTopologyIdsLock.Unlock()
 		return
 	}
 	if reported == own {
@@ -1568,17 +1585,58 @@ func (t *Topology) applyGrowthFreeze(frozen bool) {
 
 // SetTopologyInitPending gates allocation while a topology id is being
 // minted, so growth cannot slip in before the warmup window is installed.
+// Stashed reports are applied while the gate is still closed: clearing it
+// first would let an allocation through before a queued mismatch freezes
+// growth.
 func (t *Topology) SetTopologyInitPending(pending bool) {
-	t.topologyInitPending.Store(pending)
 	if pending {
+		t.topologyInitPending.Store(true)
 		return
 	}
+	// The stash is drained and the gate released inside one critical section:
+	// a mismatching report checked here, or still in flight under the same
+	// lock, lands its freeze before any allocation can observe the open gate.
 	t.reportedTopologyIdsLock.Lock()
+	defer t.reportedTopologyIdsLock.Unlock()
 	reported := t.pendingReportedTopologyIds
 	t.pendingReportedTopologyIds = nil
-	t.reportedTopologyIdsLock.Unlock()
 	for id := range reported {
-		t.CheckReportedTopologyId(id)
+		t.checkReportedTopologyIdLocked(id)
+	}
+	t.topologyInitPending.Store(false)
+}
+
+// ReleaseAllocationOnFirstReport opts the topology into lifting the post-mint
+// allocation hold once one full volume-server inventory has been applied, and
+// pins the admitted volume-server reporter: the early release is only sound
+// when a single embedded server (weed mini) can carry prior cluster state, so
+// every other reporter must be refused rather than silently trusted.
+func (t *Topology) ReleaseAllocationOnFirstReport(onlyReporter ...string) {
+	if len(onlyReporter) > 0 {
+		t.singleReporter.Store(onlyReporter[0])
+	}
+	t.releaseAllocOnReport.Store(true)
+}
+
+// AdmitVolumeReporter enforces the single-reporter restriction the early
+// release relies on: a late-arriving foreign server would carry cluster state
+// the dropped hold was meant to wait for.
+func (t *Topology) AdmitVolumeReporter(addr string) bool {
+	if !t.releaseAllocOnReport.Load() {
+		return true
+	}
+	expected, _ := t.singleReporter.Load().(string)
+	return expected == "" || expected == addr
+}
+
+// NoteInventoryApplied marks a heartbeat's full volume list as applied to this
+// topology. Under ReleaseAllocationOnFirstReport it also lifts the post-mint
+// allocation hold: the bound raise and topology-id freeze checks run while the
+// heartbeat is processed, so nothing protected remains pending afterwards.
+func (t *Topology) NoteInventoryApplied() {
+	t.inventoryApplied.Store(true)
+	if t.releaseAllocOnReport.Load() {
+		t.allocationNotBefore.Store(0)
 	}
 }
 
@@ -1586,9 +1644,19 @@ func (t *Topology) SetTopologyInitPending(pending bool) {
 // mints a fresh topology id: on a rebuilt master that window gives veteran
 // servers a pulse to raise the bound or trip the freeze.
 func (t *Topology) DeferAllocation(d time.Duration) {
+	// Under release-on-report a hold is pointless once inventory has landed:
+	// the veteran heartbeat that mattered has already been applied.
+	if t.releaseAllocOnReport.Load() && t.inventoryApplied.Load() {
+		return
+	}
 	now := time.Now()
 	t.allocationNotBefore.Store(now.Add(d).UnixNano())
 	t.allocationEvidenceSince.Store(now.UnixNano())
+	// Close the race with a report applied between the check above and the
+	// store, which would otherwise leave the full window armed.
+	if t.releaseAllocOnReport.Load() && t.inventoryApplied.Load() {
+		t.allocationNotBefore.Store(0)
+	}
 }
 
 // NoteAllocationEvidence records a heartbeat signal that could still carry