diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 9c8153fb54..46b5ad4690 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -55,9 +55,19 @@ pub async fn run_heartbeat_with_state( config.master_addresses ); + if config.master_addresses.is_empty() { + warn!("Volume server has no master addresses to heartbeat to; heartbeats disabled"); + state.is_heartbeating.store(false, Ordering::Relaxed); + return; + } + let pulse = Duration::from_secs(config.pulse_seconds.max(1)); let mut new_leader: Option = None; let mut duplicate_retry_count: u32 = 0; + // Whether a heartbeat stream was ever answered by a leader; before first + // registration the retry loop reconnects promptly instead of pacing a + // full pulse. Mirrors Go's `confirmedLeaderOnce`. + let mut confirmed_leader_once = false; loop { for master_addr in &config.master_addresses { @@ -100,6 +110,7 @@ pub async fn run_heartbeat_with_state( &target_addr, pulse, &mut shutdown_rx, + &mut confirmed_leader_once, ) .await { @@ -159,7 +170,18 @@ pub async fn run_heartbeat_with_state( // adding to it. let retry_at = tokio::time::Instant::now() + pulse; refresh_leader_from_peers(&config, &state, pulse).await; - tokio::time::sleep_until(retry_at).await; + let mut wait = retry_at.saturating_duration_since(tokio::time::Instant::now()); + if wait > Duration::from_secs(1) + && (!state.current_master().await.is_empty() || !confirmed_leader_once) + { + // A leader just named by the refresh usually accepts + // the next stream right away, and an unregistered + // server has nothing else to do, so reconnect + // promptly. Once the server was registered and no + // leader is known, keep the pulse backoff. + wait = Duration::from_secs(1); + } + tokio::time::sleep(wait).await; } } @@ -170,19 +192,9 @@ pub async fn run_heartbeat_with_state( } } - // If we have a leader redirect, skip the sleep and reconnect immediately - if new_leader.is_some() { - continue; - } - - tokio::select! { - _ = tokio::time::sleep(pulse) => {} - _ = shutdown_rx.recv() => { - state.is_heartbeating.store(false, Ordering::Relaxed); - info!("Heartbeat shutting down"); - return; - } - } + // Pacing lives in the per-attempt error path, as in Go: a failed try + // already slept its retry delay, so a round-losing pass over the seed + // list restarts right away. } } @@ -447,6 +459,7 @@ async fn do_heartbeat( current_master: &str, pulse: Duration, shutdown_rx: &mut broadcast::Receiver<()>, + confirmed_leader_once: &mut bool, ) -> Result, Box> { let channel = connect_channel( grpc_addr, @@ -516,6 +529,7 @@ async fn do_heartbeat( // naming a leader, so this peer is the one lookups // go to. confirmed_leader = true; + *confirmed_leader_once = true; publish_current_master(state, current_master).await; } if !hb_resp.duplicated_uuids.is_empty() { diff --git a/weed/command/filer.go b/weed/command/filer.go index 630ab46b5b..1cf033f55a 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -108,6 +108,15 @@ type FilerOptions struct { // gracefulStopTimeout caps how long startFiler waits for gRPC graceful // stop before forcing the server to stop. Zero means the default of 15s. gracefulStopTimeout time.Duration + // skipEmptyMembershipConfirm forwards to FilerOption: a single-master + // weed mini's embedded filer announces on the startup scan instead of + // holding registration for the confirm window. + skipEmptyMembershipConfirm bool + // clusterFreshHint, when non-nil, reports that the embedded master minted + // a fresh TopologyId this boot — proof the peer scan's empty result is + // final rather than a settling master, so the announce gate can end the + // empty-membership confirm early. Used by weed server; nil elsewhere. + clusterFreshHint func() bool } func init() { @@ -410,32 +419,34 @@ func (fo *FilerOptions) startFiler() { announce := make(chan struct{}) fs, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux, &weed_server.FilerOption{ - Masters: fo.masters, - AnnounceCh: announce, - FilerGroup: *fo.filerGroup, - Collection: *fo.collection, - DefaultReplication: *fo.defaultReplicaPlacement, - DisableDirListing: *fo.disableDirListing, - MaxMB: *fo.maxMB, - DirListingLimit: *fo.dirListingLimit, - DataCenter: *fo.dataCenter, - Rack: *fo.rack, - DefaultLevelDbDir: defaultLevelDbDirectory, - DisableHttp: *fo.disableHttp, - Host: filerAddress, - Cipher: *fo.cipher, - SaveToFilerLimit: int64(*fo.saveToFilerLimit), - ConcurrentUploadLimit: int64(*fo.concurrentUploadLimitMB) * 1024 * 1024, - ConcurrentFileUploadLimit: int64(*fo.concurrentFileUploadLimit), - ShowUIDirectoryDelete: *fo.showUIDirectoryDelete, - DownloadMaxBytesPs: int64(*fo.downloadMaxMBps) * 1024 * 1024, - DiskType: *fo.diskType, - AllowedOrigins: strings.Split(*fo.allowedOrigins, ","), - ExposeDirectoryData: *fo.exposeDirectoryData, - TusBasePath: *fo.tusBasePath, - TusMaxSize: int64(*fo.tusMaxSizeMB) * 1024 * 1024, - TusSessionExpiry: *fo.tusSessionExpiry, - CredentialManager: credentialManager, + Masters: fo.masters, + AnnounceCh: announce, + SkipEmptyMembershipConfirm: fo.skipEmptyMembershipConfirm, + ClusterFreshHint: fo.clusterFreshHint, + FilerGroup: *fo.filerGroup, + Collection: *fo.collection, + DefaultReplication: *fo.defaultReplicaPlacement, + DisableDirListing: *fo.disableDirListing, + MaxMB: *fo.maxMB, + DirListingLimit: *fo.dirListingLimit, + DataCenter: *fo.dataCenter, + Rack: *fo.rack, + DefaultLevelDbDir: defaultLevelDbDirectory, + DisableHttp: *fo.disableHttp, + Host: filerAddress, + Cipher: *fo.cipher, + SaveToFilerLimit: int64(*fo.saveToFilerLimit), + ConcurrentUploadLimit: int64(*fo.concurrentUploadLimitMB) * 1024 * 1024, + ConcurrentFileUploadLimit: int64(*fo.concurrentFileUploadLimit), + ShowUIDirectoryDelete: *fo.showUIDirectoryDelete, + DownloadMaxBytesPs: int64(*fo.downloadMaxMBps) * 1024 * 1024, + DiskType: *fo.diskType, + AllowedOrigins: strings.Split(*fo.allowedOrigins, ","), + ExposeDirectoryData: *fo.exposeDirectoryData, + TusBasePath: *fo.tusBasePath, + TusMaxSize: int64(*fo.tusMaxSizeMB) * 1024 * 1024, + TusSessionExpiry: *fo.tusSessionExpiry, + CredentialManager: credentialManager, AllowUntrustedRemoteEndpoints: *fo.allowUntrustedRemoteEndpoints, RemoteCacheEvictThreshold: *fo.remoteCacheEvictThreshold, diff --git a/weed/command/master.go b/weed/command/master.go index 93db5be139..fdc946ed58 100644 --- a/weed/command/master.go +++ b/weed/command/master.go @@ -21,6 +21,7 @@ import ( hashicorpRaft "github.com/hashicorp/raft" "slices" + "sync/atomic" "github.com/gorilla/mux" "github.com/seaweedfs/raft/protobuf" @@ -90,10 +91,16 @@ type MasterOptions struct { // 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. + // Standalone weed master can face late-arriving veterans and must keep + // the window; a single-master weed server releases on its embedded + // volume server's clean report through expectedReporterAddr instead. releaseAllocOnFirstReport bool singleReporterAddr string + // expectedReporterAddr names the reporter whose first clean inventory + // collapses the post-mint allocation hold to a confirm pulse without + // refusing others: a single-master weed server points it at its + // embedded volume server. + expectedReporterAddr string } func init() { @@ -197,7 +204,7 @@ func runMaster(cmd *Command, args []string) bool { return true } -func startMaster(masterOption MasterOptions, masterWhiteList []string) { +func startMaster(masterOption MasterOptions, masterWhiteList []string, masterOut ...*atomic.Pointer[weed_server.MasterServer]) { backend.LoadConfiguration(util.GetViper()) @@ -225,6 +232,12 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) { r := mux.NewRouter() ms := weed_server.NewMasterServer(r, masterOption.toMasterOption(masterWhiteList), masterPeers) + // Embedded setups (weed server) hand the instance back so sibling + // components can read cluster-state signals such as a fresh TopologyId + // mint. + if len(masterOut) > 0 && masterOut[0] != nil { + masterOut[0].Store(ms) + } listeningAddress := util.JoinHostPort(*masterOption.ipBind, *masterOption.port) glog.V(0).Infof("Start Seaweed Master %s at %s", version.Version(), listeningAddress) masterListener, masterLocalListener, e := util.NewIpAndLocalListeners(*masterOption.ipBind, *masterOption.port, 0, listenOnLocalhost) @@ -595,5 +608,6 @@ func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOp SingleMaster: isSingleMasterMode(*m.peers), ReleaseAllocOnFirstReport: m.releaseAllocOnFirstReport, SingleVolumeReporter: m.singleReporterAddr, + ExpectedReporterAddr: m.expectedReporterAddr, } } diff --git a/weed/command/mini.go b/weed/command/mini.go index d0fb243514..498635d302 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -1579,6 +1579,7 @@ func runMini(cmd *Command, args []string) bool { singleMaster := isSingleMasterMode(*miniMasterOptions.peers) miniMasterOptions.releaseAllocOnFirstReport = singleMaster miniMasterOptions.singleReporterAddr = "" + miniFilerOptions.skipEmptyMembershipConfirm = singleMaster if singleMaster { miniMasterOptions.singleReporterAddr = util.JoinHostPort(*miniIp, *miniOptions.v.port) } diff --git a/weed/command/server.go b/weed/command/server.go index 818364859f..d1af84da6e 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -7,10 +7,12 @@ import ( "net/http" "os" "strings" + "sync/atomic" "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" + weed_server "github.com/seaweedfs/seaweedfs/weed/server" stats_collect "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/util" flag "github.com/seaweedfs/seaweedfs/weed/util/fla9" @@ -298,10 +300,22 @@ func runServer(cmd *Command, args []string) bool { } var actualPeersForComponents string + var embeddedMaster atomic.Pointer[weed_server.MasterServer] + var peerList []pb.ServerAddress if *isStartingMasterServer { // If we are starting a master, validate and complete the peer list - _, peerList := checkPeers(*serverIp, *masterOptions.port, *masterOptions.portGrpc, *masterOptions.peers) + _, peerList = checkPeers(*serverIp, *masterOptions.port, *masterOptions.portGrpc, *masterOptions.peers) actualPeersForComponents = strings.Join(pb.ToAddressStrings(peerList), ",") + if len(peerList) == 1 && *isStartingVolumeServer { + // A single-master weed server has one guaranteed reporter: its + // embedded volume server, whose first heartbeat already carries the + // full inventory. Its clean report collapses the post-mint + // allocation hold to a confirm pulse instead of running the whole + // warmup window on every fresh start. Other volume servers stay + // admitted, so a late veteran still lands its prior-state evidence + // inside that pulse. + masterOptions.expectedReporterAddr = util.JoinHostPort(*serverIp, *serverOptions.v.port) + } } else if *masterOptions.peers != "" { if isSingleMasterMode(*masterOptions.peers) { glog.Fatalf("'-master.peers=none' is only valid when starting a master server, but master is not starting.") @@ -442,6 +456,17 @@ func runServer(cmd *Command, args []string) bool { if *filerOptions.diskType == "" && *serverOptions.v.diskType != "" { filerOptions.diskType = serverOptions.v.diskType } + if len(peerList) == 1 && *isStartingMasterServer { + // The embedded filer shares the master's process: when the master + // mints a fresh TopologyId its persisted state was empty, so the + // empty peer scan only has to survive a reconnecting peer's + // keepalive retries, not the full membership confirm. Prior state + // keeps the full window. + filerOptions.clusterFreshHint = func() bool { + ms := embeddedMaster.Load() + return ms != nil && ms.Topo.FreshMinted() + } + } go func() { time.Sleep(1 * time.Second) filerOptions.startFiler() @@ -507,7 +532,7 @@ func runServer(cmd *Command, args []string) bool { } if *isStartingMasterServer { - go startMaster(masterOptions, serverWhiteList) + go startMaster(masterOptions, serverWhiteList, &embeddedMaster) } select {} diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 8e09191723..3f8b0a9e95 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -95,6 +95,19 @@ type FilerOption struct { // master; it joins the filer list only once the gRPC port is serving and // the metadata replay has caught up with the peers found at startup. AnnounceCh chan struct{} + // SkipEmptyMembershipConfirm announces after the startup peer scan + // without re-polling to disprove an empty result: the embedded filer of + // a single-master weed mini is the only member its dev cluster can have + // yet, so the confirm window is dead time while routed traffic waits on + // the filer's master listing. A late external filer still joins and + // replays through the normal path. + SkipEmptyMembershipConfirm bool + // ClusterFreshHint, when non-nil, reports that the co-located master + // minted its TopologyId this boot — its persisted state was empty, so the + // empty-scan confirm only needs to cover a reconnecting peer's keepalive + // retries rather than the full settling window. Restarted masters + // recovering a prior id keep the full window. + ClusterFreshHint func() bool // AllowUntrustedRemoteEndpoints lets a read of a remote-only entry dial a // mounted endpoint that resolves to a loopback / private / metadata host. AllowUntrustedRemoteEndpoints bool @@ -511,7 +524,7 @@ func (fs *FilerServer) AnnounceOnceCaughtUp(announce chan struct{}, waitStart ti // late peer cannot stretch the registration hold past the cap. deadline := time.Now().Add(announceCatchUpCap) peers := fs.startupPeers - if fs.peerBootstrapPending { + if fs.peerBootstrapPending && !fs.option.SkipEmptyMembershipConfirm { peers = append(peers, fs.awaitPeerBootstrap(deadline)...) } if len(peers) == 0 { @@ -543,6 +556,13 @@ func (fs *FilerServer) AnnounceOnceCaughtUp(announce chan struct{}, waitStart ti // not a stalled startup. const emptyMembershipConfirmTimeout = 10 * time.Second +// freshMembershipConfirmTimeout bounds the same hold when the co-located +// master minted its TopologyId this boot: a fresh mint proves the master kept +// no cluster state, but not that no external filer survived its loss — one +// may still be mid-reconnect. A few keepalive cycles of empty scans cover the +// reconnect while keeping a genuinely fresh cluster's announce prompt. +const freshMembershipConfirmTimeout = 3 * time.Second + // awaitPeerBootstrap re-polls membership while a fresh store's empty startup // scan can still be disproved: masters restarted or re-elected report no // filers until the existing ones reconnect. Peers found late are backfilled @@ -551,6 +571,8 @@ const emptyMembershipConfirmTimeout = 10 * time.Second // metadata copy — is bounded by the shared announce deadline. func (fs *FilerServer) awaitPeerBootstrap(deadline time.Time) (peers []pb.ServerAddress) { confirmBy := time.Now().Add(emptyMembershipConfirmTimeout) + var freshSince time.Time + peerSeen := false for time.Now().Before(deadline) { ctx, cancel := context.WithDeadline(context.Background(), deadline) nodes, listErr := fs.filer.ListExistingPeerUpdates(ctx) @@ -566,12 +588,28 @@ func (fs *FilerServer) awaitPeerBootstrap(deadline time.Time) (peers []pb.Server } } if len(remotes) == 0 { - if !time.Now().Before(confirmBy) { + exitBy := confirmBy + // The mint lands after raft election, which can postdate this + // wait's start, so the hint is re-evaluated each pass rather than + // once up front. An observed peer — even one that later + // disconnects — disables the shortened deadline: an empty scan + // afterward no longer proves a fresh cluster. + if !peerSeen && fs.option.ClusterFreshHint != nil && fs.option.ClusterFreshHint() { + if freshSince.IsZero() { + freshSince = time.Now() + glog.V(0).Infof("%s embedded master minted a fresh topology id; shortening the empty-scan confirm", fs.option.Host) + } + if freshBy := freshSince.Add(freshMembershipConfirmTimeout); freshBy.Before(exitBy) { + exitBy = freshBy + } + } + if !time.Now().Before(exitBy) { return nil } time.Sleep(time.Second) continue } + peerSeen = true bootstrapFrom := time.Now().Add(-filer.LogFlushInterval) ctx, cancel = context.WithDeadline(context.Background(), deadline) err := fs.filer.MaybeBootstrapFromOnePeer(ctx, fs.option.Host, remotes, bootstrapFrom) diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 2df9779a93..9893b75457 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -376,11 +376,11 @@ 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. + // prior cluster state only owes other reporters a confirm pulse. 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() + ms.Topo.NoteInventoryApplied(dn.Url()) } if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 { diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 576df80fdd..cefed50232 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -77,6 +77,10 @@ type MasterOption struct { // server trying to register is rejected. ReleaseAllocOnFirstReport bool SingleVolumeReporter string + // ExpectedReporterAddr names the one volume server whose first clean full + // inventory lifts the post-mint allocation hold (a single-master weed + // server's embedded volume server). Other reporters are still admitted. + ExpectedReporterAddr string } type MasterServer struct { @@ -192,6 +196,9 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se if option.ReleaseAllocOnFirstReport { ms.Topo.ReleaseAllocationOnFirstReport(option.SingleVolumeReporter) } + if option.ExpectedReporterAddr != "" { + ms.Topo.ReleaseAllocationOnReport(option.ExpectedReporterAddr) + } 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) diff --git a/weed/server/raft_common.go b/weed/server/raft_common.go index 50dd647254..28925e50cd 100644 --- a/weed/server/raft_common.go +++ b/weed/server/raft_common.go @@ -65,6 +65,11 @@ func EnsureTopologyId(topo *topology.Topology, checkLeaderFn func() bool, persis } else { glog.V(0).Infof("TopologyId generated: %s", topologyId) } + // A recovered bound means cluster state survived without a + // TopologyId, so the mint is not proof of an empty cluster. + if topo.GetMaxVolumeId() == 0 { + topo.MarkFreshMint() + } topo.DeferAllocation(topo.AllocationDeferralDuration()) break } diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index 639bb5ef23..5237cf2b21 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -128,6 +128,11 @@ func (vs *VolumeServer) heartbeat() { vs.store.SetDataCenter(vs.dataCenter) vs.store.SetRack(vs.rack) + if len(vs.SeedMasterNodes) == 0 { + glog.Warningf("Volume server has no master nodes to heartbeat to; heartbeats disabled") + return + } + grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.volume") var err error @@ -158,7 +163,16 @@ func (vs *VolumeServer) heartbeat() { // adding to it. retryAt := time.Now().Add(vs.pulsePeriod) vs.refreshLeaderFromPeers(vs.pulsePeriod) - time.Sleep(time.Until(retryAt)) + wait := time.Until(retryAt) + if wait > time.Second && (vs.getCurrentMaster() != "" || !vs.confirmedLeaderOnce.Load()) { + // A leader just named by the refresh usually accepts the + // next stream right away, and an unregistered server has + // nothing else to do, so reconnect promptly. Once the + // server was registered and no leader is known, keep the + // pulse backoff. + wait = time.Second + } + time.Sleep(wait) } stats.VolumeServerMasterDisconnections.WithLabelValues(master.String()).Inc() @@ -235,6 +249,7 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp // The leader answers every new heartbeat stream without naming a // leader, so this peer is the one lookups should go to. confirmedLeader = true + vs.confirmedLeaderOnce.Store(true) vs.setCurrentMaster(masterAddress) } if len(in.DuplicatedUuids) > 0 { diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go index 93351eca44..e4814f6752 100644 --- a/weed/server/volume_server.go +++ b/weed/server/volume_server.go @@ -42,6 +42,10 @@ type VolumeServer struct { whiteList []string currentMaster pb.ServerAddress currentMasterLock sync.RWMutex + // confirmedLeaderOnce records that a heartbeat stream was ever answered + // by a leader; before first registration the retry loop reconnects + // promptly instead of pacing a full pulse. + confirmedLeaderOnce atomic.Bool pulsePeriod time.Duration dataCenter string rack string diff --git a/weed/topology/growth_freeze_test.go b/weed/topology/growth_freeze_test.go index 13d849e4e8..65e6e5812c 100644 --- a/weed/topology/growth_freeze_test.go +++ b/weed/topology/growth_freeze_test.go @@ -194,17 +194,17 @@ func TestAllocationDeferralDuration(t *testing.T) { func TestAllocationDeferralReleasedByInventory(t *testing.T) { topo := NewTopology("topo", nil, 128, 3, false) - topo.ReleaseAllocationOnFirstReport() + topo.ReleaseAllocationOnFirstReport("10.0.0.1:8080") topo.DeferAllocation(time.Minute) - topo.NoteInventoryApplied() + topo.NoteInventoryApplied("10.0.0.1:8080") 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.ReleaseAllocationOnFirstReport("10.0.0.1:8080") + fresh.NoteInventoryApplied("10.0.0.1:8080") 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) @@ -212,12 +212,49 @@ func TestAllocationDeferralReleasedByInventory(t *testing.T) { walled := NewTopology("topo", nil, 128, 3, false) walled.DeferAllocation(time.Minute) - walled.NoteInventoryApplied() + walled.NoteInventoryApplied("10.0.0.1:8080") if wait := walled.allocationDelay(); wait <= 0 { t.Fatal("without opt-in an applied inventory must not lift the deferral") } } +func TestAllocationDeferralReleasedByExpectedReporter(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + topo.ReleaseAllocationOnReport("10.0.0.1:8080") + + if !topo.AdmitVolumeReporter("10.0.0.2:8080") { + t.Fatal("a named expected reporter must not refuse other reporters") + } + topo.DeferAllocation(time.Minute) + topo.NoteInventoryApplied("10.0.0.2:8080") + if wait := topo.allocationDelay(); wait <= 0 { + t.Fatal("a foreign reporter's inventory must not lift the deferral") + } + topo.NoteInventoryApplied("10.0.0.1:8080") + wait := topo.allocationDelay() + if wait <= 0 || wait > 3*time.Second { + t.Fatalf("the expected reporter's inventory should collapse the deferral to a confirm pulse, got %v", wait) + } +} + +func TestAllocationDeferralConfirmPulseAfterAppliedReport(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + topo.ReleaseAllocationOnReport("10.0.0.1:8080") + + topo.DeferAllocation(time.Minute) + topo.NoteInventoryApplied("10.0.0.1:8080") + if wait := topo.allocationDelay(); wait <= 0 || wait > 3*time.Second { + t.Fatalf("an unpinned release should leave a confirm pulse, got %v", wait) + } + + // A later mint re-arms the confirm pulse rather than the full warmup or + // nothing at all: outside reporters may still carry prior cluster state. + topo.DeferAllocation(time.Minute) + if wait := topo.allocationDelay(); wait <= 0 || wait > 3*time.Second { + t.Fatalf("a re-mint after the applied report should arm the confirm pulse, got %v", wait) + } +} + func TestQueuedTopologyReportFreezesBeforeGateOpens(t *testing.T) { topo := NewTopology("topo", nil, 128, 3, false) topo.SetGrowthFreezeConfig(true, 0) diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 8811ba6043..af73eb0cef 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -101,15 +101,24 @@ type Topology struct { allocationEvidenceSince atomic.Int64 allocationDeferral atomic.Int64 topologyInitPending atomic.Bool + // freshMinted marks that this boot minted the TopologyId instead of + // recovering one: the master's persisted state was empty, so no cluster + // member registered under a prior control-plane instance. + freshMinted 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. + // volume-server inventory has been applied, for setups with a known + // reporter (weed mini, single-master weed server) whose embedded volume + // server reports the full inventory in its first heartbeat. singleReporter + // additionally refuses every other reporter, so its report drops the hold + // outright; expectedReporter only gates which report may lift the hold, + // and the lift leaves a confirm pulse for reporters still admitted. + // inventoryApplied records that the expected report has been synced into + // this topology. releaseAllocOnReport atomic.Bool inventoryApplied atomic.Bool singleReporter atomic.Value // string + expectedReporter atomic.Value // string reportedTopologyIdsLock sync.Mutex pendingReportedTopologyIds map[string]struct{} @@ -1601,6 +1610,18 @@ func (t *Topology) applyGrowthFreeze(frozen bool) { } } +// MarkFreshMint records that this boot minted the TopologyId rather than +// recovering a persisted one — the strongest in-process proof that the +// control plane holds no prior cluster state. +func (t *Topology) MarkFreshMint() { + t.freshMinted.Store(true) +} + +// FreshMinted reports whether the current TopologyId was minted this boot. +func (t *Topology) FreshMinted() bool { + return t.freshMinted.Load() +} + // 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 @@ -1632,11 +1653,24 @@ func (t *Topology) SetTopologyInitPending(pending bool) { func (t *Topology) ReleaseAllocationOnFirstReport(onlyReporter ...string) { if len(onlyReporter) > 0 { t.singleReporter.Store(onlyReporter[0]) + t.expectedReporter.Store(onlyReporter[0]) } t.releaseAllocOnReport.Store(true) } -// AdmitVolumeReporter enforces the single-reporter restriction the early +// ReleaseAllocationOnReport lifts the post-mint allocation hold once the named +// reporter's first clean full inventory has been applied. Unlike +// ReleaseAllocationOnFirstReport it does not refuse other reporters: a +// single-master weed server still admits outside volume servers, so the +// release collapses the hold to a confirm pulse rather than clearing it — +// outside reporters keep a heartbeat interval to land prior-state evidence +// or a freeze before new ids issue. +func (t *Topology) ReleaseAllocationOnReport(expected string) { + t.expectedReporter.Store(expected) + t.releaseAllocOnReport.Store(true) +} + +// AdmitVolumeReporter enforces the single-reporter restriction the outright // 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 { @@ -1648,13 +1682,41 @@ func (t *Topology) AdmitVolumeReporter(addr string) bool { } // 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() { +// topology. Under releaseAllocOnReport it also lifts the post-mint allocation +// hold once the report comes from the expected reporter: the bound raise and +// topology-id freeze checks run while the heartbeat is processed, so nothing +// protected remains pending afterwards. +func (t *Topology) NoteInventoryApplied(reporter string) { + if expected, _ := t.expectedReporter.Load().(string); expected != "" && expected != reporter { + return + } t.inventoryApplied.Store(true) if t.releaseAllocOnReport.Load() { + t.releaseAllocationHold() + } +} + +// releaseAllocationHold ends a post-mint allocation hold on the expected +// reporter's clean inventory. A pinned single reporter is the only reporter +// admitted, so its inventory settles prior cluster state outright and the +// hold can drop. With outside reporters still admitted the release collapses +// the hold to one heartbeat interval instead: the expected report cannot +// prove no other reporter carries prior cluster state, so they keep a pulse +// to land their evidence or a freeze before new ids issue. +func (t *Topology) releaseAllocationHold() { + if pinned, _ := t.singleReporter.Load().(string); pinned != "" { t.allocationNotBefore.Store(0) + return + } + confirm := time.Now().Add(time.Duration(t.pulse) * time.Second).UnixNano() + for { + cur := t.allocationNotBefore.Load() + if cur <= confirm { + return + } + if t.allocationNotBefore.CompareAndSwap(cur, confirm) { + return + } } } @@ -1662,10 +1724,17 @@ func (t *Topology) NoteInventoryApplied() { // 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. + // Under release-on-report a full hold is pointless once inventory has + // landed: the veteran heartbeat that mattered has already been applied. + // Outside reporters stay admitted without a pinned single reporter, so + // the hold caps at one confirm pulse rather than disappearing. if t.releaseAllocOnReport.Load() && t.inventoryApplied.Load() { - return + if pinned, _ := t.singleReporter.Load().(string); pinned != "" { + return + } + if pulse := time.Duration(t.pulse) * time.Second; d > pulse { + d = pulse + } } now := time.Now() t.allocationNotBefore.Store(now.Add(d).UnixNano()) @@ -1673,7 +1742,7 @@ func (t *Topology) DeferAllocation(d time.Duration) { // 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) + t.releaseAllocationHold() } }