peer chunk sharing 4/8: mount registrar + HRW owner selection (#9133)

* proto: define MountRegister/MountList and MountPeer service

Adds the wire types for peer chunk sharing between weed mount clients:

* filer.proto: MountRegister / MountList RPCs so each mount can heartbeat
  its peer-serve address into a filer-hosted registry, and refresh the
  list of peers. Tiny payload; the filer stores only O(fleet_size) state.

* mount_peer.proto (new): ChunkAnnounce / ChunkLookup RPCs for the
  mount-to-mount chunk directory. Each fid's directory entry lives on
  an HRW-assigned mount; announces and lookups route to that mount.

No behavior yet — later PRs wire the RPCs into the filer and mount.
See design-weed-mount-peer-chunk-sharing.md for the full design.

* filer: add mount-server registry behind -peer.registry.enable

Implements tier 1 of the peer chunk sharing design: an in-memory registry
of live weed mount servers, keyed by peer address, refreshed by
MountRegister heartbeats and served by MountList.

* weed/filer/peer_registry.go: thread-safe map with TTL eviction; lazy
  sweep on List plus a background sweeper goroutine for bounded memory.

* weed/server/filer_grpc_server_peer.go: MountRegister / MountList RPC
  handlers. When -peer.registry.enable is false (the default), both RPCs
  are silent no-ops so probing older filers is harmless.

* -peer.registry.enable flag on weed filer; FilerOption.PeerRegistryEnabled
  wires it through.

Phase 1 is single-filer (no cross-filer replication of the registry);
mounts that fail over to another filer will re-register on the next
heartbeat, so the registry self-heals within one TTL cycle.

Part of the peer-chunk-sharing design; no behavior change at runtime
until a later PR enables the flag on both filer and mount.

* filer: nil-safe peerRegistryEnable + registry hardening

Addresses review feedback on PR #9131.

* Fix: nil pointer deref in the mini cluster. FilerOptions instances
  constructed outside weed/command/filer.go (e.g. miniFilerOptions in
  mini.go) do not populate peerRegistryEnable, so dereferencing the
  pointer panics at Filer startup. Use the same
  `nil && deref` idiom already used for distributedLock / writebackCache.

* Hardening (gemini review): registry now enforces three invariants:
  - empty peer_addr is silently rejected (no client-controlled sentinel
    mass-inserts)
  - TTL is capped at 1 hour so a runaway client cannot pin entries
  - new-entry count is capped at 10000 to bound memory; renewals of
    existing entries are always honored, so a full registry still
    heartbeats its existing members correctly

Covered by new unit tests.

* filer: rename -peer.registry.enable flag to -mount.p2p

Per review feedback: the old name "peer.registry.enable" leaked
the implementation ("registry") into the CLI surface. "mount.p2p"
is shorter and describes what it actually controls — whether this
filer participates in mount-to-mount peer chunk sharing.

Flag renames (all three keep default=true, idle cost is near-zero):
  -peer.registry.enable        ->  -mount.p2p         (weed filer)
  -filer.peer.registry.enable  ->  -filer.mount.p2p   (weed mini, weed server)

Internal variable names (mountPeerRegistryEnable, MountPeerRegistry)
keep their longer form — they describe the component, not the knob.

* filer: MountList returns DataCenter + List uses RLock

Two review follow-ups on the mount peer registry:

* weed/server/filer_grpc_server_mount_peer.go: MountList was dropping
  the DataCenter on the wire. The whole point of carrying DC separately
  from Rack is letting the mount-side fetcher re-rank peers by the
  two-level locality hierarchy (same-rack > same-DC > cross-DC); without
  DC in the response every remote peer collapsed to "unknown locality."

* weed/filer/mount_peer_registry.go: List() was taking a write lock so
  it could lazy-delete expired entries inline. But MountList is a
  read-heavy RPC hit on every mount's 30 s refresh loop, and Sweep is
  already wired as the sole reclamation path (same pattern as the
  mount-side PeerDirectory). Switch List to RLock + filter, let Sweep
  do the map mutation, so concurrent MountList callers don't serialize
  on each other.

Test updated to reflect the new contract (List no longer mutates the
map; Sweep is what drops expired entries).

* mount: add peer chunk sharing options + advertise address resolver

First cut at the peer chunk sharing wiring on the mount side. No
functional behavior yet — this PR just introduces the option fields,
the -peer.* flags, and the helper that resolves a reachable
host:port from them. The server implementation arrives in PR #5
(gRPC service) and the fetcher in PR #7.

* ResolvePeerAdvertiseAddr: an explicit -peer.advertise wins; else we
  use -peer.listen's bind host if specific; else util.DetectedHostAddress
  combined with the port. This is what gets registered with the filer
  and announced to peers, so wildcard binds no longer result in
  unreachable identities like "[::]:18080".

* Option fields: PeerEnabled, PeerListen, PeerAdvertise, PeerRack.
  One port handles both directory RPCs and streaming chunk fetches
  (see PR #1 FetchChunk proto), so there is no second -peer.grpc.*
  flag — the old HTTP byte-transfer path is gone.

* New flags on weed mount: -peer.enable, -peer.listen (default :18080),
  -peer.advertise (default auto), -peer.rack.

* mount: register with filer and maintain HRW seed view

Adds the mount-side tier-1 client. On startup the mount calls
MountRegister with its advertise address (PR #3) and keeps both the
filer entry and the local seed view fresh via background tickers
(30 s register / 30 s list, 90 s filer TTL).

* peer_hrw.go: pure rendezvous-hashing helper picking a single owner
  per fid via top-1 HRW. Adding or removing one seed moves only
  ~1/N fids.

* peer_registrar.go: heartbeat + list poller. Seeds() returns the
  slice directly (no per-call copy) since listOnce atomically swaps;
  background RPCs bind their context to Stop() so unmount doesn't
  hang on a slow filer.

* WFS wiring uses ResolvePeerAdvertiseAddr from PR #3 for the
  identity registered with the filer. No HTTP server, no second
  port — one reachable address represents the mount.

* mount: broadcast MountRegister/MountList to every filer

Previously the registrar called through wfs.WithFilerClient, which only
reaches whichever filer the WFS filer-client session happens to be on.
That meant two mounts pointing at different filers would never see each
other: the filer mount registries are in-memory and per-filer (no
filer-to-filer sync), so each mount's MountList only returned peers
that had also registered through the same filer.

This commit makes the registrar multi-filer aware:

  * NewPeerRegistrar now takes the full FilerAddresses slice and a
    per-filer dial function. The old single-filer peerFilerClient
    interface is gone.

  * registerOnce fans a MountRegister RPC out to every filer in
    parallel. Succeeds if at least one filer accepted — an unreachable
    filer is tolerated, logged, and retried on the next heartbeat.

  * listOnce polls every filer's MountList in parallel and merges the
    responses by peer_addr, keeping the newest LastSeenNs on duplicates.
    Mounts talking to different filers therefore converge once every
    filer has been polled once.

The merged-list property is what lets a fleet of mounts spread across
multiple filers still form a single HRW seed view. Each filer only ever
sees the subset of mounts that heartbeat through it, but the registrar
reconstructs the union client-side.

New unit tests guard both properties:
  - RegisterBroadcastsToAllFilers: one registerOnce hits all N filers.
  - ListMergesAcrossFilers: mount-a on filer-1 and mount-b on filer-2
    both appear in the merged seed set.
  - ListMergeKeepsNewestLastSeen: the same mount reported by two
    filers collapses to one entry with the freshest timestamp.

weed/mount/peer_hrw.go

@@ -0,0 +1,51 @@
+package mount
+
+import (
+	"hash/fnv"
+)
+
+// SeedPeer is a mount server present in the filer's mount registry.
+// DataCenter and Rack ride along so downstream locality-aware decisions
+// (peer re-ranking in the fetcher) don't have to re-query the filer.
+type SeedPeer struct {
+	PeerAddr   string
+	DataCenter string
+	Rack       string
+}
+
+// OwnerFor returns the peer_addr of the mount that is the HRW-assigned
+// directory owner for fid on the given seed list.
+//
+// Rendezvous (highest-random-weight) hashing is deterministic for a given
+// (fid, seed set): every mount computes the same owner without any
+// consensus. When a mount joins or leaves the seed set, only 1/N of fids
+// change owner — the rest remain on their existing mount.
+//
+// If seeds is empty, returns "".
+//
+// See design-weed-mount-peer-chunk-sharing.md §4.2.2.
+func OwnerFor(fid string, seeds []SeedPeer) string {
+	if len(seeds) == 0 {
+		return ""
+	}
+	var bestScore uint64
+	var bestAddr string
+	for _, s := range seeds {
+		score := hrwScore(s.PeerAddr, fid)
+		if bestAddr == "" || score > bestScore || (score == bestScore && s.PeerAddr < bestAddr) {
+			bestScore = score
+			bestAddr = s.PeerAddr
+		}
+	}
+	return bestAddr
+}
+
+// hrwScore combines peer_addr and fid into a 64-bit score. Uses FNV-1a
+// (cheap and fine for balancing; not a security primitive).
+func hrwScore(peerAddr, fid string) uint64 {
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(peerAddr))
+	_, _ = h.Write([]byte{0}) // separator so "ab" + "c" != "a" + "bc"
+	_, _ = h.Write([]byte(fid))
+	return h.Sum64()
+}

weed/mount/peer_hrw_test.go

@@ -0,0 +1,90 @@
+package mount
+
+import (
+	"fmt"
+	"testing"
+)
+
+func mkSeeds(addrs ...string) []SeedPeer {
+	out := make([]SeedPeer, len(addrs))
+	for i, a := range addrs {
+		out[i] = SeedPeer{PeerAddr: a}
+	}
+	return out
+}
+
+func TestOwnerFor_EmptySeeds(t *testing.T) {
+	if got := OwnerFor("3,01637037d6", nil); got != "" {
+		t.Errorf("expected empty owner for empty seed list, got %q", got)
+	}
+}
+
+func TestOwnerFor_Deterministic(t *testing.T) {
+	seeds := mkSeeds("a:1", "b:1", "c:1", "d:1")
+	fid := "3,01637037d6"
+	first := OwnerFor(fid, seeds)
+	for i := 0; i < 100; i++ {
+		if got := OwnerFor(fid, seeds); got != first {
+			t.Fatalf("non-deterministic: iter %d got %q first %q", i, got, first)
+		}
+	}
+}
+
+func TestOwnerFor_DistributesEvenly(t *testing.T) {
+	const N = 4
+	seeds := mkSeeds("a:1", "b:1", "c:1", "d:1")
+	counts := map[string]int{}
+	for i := 0; i < 10000; i++ {
+		fid := fmt.Sprintf("%d,%x", i, i*17)
+		counts[OwnerFor(fid, seeds)]++
+	}
+	// Each seed should get roughly 25% (±10% slack for a 10k sample).
+	for addr, c := range counts {
+		ratio := float64(c) / 10000.0
+		if ratio < 0.225 || ratio > 0.275 {
+			t.Errorf("HRW distribution skewed: %s got %.3f", addr, ratio)
+		}
+	}
+	if len(counts) != N {
+		t.Errorf("expected %d distinct owners, got %d", N, len(counts))
+	}
+}
+
+func TestOwnerFor_MinimalShuffleOnSeedChange(t *testing.T) {
+	// Adding one seed should move ~1/(N+1) fids to the new seed and leave
+	// the rest on their prior owners. Tolerance generous for a 10k sample.
+	const trials = 10000
+	before := mkSeeds("a:1", "b:1", "c:1")
+	after := mkSeeds("a:1", "b:1", "c:1", "d:1")
+
+	moved := 0
+	toNewSeed := 0
+	for i := 0; i < trials; i++ {
+		fid := fmt.Sprintf("%d,%x", i, i*31)
+		pre := OwnerFor(fid, before)
+		post := OwnerFor(fid, after)
+		if pre != post {
+			moved++
+			if post == "d:1" {
+				toNewSeed++
+			}
+		}
+	}
+	// Expected: ~1/4 = 25% of fids move, all to the new seed.
+	ratio := float64(moved) / float64(trials)
+	if ratio < 0.20 || ratio > 0.30 {
+		t.Errorf("expected ~25%% fids to move on seed-add, got %.3f", ratio)
+	}
+	if moved != toNewSeed {
+		t.Errorf("expected every moved fid to land on the new seed; moved=%d toNewSeed=%d", moved, toNewSeed)
+	}
+}
+
+func TestOwnerFor_TieBreakerDeterministic(t *testing.T) {
+	// Two seeds with equal hash score (engineered collision is unlikely in
+	// practice, but we guarantee determinism by lex-comparing addresses).
+	// This test just confirms OwnerFor runs without panicking on duplicates.
+	seeds := mkSeeds("a:1", "a:1", "b:1")
+	fid := "3,01637037d6"
+	_ = OwnerFor(fid, seeds)
+}

weed/mount/peer_registrar.go

@@ -0,0 +1,251 @@
+package mount
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+// PeerRegistrar maintains this mount's presence in every configured
+// filer's mount registry and its local snapshot of the merged seed set.
+// It runs two background tickers:
+//
+//   - MountRegister heartbeats, fanned out to every filer in parallel so
+//     each filer keeps a fresh TTL entry for us. Mounts pointing at
+//     different filers therefore still see each other once all filers
+//     have been heartbeated.
+//   - MountList polling, fanned out identically, merged by peer_addr
+//     (newest LastSeenNs wins) so OwnerFor() has a view of the whole
+//     fleet regardless of which filer each peer happens to heartbeat
+//     through.
+//
+// An unreachable filer is tolerated: we log and continue as long as at
+// least one filer succeeds. Entries cached on a permanently-gone filer
+// fall out of the merged view on their own TTL.
+type PeerRegistrar struct {
+	filerAddrs       []pb.ServerAddress
+	dialFiler        filerDialFn
+	selfPeerAddr     string
+	selfDc           string
+	selfRack         string
+	registerInterval time.Duration
+	registerTTL      time.Duration
+	listInterval     time.Duration
+
+	mu    sync.RWMutex
+	seeds []SeedPeer
+
+	// stopCtx cancels when Stop() is called. Background RPCs scope their
+	// deadline to this so unmount does not block on pending filer calls.
+	stopCtx    context.Context
+	stopCancel context.CancelFunc
+	stopped    atomic.Bool
+}
+
+// filerDialFn is how the registrar reaches one configured filer. The
+// production wiring is pb.WithGrpcFilerClient; tests inject a fake.
+type filerDialFn func(ctx context.Context, addr pb.ServerAddress, fn func(client filer_pb.SeaweedFilerClient) error) error
+
+// NewPeerRegistrar constructs the registrar; Start launches the background
+// loops. Callers must supply the full filer set so heartbeats and list
+// polls reach every filer — otherwise mounts talking to different filers
+// never observe each other.
+func NewPeerRegistrar(filers []pb.ServerAddress, dial filerDialFn, selfAddr, dc, rack string) *PeerRegistrar {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &PeerRegistrar{
+		filerAddrs:       filers,
+		dialFiler:        dial,
+		selfPeerAddr:     selfAddr,
+		selfRack:         rack,
+		selfDc:           dc,
+		registerInterval: 30 * time.Second,
+		registerTTL:      90 * time.Second,
+		listInterval:     30 * time.Second,
+		stopCtx:          ctx,
+		stopCancel:       cancel,
+	}
+}
+
+// Start does an initial register+list synchronously (so OwnerFor has a
+// usable view immediately) and then kicks off the background loops.
+func (r *PeerRegistrar) Start(ctx context.Context) error {
+	if err := r.registerOnce(ctx); err != nil {
+		glog.V(1).Infof("initial MountRegister: %v", err)
+		// Do not fail startup — the mount must still serve reads even if
+		// no filer yet knows about us.
+	}
+	if err := r.listOnce(ctx); err != nil {
+		glog.V(1).Infof("initial MountList: %v", err)
+	}
+	go r.loopRegister()
+	go r.loopList()
+	return nil
+}
+
+// Stop halts the background loops and cancels any in-flight RPCs they
+// may have launched. Safe to call multiple times.
+func (r *PeerRegistrar) Stop() {
+	if r.stopped.Swap(true) {
+		return
+	}
+	r.stopCancel()
+}
+
+// Seeds returns the currently-known seed set. Callers MUST NOT mutate the
+// returned slice — it is shared with concurrent listOnce readers. Because
+// listOnce atomically swaps to a brand-new slice on every refresh rather
+// than mutating in place, returning the slice header directly is safe and
+// avoids a per-call allocation on the read-hot OwnerFor path.
+func (r *PeerRegistrar) Seeds() []SeedPeer {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.seeds
+}
+
+// OwnerFor is a convenience wrapper that runs HRW against the current
+// seed snapshot.
+func (r *PeerRegistrar) OwnerFor(fid string) string {
+	return OwnerFor(fid, r.Seeds())
+}
+
+// filerRPCTimeout bounds a single MountRegister / MountList call so a slow
+// or partitioned filer can't wedge the background loops (which run on a
+// 30 s cadence). 20 s gives room for TLS handshakes on cold connections
+// while leaving headroom before the next tick.
+const filerRPCTimeout = 20 * time.Second
+
+// registerOnce fans a MountRegister out to every configured filer in
+// parallel. Returns an error only if every filer failed; otherwise the
+// best-effort semantics let the mount proceed when some filer is down.
+func (r *PeerRegistrar) registerOnce(ctx context.Context) error {
+	if len(r.filerAddrs) == 0 {
+		return fmt.Errorf("no filers configured")
+	}
+	ctx, cancel := context.WithTimeout(ctx, filerRPCTimeout)
+	defer cancel()
+	req := &filer_pb.MountRegisterRequest{
+		PeerAddr:   r.selfPeerAddr,
+		Rack:       r.selfRack,
+		DataCenter: r.selfDc,
+		TtlSeconds: int32(r.registerTTL / time.Second),
+	}
+	var wg sync.WaitGroup
+	var successes atomic.Int32
+	for _, addr := range r.filerAddrs {
+		wg.Add(1)
+		go func(addr pb.ServerAddress) {
+			defer wg.Done()
+			err := r.dialFiler(ctx, addr, func(c filer_pb.SeaweedFilerClient) error {
+				_, err := c.MountRegister(ctx, req)
+				return err
+			})
+			if err != nil {
+				glog.V(2).Infof("MountRegister %s: %v", addr, err)
+				return
+			}
+			successes.Add(1)
+		}(addr)
+	}
+	wg.Wait()
+	if successes.Load() == 0 {
+		return fmt.Errorf("MountRegister failed on all %d filer(s)", len(r.filerAddrs))
+	}
+	return nil
+}
+
+// listOnce polls MountList from every filer in parallel and merges the
+// responses by peer_addr (newest LastSeenNs wins). This way two mounts
+// heartbeating through different filers still end up in each other's
+// seed view as soon as at least one filer has been listed on each side.
+func (r *PeerRegistrar) listOnce(ctx context.Context) error {
+	if len(r.filerAddrs) == 0 {
+		r.mu.Lock()
+		r.seeds = nil
+		r.mu.Unlock()
+		return nil
+	}
+	ctx, cancel := context.WithTimeout(ctx, filerRPCTimeout)
+	defer cancel()
+	var (
+		mu     sync.Mutex
+		merged = map[string]*filer_pb.MountInfo{}
+		fails  int
+	)
+	var wg sync.WaitGroup
+	for _, addr := range r.filerAddrs {
+		wg.Add(1)
+		go func(addr pb.ServerAddress) {
+			defer wg.Done()
+			err := r.dialFiler(ctx, addr, func(c filer_pb.SeaweedFilerClient) error {
+				resp, err := c.MountList(ctx, &filer_pb.MountListRequest{})
+				if err != nil {
+					return err
+				}
+				mu.Lock()
+				for _, m := range resp.Mounts {
+					if prev, ok := merged[m.PeerAddr]; !ok || m.LastSeenNs > prev.LastSeenNs {
+						merged[m.PeerAddr] = m
+					}
+				}
+				mu.Unlock()
+				return nil
+			})
+			if err != nil {
+				mu.Lock()
+				fails++
+				mu.Unlock()
+				glog.V(2).Infof("MountList %s: %v", addr, err)
+			}
+		}(addr)
+	}
+	wg.Wait()
+
+	if fails == len(r.filerAddrs) {
+		return fmt.Errorf("MountList failed on all %d filer(s)", len(r.filerAddrs))
+	}
+
+	next := make([]SeedPeer, 0, len(merged))
+	for _, m := range merged {
+		next = append(next, SeedPeer{PeerAddr: m.PeerAddr, DataCenter: m.DataCenter, Rack: m.Rack})
+	}
+	r.mu.Lock()
+	r.seeds = next
+	r.mu.Unlock()
+	return nil
+}
+
+func (r *PeerRegistrar) loopRegister() {
+	t := time.NewTicker(r.registerInterval)
+	defer t.Stop()
+	for {
+		select {
+		case <-r.stopCtx.Done():
+			return
+		case <-t.C:
+			if err := r.registerOnce(r.stopCtx); err != nil {
+				glog.V(2).Infof("MountRegister heartbeat: %v", err)
+			}
+		}
+	}
+}
+
+func (r *PeerRegistrar) loopList() {
+	t := time.NewTicker(r.listInterval)
+	defer t.Stop()
+	for {
+		select {
+		case <-r.stopCtx.Done():
+			return
+		case <-t.C:
+			if err := r.listOnce(r.stopCtx); err != nil {
+				glog.V(2).Infof("MountList refresh: %v", err)
+			}
+		}
+	}
+}

weed/mount/peer_registrar_test.go

@@ -0,0 +1,210 @@
+package mount
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"google.golang.org/grpc"
+)
+
+// fakeFilerClient captures MountRegister/MountList calls and lets the test
+// pre-seed MountList responses. Implements just enough of
+// filer_pb.SeaweedFilerClient to drive the registrar.
+type fakeFilerClient struct {
+	filer_pb.SeaweedFilerClient // embed for methods we don't need
+
+	mu            sync.Mutex
+	registerCalls []filer_pb.MountRegisterRequest
+	listResponse  filer_pb.MountListResponse
+}
+
+func (f *fakeFilerClient) MountRegister(ctx context.Context, req *filer_pb.MountRegisterRequest, opts ...grpc.CallOption) (*filer_pb.MountRegisterResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.registerCalls = append(f.registerCalls, *req)
+	return &filer_pb.MountRegisterResponse{}, nil
+}
+
+func (f *fakeFilerClient) MountList(ctx context.Context, req *filer_pb.MountListRequest, opts ...grpc.CallOption) (*filer_pb.MountListResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	resp := f.listResponse // value copy
+	return &resp, nil
+}
+
+// fakeFilerFleet maps each addr to its own fakeFilerClient so a test can
+// simulate several filers with different registered/listed state.
+type fakeFilerFleet struct {
+	clients map[pb.ServerAddress]*fakeFilerClient
+}
+
+func (f *fakeFilerFleet) dial(ctx context.Context, addr pb.ServerAddress, fn func(client filer_pb.SeaweedFilerClient) error) error {
+	c, ok := f.clients[addr]
+	if !ok {
+		// Treat unknown filer as "reachable but empty" — lets tests omit
+		// pre-populating a client when they don't care.
+		c = &fakeFilerClient{}
+		f.clients[addr] = c
+	}
+	return fn(c)
+}
+
+func singleFilerFleet(c *fakeFilerClient) ([]pb.ServerAddress, filerDialFn) {
+	addr := pb.ServerAddress("filer-1:18888")
+	fleet := &fakeFilerFleet{clients: map[pb.ServerAddress]*fakeFilerClient{addr: c}}
+	return []pb.ServerAddress{addr}, fleet.dial
+}
+
+func TestPeerRegistrar_StartPopulatesSeedsFromFiler(t *testing.T) {
+	fc := &fakeFilerClient{
+		listResponse: filer_pb.MountListResponse{
+			Mounts: []*filer_pb.MountInfo{
+				{PeerAddr: "mount-a:18080", Rack: "r1"},
+				{PeerAddr: "mount-b:18080", Rack: "r2"},
+			},
+		},
+	}
+	filers, dial := singleFilerFleet(fc)
+
+	r := NewPeerRegistrar(filers, dial, "self:18080", "dc1", "r1")
+	if err := r.registerOnce(context.Background()); err != nil {
+		t.Fatalf("registerOnce: %v", err)
+	}
+	if err := r.listOnce(context.Background()); err != nil {
+		t.Fatalf("listOnce: %v", err)
+	}
+
+	fc.mu.Lock()
+	if len(fc.registerCalls) != 1 {
+		t.Errorf("expected 1 register call, got %d", len(fc.registerCalls))
+	} else if fc.registerCalls[0].PeerAddr != "self:18080" {
+		t.Errorf("register sent wrong peer addr: %q", fc.registerCalls[0].PeerAddr)
+	}
+	fc.mu.Unlock()
+
+	seeds := r.Seeds()
+	if len(seeds) != 2 {
+		t.Errorf("expected 2 seeds, got %d", len(seeds))
+	}
+
+	owner := r.OwnerFor("3,01637037d6")
+	if owner != "mount-a:18080" && owner != "mount-b:18080" {
+		t.Errorf("OwnerFor returned unexpected addr: %q", owner)
+	}
+}
+
+func TestPeerRegistrar_HeartbeatTTLMatchesConfig(t *testing.T) {
+	fc := &fakeFilerClient{}
+	filers, dial := singleFilerFleet(fc)
+	r := NewPeerRegistrar(filers, dial, "self:18080", "", "")
+
+	if err := r.registerOnce(context.Background()); err != nil {
+		t.Fatalf("registerOnce: %v", err)
+	}
+
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	if len(fc.registerCalls) != 1 {
+		t.Fatalf("expected 1 register call, got %d", len(fc.registerCalls))
+	}
+	want := int32(r.registerTTL.Seconds())
+	if got := fc.registerCalls[0].TtlSeconds; got != want {
+		t.Errorf("TtlSeconds: got %d want %d", got, want)
+	}
+}
+
+func TestPeerRegistrar_StopIsIdempotent(t *testing.T) {
+	filers, dial := singleFilerFleet(&fakeFilerClient{})
+	r := NewPeerRegistrar(filers, dial, "self:18080", "", "")
+	r.Stop()
+	r.Stop() // second call must be a no-op (no panic)
+}
+
+// TestPeerRegistrar_RegisterBroadcastsToAllFilers guards the core
+// multi-filer property: a single registerOnce must hit every configured
+// filer so mounts pointing at different filers still converge.
+func TestPeerRegistrar_RegisterBroadcastsToAllFilers(t *testing.T) {
+	fc1 := &fakeFilerClient{}
+	fc2 := &fakeFilerClient{}
+	fc3 := &fakeFilerClient{}
+	a1, a2, a3 := pb.ServerAddress("f1:18888"), pb.ServerAddress("f2:18888"), pb.ServerAddress("f3:18888")
+	fleet := &fakeFilerFleet{clients: map[pb.ServerAddress]*fakeFilerClient{a1: fc1, a2: fc2, a3: fc3}}
+
+	r := NewPeerRegistrar([]pb.ServerAddress{a1, a2, a3}, fleet.dial, "self:18080", "", "")
+	if err := r.registerOnce(context.Background()); err != nil {
+		t.Fatalf("registerOnce: %v", err)
+	}
+
+	for addr, fc := range fleet.clients {
+		fc.mu.Lock()
+		if len(fc.registerCalls) != 1 {
+			t.Errorf("filer %s: got %d register calls, want 1", addr, len(fc.registerCalls))
+		}
+		fc.mu.Unlock()
+	}
+}
+
+// TestPeerRegistrar_ListMergesAcrossFilers guards cross-filer convergence:
+// mount A registered on filer-1, mount B on filer-2; a registrar that
+// lists both filers must see both mounts.
+func TestPeerRegistrar_ListMergesAcrossFilers(t *testing.T) {
+	fc1 := &fakeFilerClient{
+		listResponse: filer_pb.MountListResponse{
+			Mounts: []*filer_pb.MountInfo{{PeerAddr: "mount-a:18080", Rack: "r1", LastSeenNs: 200}},
+		},
+	}
+	fc2 := &fakeFilerClient{
+		listResponse: filer_pb.MountListResponse{
+			Mounts: []*filer_pb.MountInfo{{PeerAddr: "mount-b:18080", Rack: "r2", LastSeenNs: 200}},
+		},
+	}
+	a1, a2 := pb.ServerAddress("f1:18888"), pb.ServerAddress("f2:18888")
+	fleet := &fakeFilerFleet{clients: map[pb.ServerAddress]*fakeFilerClient{a1: fc1, a2: fc2}}
+
+	r := NewPeerRegistrar([]pb.ServerAddress{a1, a2}, fleet.dial, "self:18080", "", "")
+	if err := r.listOnce(context.Background()); err != nil {
+		t.Fatalf("listOnce: %v", err)
+	}
+
+	seeds := r.Seeds()
+	if len(seeds) != 2 {
+		t.Fatalf("want 2 merged seeds, got %d: %+v", len(seeds), seeds)
+	}
+	addrs := map[string]bool{}
+	for _, s := range seeds {
+		addrs[s.PeerAddr] = true
+	}
+	if !addrs["mount-a:18080"] || !addrs["mount-b:18080"] {
+		t.Errorf("merged seeds missing an entry: %+v", addrs)
+	}
+}
+
+// TestPeerRegistrar_ListMergeKeepsNewestLastSeen guards the dedupe rule:
+// the same mount reported by two filers collapses to one entry, keeping
+// the freshest LastSeenNs for liveness-ordering decisions.
+func TestPeerRegistrar_ListMergeKeepsNewestLastSeen(t *testing.T) {
+	fc1 := &fakeFilerClient{
+		listResponse: filer_pb.MountListResponse{
+			Mounts: []*filer_pb.MountInfo{{PeerAddr: "mount-a:18080", Rack: "r1", LastSeenNs: 100}},
+		},
+	}
+	fc2 := &fakeFilerClient{
+		listResponse: filer_pb.MountListResponse{
+			Mounts: []*filer_pb.MountInfo{{PeerAddr: "mount-a:18080", Rack: "r1", LastSeenNs: 500}},
+		},
+	}
+	a1, a2 := pb.ServerAddress("f1:18888"), pb.ServerAddress("f2:18888")
+	fleet := &fakeFilerFleet{clients: map[pb.ServerAddress]*fakeFilerClient{a1: fc1, a2: fc2}}
+
+	r := NewPeerRegistrar([]pb.ServerAddress{a1, a2}, fleet.dial, "self:18080", "", "")
+	if err := r.listOnce(context.Background()); err != nil {
+		t.Fatalf("listOnce: %v", err)
+	}
+	seeds := r.Seeds()
+	if len(seeds) != 1 {
+		t.Fatalf("want 1 deduped seed, got %d", len(seeds))
+	}
+}

weed/mount/weedfs.go

@@ -139,6 +139,7 @@ type WFS struct {
 	hardLinkLockTable    *util.LockTable[string]
 	posixLocks           *PosixLockTable
 	rdmaClient           *RDMAMountClient
+	peerRegistrar        *PeerRegistrar
 	FilerConf            *filer.FilerConf
 	filerClient          *wdclient.FilerClient // Cached volume location client
 	refreshMu            sync.Mutex
@@ -336,6 +337,9 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		if wfs.rdmaClient != nil {
 			wfs.rdmaClient.Close()
 		}
+		if wfs.peerRegistrar != nil {
+			wfs.peerRegistrar.Stop()
+		}
 	})
 
 	// Initialize RDMA client if enabled
@@ -355,6 +359,36 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		}
 	}
 
+	// Peer chunk sharing: register with every configured filer's mount
+	// registry. PR 3 resolved the advertise address; pass it through so
+	// the registrar heartbeats a reachable identity rather than a wildcard
+	// bind. Broadcasting to the full filer set is what lets mounts pointing
+	// at different filers see each other — each filer's registry is
+	// in-memory and there is no filer-to-filer sync. The gRPC server that
+	// serves ChunkAnnounce/Lookup/FetchChunk is started later (PR 5); until
+	// then the registrar is a no-op beyond heartbeats.
+	if option.PeerEnabled {
+		selfAddr, err := ResolvePeerAdvertiseAddr(option.PeerListen, option.PeerAdvertise)
+		if err != nil {
+			// Downstream code treats PeerEnabled as "peer infrastructure
+			// is ready": later PRs wire the gRPC server, fetcher hook,
+			// and announcer from this flag. If we can't resolve a
+			// reachable self-address those components would nil-deref
+			// or advertise garbage, so disable the feature instead of
+			// limping along half-initialized.
+			glog.Warningf("peer: cannot resolve advertise addr, disabling peer sharing: %v", err)
+			option.PeerEnabled = false
+		} else {
+			dial := func(ctx context.Context, addr pb.ServerAddress, fn func(client filer_pb.SeaweedFilerClient) error) error {
+				return pb.WithGrpcFilerClient(false, 0, addr, option.GrpcDialOption, fn)
+			}
+			wfs.peerRegistrar = NewPeerRegistrar(option.FilerAddresses, dial, selfAddr, option.PeerDataCenter, option.PeerRack)
+			if err := wfs.peerRegistrar.Start(context.Background()); err != nil {
+				glog.Warningf("peer registrar start: %v", err)
+			}
+		}
+	}
+
 	if wfs.option.ConcurrentWriters > 0 {
 		wfs.concurrentWriters = util.NewLimitedConcurrentExecutor(wfs.option.ConcurrentWriters)
 		wfs.concurrentCopiersSem = make(chan struct{}, wfs.option.ConcurrentWriters)