peer chunk sharing 7/8: tryPeerRead read-path hook (#9136)

* mount: batched announcer + pooled peer conns for mount-to-mount RPCs

* peer_announcer.go: non-blocking EnqueueAnnounce + ticker flush that
  groups fids by HRW owner, fans out one ChunkAnnounce per owner in
  parallel. announcedAt is pruned at 2× TTL so it stays bounded.

* peer_dialer.go: PeerConnPool caches one grpc.ClientConn per peer
  address; the announcer and (next PR) the fetcher share it so
  steady-state owner RPCs skip the handshake cost entirely. Bounded
  at 4096 cached entries; shutdown conns are transparently replaced.

* WFS starts both alongside the gRPC server; stops them on unmount.

* mount: wire tryPeerRead via FetchChunk streaming gRPC

Replaces the HTTP GET byte-transfer path with a gRPC server-stream
FetchChunk call. Same fall-through semantics: any failure drops
through to entryChunkGroup.ReadDataAt, so reads never slow below
status quo.

* peer_fetcher.go: tryPeerRead resolves the offset to a leaf chunk
  (flattening manifests), asks the HRW owner for holders via
  ChunkLookup, then opens FetchChunk on each holder in LRU order
  (PR #5) until one succeeds. Assembled bytes are verified against
  FileChunk.ETag end-to-end — the peer is still treated as
  untrusted. Reuses the shared PeerConnPool from PR #6 for all
  outbound gRPC.

* peer_grpc.go: expose SelfAddr() so the fetcher can avoid dialing
  itself on a self-owned fid.

* filehandle_read.go: tryPeerRead slot between tryRDMARead and
  entryChunkGroup.ReadDataAt. Gated by option.PeerEnabled and the
  presence of peerGrpcServer (the single identity test).

Read ordering with the feature enabled is now:
   local cache -> RDMA sidecar -> peer mount (gRPC stream) -> volume server

One port, one identity, one connection pool — no more HTTP bytecast.

* test(fuse_p2p): end-to-end CI test for peer chunk sharing

Adds a FUSE-backed integration test that proves mount B can satisfy a
read from mount A's chunk cache instead of the volume tier.

Layout (modelled on test/fuse_dlm):

  test/fuse_p2p/framework_test.go        — cluster harness (1 master,
                                           1 volume, 1 filer, N mounts,
                                           all with -peer.enable)
  test/fuse_p2p/peer_chunk_sharing_test.go
                                         — writer-reader scenario

The test (TestPeerChunkSharing_ReadersPullFromPeerCache):

  1. Starts 3 mounts. Three is the sweet spot: with 2 mounts, HRW owner
     of a chunk is self ~50 % of the time (peer path short-circuits);
     with 3+ it drops to ≤ 1/3, so a multi-chunk file almost certainly
     exercises the remote-owner fan-out.
  2. Mount 0 writes a ~8 MiB file, then reads it back through its own
     FUSE to warm its chunk cache.
  3. Waits for seed convergence (one full MountList refresh) plus an
     announcer flush cycle, so chunk-holder entries have reached each
     HRW owner.
  4. Mount 1 reads the same file.
  5. Verifies byte-for-byte equality AND greps mount 1's log for
     "peer read successful" — content matching alone is not proof
     (the volume fallback would also succeed), so the log marker is
     what distinguishes p2p from fallback.

Workflow .github/workflows/fuse-p2p-integration.yml triggers on any
change to mount/filer peer code, the p2p protos, or the test itself.
Failure artifacts (server + mount logs) are uploaded for 3 days.

Mounts run with -v=4 so the tryPeerRead success/failure glog messages
land in the log file the test greps.

.github/workflows/fuse-p2p-integration.yml

@@ -0,0 +1,69 @@
+name: "FUSE P2P Peer Chunk Sharing Integration Tests"
+
+on:
+  pull_request:
+    paths:
+      - 'weed/command/mount*.go'
+      - 'weed/mount/**'
+      - 'weed/filer/mount_peer_registry*.go'
+      - 'weed/server/filer_grpc_server_mount_peer.go'
+      - 'weed/pb/mount_peer.proto'
+      - 'weed/pb/filer.proto'
+      - 'test/fuse_p2p/**'
+      - '.github/workflows/fuse-p2p-integration.yml'
+  push:
+    branches: [master]
+    paths:
+      - 'weed/command/mount*.go'
+      - 'weed/mount/**'
+      - 'weed/filer/mount_peer_registry*.go'
+      - 'weed/server/filer_grpc_server_mount_peer.go'
+      - 'weed/pb/mount_peer.proto'
+      - 'weed/pb/filer.proto'
+      - 'test/fuse_p2p/**'
+
+concurrency:
+  group: ${{ github.head_ref || github.ref }}/fuse-p2p-integration
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  fuse-p2p-integration:
+    name: FUSE P2P Peer Chunk Sharing
+    runs-on: ubuntu-22.04
+    timeout-minutes: 20
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v6
+
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: 'go.mod'
+
+      - name: Install FUSE dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y libfuse3-dev
+          echo 'user_allow_other' | sudo tee -a /etc/fuse.conf
+          sudo chmod 644 /etc/fuse.conf
+
+      - name: Build SeaweedFS
+        run: go build -o weed/weed -buildvcs=false ./weed
+
+      - name: Run P2P integration tests
+        timeout-minutes: 15
+        env:
+          WEED_BINARY: ${{ github.workspace }}/weed/weed
+        run: go test -v -count=1 -timeout=12m ./test/fuse_p2p/...
+
+      - name: Upload logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: fuse-p2p-test-logs
+          path: /tmp/seaweedfs-fuse-p2p-logs/
+          retention-days: 3

test/fuse_p2p/framework_test.go

@@ -0,0 +1,390 @@
+//go:build linux || darwin
+
+package fuse_p2p
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/stretchr/testify/require"
+)
+
+// p2pTestCluster manages a minimal SeaweedFS cluster exercising the peer
+// chunk-sharing (p2p) read path: 1 master, 1 volume, 1 filer, and N FUSE
+// mounts that all have -peer.enable set. Three mounts is the sweet spot
+// for integration testing — with 2 mounts the HRW owner of a chunk is
+// ~50% the reader itself (which short-circuits the peer path); with 3+
+// mounts it's ≤ 1/3, so a multi-chunk file almost certainly exercises
+// the remote-owner fan-out.
+type p2pTestCluster struct {
+	t          testing.TB
+	baseDir    string
+	weedBinary string
+
+	masterPort     int
+	masterGrpcPort int
+	volumePort     int
+	volumeGrpcPort int
+	filerPort      int
+	filerGrpcPort  int
+	mountPeerPorts []int
+	mountPoints    []string
+
+	masterCmd *exec.Cmd
+	volumeCmd *exec.Cmd
+	filerCmd  *exec.Cmd
+	mountCmds []*exec.Cmd
+	logFiles  []*os.File
+
+	cleanupOnce sync.Once
+}
+
+// startP2PTestCluster brings up a cluster with numMounts FUSE mounts,
+// every one advertising on its own -peer.listen port. All mounts register
+// with the same filer. Verbose logging (-v=4) is enabled on each mount so
+// peer-read success/failure messages land in the log file and the test
+// can grep them to verify the p2p path fired.
+func startP2PTestCluster(t testing.TB, numMounts int) *p2pTestCluster {
+	require.GreaterOrEqual(t, numMounts, 2, "need at least 2 mounts to exercise p2p")
+	binary := findWeedBinary()
+	if binary == "" {
+		t.Skip("weed binary not found; set WEED_BINARY or ensure it is on PATH")
+	}
+	baseDir, err := os.MkdirTemp("", "seaweedfs_fuse_p2p_test_")
+	require.NoError(t, err)
+
+	c := &p2pTestCluster{
+		t:              t,
+		baseDir:        baseDir,
+		weedBinary:     binary,
+		mountPeerPorts: make([]int, numMounts),
+		mountPoints:    make([]string, numMounts),
+		mountCmds:      make([]*exec.Cmd, numMounts),
+	}
+	t.Cleanup(c.Stop)
+
+	// master(2) + volume(2) + filer(2) + one peer port per mount.
+	// testutil.AllocatePorts holds all listeners open until every port
+	// is reserved, avoiding the brief close→bind race that would
+	// happen with a per-listener close loop.
+	ports, err := testutil.AllocatePorts(6 + numMounts)
+	require.NoError(t, err)
+	c.masterPort = ports[0]
+	c.masterGrpcPort = ports[1]
+	c.volumePort = ports[2]
+	c.volumeGrpcPort = ports[3]
+	c.filerPort = ports[4]
+	c.filerGrpcPort = ports[5]
+	for i := 0; i < numMounts; i++ {
+		c.mountPeerPorts[i] = ports[6+i]
+	}
+
+	configDir := filepath.Join(baseDir, "config")
+	require.NoError(t, os.MkdirAll(configDir, 0755))
+	require.NoError(t, os.WriteFile(filepath.Join(configDir, "security.toml"), []byte(""), 0644))
+
+	require.NoError(t, c.startMaster(configDir))
+	require.NoError(t, c.waitForTCP(c.masterCmd, "master",
+		fmt.Sprintf("127.0.0.1:%d", c.masterPort), 30*time.Second))
+
+	require.NoError(t, c.startVolume(configDir))
+	require.NoError(t, c.waitForTCP(c.volumeCmd, "volume",
+		fmt.Sprintf("127.0.0.1:%d", c.volumePort), 30*time.Second))
+
+	require.NoError(t, c.startFiler(configDir))
+	require.NoError(t, c.waitForTCP(c.filerCmd, "filer",
+		fmt.Sprintf("127.0.0.1:%d", c.filerGrpcPort), 30*time.Second))
+
+	for i := 0; i < numMounts; i++ {
+		mp := filepath.Join(baseDir, fmt.Sprintf("mount%d", i))
+		require.NoError(t, os.MkdirAll(mp, 0755))
+		c.mountPoints[i] = mp
+		require.NoError(t, c.startMount(i, configDir))
+		require.NoError(t, c.waitForMount(mp, 30*time.Second),
+			"mount %d not ready\n%s", i, c.tailLog(fmt.Sprintf("mount%d", i)))
+	}
+	return c
+}
+
+func (c *p2pTestCluster) Stop() {
+	if c == nil {
+		return
+	}
+	c.cleanupOnce.Do(func() {
+		for i := len(c.mountCmds) - 1; i >= 0; i-- {
+			stopCmd(c.mountCmds[i])
+			// Backup unmount in case the FUSE teardown didn't clean up.
+			exec.Command("fusermount3", "-u", c.mountPoints[i]).Run()
+			exec.Command("fusermount", "-u", c.mountPoints[i]).Run()
+		}
+		stopCmd(c.filerCmd)
+		stopCmd(c.volumeCmd)
+		stopCmd(c.masterCmd)
+
+		for _, f := range c.logFiles {
+			f.Close()
+		}
+		c.copyLogsForCI()
+		if !c.t.Failed() {
+			os.RemoveAll(c.baseDir)
+		}
+		time.Sleep(2 * time.Second) // let ports drain
+	})
+}
+
+// MountDir returns the filesystem path of the i-th mount.
+func (c *p2pTestCluster) MountDir(i int) string { return c.mountPoints[i] }
+
+// masterAddress / filerAddress return SeaweedFS-style addresses encoding
+// both ports as "host:httpPort.grpcPort". Without this, downstream
+// components fall back to the grpcPort = httpPort + 10000 default,
+// which doesn't match the random port we allocate.
+func (c *p2pTestCluster) masterAddress() string {
+	return string(pb.NewServerAddress("127.0.0.1", c.masterPort, c.masterGrpcPort))
+}
+
+func (c *p2pTestCluster) filerAddress() string {
+	return string(pb.NewServerAddress("127.0.0.1", c.filerPort, c.filerGrpcPort))
+}
+
+// MountLog returns the contents of mount i's log file.
+func (c *p2pTestCluster) MountLog(i int) string {
+	return c.tailLogFull(fmt.Sprintf("mount%d", i))
+}
+
+func (c *p2pTestCluster) startMaster(configDir string) error {
+	c.masterCmd = exec.Command(c.weedBinary,
+		"-logdir="+filepath.Join(c.baseDir, "logs"),
+		"master",
+		"-ip=127.0.0.1",
+		"-ip.bind=127.0.0.1",
+		"-port="+strconv.Itoa(c.masterPort),
+		"-port.grpc="+strconv.Itoa(c.masterGrpcPort),
+		"-mdir="+filepath.Join(c.baseDir, "master"),
+	)
+	return c.startCmd(c.masterCmd, "master")
+}
+
+func (c *p2pTestCluster) startVolume(configDir string) error {
+	volDir := filepath.Join(c.baseDir, "volume")
+	if err := os.MkdirAll(volDir, 0755); err != nil {
+		return fmt.Errorf("create volume dir: %w", err)
+	}
+	c.volumeCmd = exec.Command(c.weedBinary,
+		"-logdir="+filepath.Join(c.baseDir, "logs"),
+		"volume",
+		"-ip=127.0.0.1",
+		"-ip.bind=127.0.0.1",
+		"-port="+strconv.Itoa(c.volumePort),
+		"-port.grpc="+strconv.Itoa(c.volumeGrpcPort),
+		"-master="+c.masterAddress(),
+		"-dir="+volDir,
+		"-max=10",
+	)
+	return c.startCmd(c.volumeCmd, "volume")
+}
+
+func (c *p2pTestCluster) startFiler(configDir string) error {
+	filerDir := filepath.Join(c.baseDir, "filer")
+	if err := os.MkdirAll(filerDir, 0755); err != nil {
+		return fmt.Errorf("create filer dir: %w", err)
+	}
+	c.filerCmd = exec.Command(c.weedBinary,
+		"-logdir="+filepath.Join(c.baseDir, "logs"),
+		"filer",
+		"-ip=127.0.0.1",
+		"-ip.bind=127.0.0.1",
+		"-port="+strconv.Itoa(c.filerPort),
+		"-port.grpc="+strconv.Itoa(c.filerGrpcPort),
+		"-master="+c.masterAddress(),
+		"-defaultStoreDir="+filerDir,
+		// default is -mount.p2p=true; be explicit so the intent is grep-able.
+		"-mount.p2p=true",
+	)
+	return c.startCmd(c.filerCmd, "filer")
+}
+
+func (c *p2pTestCluster) startMount(idx int, configDir string) error {
+	cacheDir := filepath.Join(c.baseDir, fmt.Sprintf("cache%d", idx))
+	if err := os.MkdirAll(cacheDir, 0755); err != nil {
+		return fmt.Errorf("create cache dir: %w", err)
+	}
+	c.mountCmds[idx] = exec.Command(c.weedBinary,
+		"-logdir="+filepath.Join(c.baseDir, "logs"),
+		"-v=4",
+		"mount",
+		"-filer="+c.filerAddress(),
+		"-dir="+c.mountPoints[idx],
+		"-filer.path=/",
+		"-dirAutoCreate",
+		"-allowOthers=false",
+		"-cacheDir="+cacheDir,
+		"-peer.enable=true",
+		fmt.Sprintf("-peer.listen=127.0.0.1:%d", c.mountPeerPorts[idx]),
+		fmt.Sprintf("-peer.advertise=127.0.0.1:%d", c.mountPeerPorts[idx]),
+		"-peer.dataCenter=dc1",
+		fmt.Sprintf("-peer.rack=rack%d", idx),
+	)
+	return c.startCmd(c.mountCmds[idx], fmt.Sprintf("mount%d", idx))
+}
+
+func (c *p2pTestCluster) startCmd(cmd *exec.Cmd, name string) error {
+	logPath := filepath.Join(c.baseDir, "logs")
+	if err := os.MkdirAll(logPath, 0755); err != nil {
+		return fmt.Errorf("create log dir: %w", err)
+	}
+	logFile, err := os.Create(filepath.Join(logPath, name+".log"))
+	if err != nil {
+		return err
+	}
+	c.logFiles = append(c.logFiles, logFile)
+	cmd.Stdout = logFile
+	cmd.Stderr = logFile
+	return cmd.Start()
+}
+
+func (c *p2pTestCluster) tailLog(name string) string {
+	data, err := os.ReadFile(filepath.Join(c.baseDir, "logs", name+".log"))
+	if err != nil {
+		return fmt.Sprintf("(log %s not available: %v)", name, err)
+	}
+	const maxTail = 8192
+	if len(data) > maxTail {
+		data = data[len(data)-maxTail:]
+	}
+	return string(data)
+}
+
+func (c *p2pTestCluster) tailLogFull(name string) string {
+	data, err := os.ReadFile(filepath.Join(c.baseDir, "logs", name+".log"))
+	if err != nil {
+		return ""
+	}
+	return string(data)
+}
+
+func (c *p2pTestCluster) copyLogsForCI() {
+	ciLogDir := "/tmp/seaweedfs-fuse-p2p-logs"
+	os.MkdirAll(ciLogDir, 0755)
+	logsDir := filepath.Join(c.baseDir, "logs")
+	entries, err := os.ReadDir(logsDir)
+	if err != nil {
+		return
+	}
+	for _, e := range entries {
+		data, err := os.ReadFile(filepath.Join(logsDir, e.Name()))
+		if err != nil {
+			continue
+		}
+		os.WriteFile(filepath.Join(ciLogDir, e.Name()), data, 0644)
+	}
+}
+
+// waitForTCP polls addr until it accepts a connection, OR the supplied
+// subprocess exits — whichever comes first. Short-circuiting on child
+// exit turns a 30 s-spin-on-dead-process into an immediate failure with
+// the tail of its log. cmd may be nil for callers that don't track a
+// process; in that case we fall back to pure polling.
+//
+// Liveness is checked with signal 0 (POSIX "is this process alive").
+// We deliberately do NOT call cmd.Wait() in a goroutine here because
+// stopCmd() later calls Wait() at shutdown, and only one Wait per
+// process is allowed.
+func (c *p2pTestCluster) waitForTCP(cmd *exec.Cmd, name, addr string, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		conn, err := net.DialTimeout("tcp", addr, time.Second)
+		if err == nil {
+			conn.Close()
+			return nil
+		}
+		if cmd != nil && cmd.Process != nil {
+			// signal 0 doesn't actually send anything; it just asks
+			// the kernel whether the process exists. An error here
+			// (typically "process already finished" or ESRCH) means
+			// the child died before coming up.
+			if sigErr := cmd.Process.Signal(syscall.Signal(0)); sigErr != nil {
+				return fmt.Errorf("%s exited before listening on %s: %v\n%s",
+					name, addr, sigErr, c.tailLog(name))
+			}
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	return fmt.Errorf("service at %s not ready within timeout\n%s", addr, c.tailLog(name))
+}
+
+// waitForMount waits for a FUSE filesystem to actually be mounted by
+// watching the device id flip (FUSE mounts have a different Dev than
+// their parent dir).
+func (c *p2pTestCluster) waitForMount(mountPoint string, timeout time.Duration) error {
+	parentDir := filepath.Dir(mountPoint)
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		parentStat, err := os.Stat(parentDir)
+		if err != nil {
+			time.Sleep(200 * time.Millisecond)
+			continue
+		}
+		mountStat, err := os.Stat(mountPoint)
+		if err != nil {
+			time.Sleep(200 * time.Millisecond)
+			continue
+		}
+		parentSys := parentStat.Sys().(*syscall.Stat_t)
+		mountSys := mountStat.Sys().(*syscall.Stat_t)
+		if parentSys.Dev != mountSys.Dev {
+			return nil
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	return fmt.Errorf("mount point %s not ready within timeout (FUSE not detected)", mountPoint)
+}
+
+// --- utilities (binary discovery, process shutdown) ---
+//
+// Port allocation is delegated to testutil.MustAllocatePorts, which
+// holds all listeners open until every port is reserved before
+// returning them in a batch — safer than the per-listener
+// close-then-reserve pattern fuse_dlm originally used.
+
+func findWeedBinary() string {
+	if env := os.Getenv("WEED_BINARY"); env != "" {
+		if _, err := os.Stat(env); err == nil {
+			return env
+		}
+	}
+	if p, err := exec.LookPath("weed"); err == nil {
+		return p
+	}
+	return ""
+}
+
+// stopCmd kills a child process with SIGTERM and waits up to 5 s, then
+// escalates to SIGKILL. Tolerates nil and already-exited processes.
+func stopCmd(cmd *exec.Cmd) {
+	if cmd == nil || cmd.Process == nil {
+		return
+	}
+	cmd.Process.Signal(syscall.SIGTERM)
+	done := make(chan struct{})
+	go func() {
+		cmd.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		cmd.Process.Signal(syscall.SIGKILL)
+		<-done
+	}
+}

test/fuse_p2p/peer_chunk_sharing_test.go

@@ -0,0 +1,147 @@
+package fuse_p2p
+
+import (
+	"bytes"
+	"crypto/md5"
+	"fmt"
+	"math/rand/v2"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// peerReadSuccessMarker is the log line tryPeerRead emits when a peer
+// fetch succeeded. The test greps non-writer mount logs for it to
+// prove the p2p path fired. At glog verbosity 4 (framework sets -v=4).
+const peerReadSuccessMarker = "peer read successful"
+
+// seedConvergenceTimeout bounds how long the test waits for:
+//   - the filer registry to list all mounts,
+//   - every mount's seed view (via MountList poll) to include all peers,
+//   - the first announcer flush cycle to publish chunk holders.
+//
+// With defaults the mount polls MountList every 30 s and flushes
+// ChunkAnnounce every 15 s. Allowing 45 s absorbs one MountList refresh
+// plus one announce cycle plus some slop for CI variance.
+const seedConvergenceTimeout = 90 * time.Second
+
+// TestPeerChunkSharing_ReadersPullFromPeerCache is the headline p2p
+// integration test. It proves at least one non-writer mount can satisfy
+// a read from the writer's chunk cache instead of the volume tier.
+//
+//  1. Bring up 1 master/volume/filer + 3 mounts, all with -peer.enable.
+//  2. Mount 0 writes a ~8 MiB file and reads it back so chunks land in
+//     its local cache and the announcer publishes them.
+//  3. Wait for seed convergence + at least one announcer flush cycle.
+//  4. BOTH mount 1 and mount 2 read the file.
+//
+// Why both readers: with 3 mounts, HRW picks one owner for the chunk.
+// If that owner is mount 1, only mount 2's read will hit the peer
+// path (mount 1's tryPeerRead bails on owner==self). If that owner is
+// mount 2, only mount 1's read will. If that owner is mount 0 (the
+// writer), both can. So by reading from both, we deterministically
+// guarantee at least one non-writer mount exercises the peer path.
+//
+// Once the peer fetch populates its local cache, subsequent reads
+// short-circuit on IsInCache — so we only get one real shot per mount
+// per chunk. That's fine: one success is all the test needs.
+func TestPeerChunkSharing_ReadersPullFromPeerCache(t *testing.T) {
+	c := startP2PTestCluster(t, 3)
+
+	// ~8 MiB, pseudo-random so compression doesn't collapse it to one block.
+	payload := make([]byte, 8*1024*1024)
+	rng := rand.New(rand.NewPCG(1, 2))
+	for i := range payload {
+		payload[i] = byte(rng.Uint32())
+	}
+
+	const relPath = "p2p-test.bin"
+	writer := c.MountDir(0)
+	require.NoError(t, os.WriteFile(filepath.Join(writer, relPath), payload, 0644))
+
+	// Warm mount 0's chunk cache by reading back through its own FUSE.
+	// Without this the chunks are on the volume server but not yet
+	// in anyone's peer-servable cache.
+	readBack, err := os.ReadFile(filepath.Join(writer, relPath))
+	require.NoError(t, err)
+	require.True(t, bytes.Equal(readBack, payload), "write-then-read on writer mount should match")
+
+	waitForSeedConvergence(t, c, seedConvergenceTimeout)
+
+	// Give the announcer several flush windows to push chunk-holder
+	// entries to the HRW owners. First flush may see the writer's seed
+	// view incomplete (only self) and defer all fids; subsequent
+	// flushes re-check against a refreshed seed view. announce interval
+	// is 15 s, so 45 s covers three attempts.
+	time.Sleep(45 * time.Second)
+
+	// Read from both non-writer mounts. For any HRW outcome on any
+	// chunk, at least one of these reads will NOT have the reader as
+	// the HRW owner, so its tryPeerRead will proceed to ChunkLookup +
+	// FetchChunk.
+	for _, idx := range []int{1, 2} {
+		got, err := os.ReadFile(filepath.Join(c.MountDir(idx), relPath))
+		require.NoError(t, err, "read from mount %d must succeed\n--- mount%d ---\n%s",
+			idx, idx, tailLines(c.MountLog(idx), 80))
+		require.Equal(t, md5.Sum(payload), md5.Sum(got),
+			"mount %d returned mismatched bytes (len got=%d want=%d)", idx, len(got), len(payload))
+	}
+
+	// Content matches alone doesn't prove p2p — the volume fallback
+	// would also satisfy the reads. Require at least one non-writer
+	// mount's log to contain the peer-read success marker.
+	var sawPeerRead bool
+	for _, idx := range []int{1, 2} {
+		if strings.Contains(c.MountLog(idx), peerReadSuccessMarker) {
+			sawPeerRead = true
+			break
+		}
+	}
+	if !sawPeerRead {
+		t.Fatalf("no non-writer mount logged %q — peer read path never fired.\n"+
+			"--- mount0 (writer) tail ---\n%s\n--- mount1 tail ---\n%s\n--- mount2 tail ---\n%s",
+			peerReadSuccessMarker,
+			tailLines(c.MountLog(0), 60), tailLines(c.MountLog(1), 60), tailLines(c.MountLog(2), 60))
+	}
+}
+
+// waitForSeedConvergence polls each mount's log looking for any sign
+// that MountList returned a peer list containing the other mounts.
+func waitForSeedConvergence(t *testing.T, c *p2pTestCluster, timeout time.Duration) {
+	t.Helper()
+	// The first MountRegister happens synchronously during mount startup,
+	// and the first MountList is pulled right after. 30 s is the refresh
+	// interval; waiting one full cycle here guarantees every mount has
+	// at minimum observed the others in its seed view.
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		ready := true
+		for i := range c.mountCmds {
+			if !strings.Contains(c.MountLog(i), "peer-grpc listening on") {
+				ready = false
+				break
+			}
+		}
+		if ready {
+			time.Sleep(30 * time.Second)
+			return
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+	t.Fatalf("peer-grpc servers did not come up within %s", timeout)
+}
+
+// tailLines returns the last n newline-delimited lines of s, or the
+// whole thing if shorter. Keeps test failures readable.
+func tailLines(s string, n int) string {
+	lines := strings.Split(s, "\n")
+	if len(lines) <= n {
+		return s
+	}
+	return fmt.Sprintf("... (%d earlier lines omitted) ...\n%s",
+		len(lines)-n, strings.Join(lines[len(lines)-n:], "\n"))
+}

weed/mount/filehandle_read.go

@@ -75,6 +75,23 @@ func (fh *FileHandle) readFromChunksWithContext(ctx context.Context, buff []byte
 		glog.V(4).Infof("RDMA read failed for %s, falling back to HTTP: %v", fileFullPath, err)
 	}
 
+	// Peer chunk sharing: try a peer mount's cache before the volume tier.
+	// Any failure falls through transparently. See design-weed-mount-
+	// peer-chunk-sharing.md §4.3.
+	if fh.wfs.option.PeerEnabled && fh.wfs.peerGrpcServer != nil {
+		totalRead, ts, err := fh.tryPeerRead(ctx, fileSize, buff, offset, entry)
+		if err == nil {
+			glog.V(4).Infof("peer read successful for %s [%d,%d] %d", fileFullPath, offset, offset+int64(totalRead), totalRead)
+			return int64(totalRead), ts, nil
+		}
+		// Skip the "failed" log for benign skip reasons (local cache
+		// hit, no peer owner yet, etc.) — the cache/volume fallback is
+		// the expected outcome, not a failure.
+		if err != errPeerReadSkipped {
+			glog.V(4).Infof("peer read failed for %s, falling back to volume: %v", fileFullPath, err)
+		}
+	}
+
 	// Fall back to normal chunk reading
 	totalRead, ts, err := fh.entryChunkGroup.ReadDataAt(ctx, fileSize, buff, offset)
 

weed/mount/peer_fetcher.go

@@ -0,0 +1,324 @@
+package mount
+
+import (
+	"bytes"
+	"context"
+	"crypto/md5"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"sort"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/mount_peer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// errPeerReadSkipped signals that the read didn't go through the peer
+// path for a benign reason (local cache hit, no peer owner, etc.) rather
+// than a genuine failure. Callers fall through to the volume path just
+// as they would for a real error, but suppress the "peer read failed"
+// log that would otherwise mislead operators into thinking peer sharing
+// is broken.
+var errPeerReadSkipped = fmt.Errorf("peer read skipped (not an error)")
+
+// peerLookupTimeout bounds the ChunkLookup RPC. Short because the result
+// is consumed on the read critical path.
+const peerLookupTimeout = 500 * time.Millisecond
+
+// peerFetchTimeout bounds a single FetchChunk stream; on expiry we fall
+// through to the next holder or the volume server.
+const peerFetchTimeout = 5 * time.Second
+
+// maxPeerFetchChunkBytes caps how much we will accept from a single
+// FetchChunk stream. gRPC per-message size is already capped by the
+// server option; this is belt-and-suspenders against a runaway peer.
+const maxPeerFetchChunkBytes = 64 * 1024 * 1024
+
+// tryPeerRead attempts to satisfy a read from a peer mount's chunk cache.
+// Returns (bytesRead, modifiedTsNs, nil) on success. On any failure it
+// returns (0, 0, err) so the caller falls through to
+// entryChunkGroup.ReadDataAt (the volume-server path).
+//
+// Flow:
+//   1. Resolve the offset's leaf chunk (flattening manifests).
+//   2. Ask the HRW owner mount for current holders via ChunkLookup.
+//   3. For each holder (LRU order from PR #5), open a FetchChunk stream,
+//      assemble frames into a size-bounded buffer, and verify MD5
+//      end-to-end against FileChunk.ETag.
+//   4. On success, populate chunk_cache and enqueue an announce so
+//      other mounts can discover us as a new holder.
+func (fh *FileHandle) tryPeerRead(ctx context.Context, fileSize int64, buff []byte, offset int64, entry *LockedEntry) (int64, int64, error) {
+	if fh.wfs.peerRegistrar == nil || fh.wfs.peerConnPool == nil {
+		return 0, 0, fmt.Errorf("peer sharing not configured")
+	}
+
+	// Resolve offset → leaf chunk, flattening any manifest indirection.
+	readStop := offset + int64(len(buff))
+	if readStop > fileSize {
+		readStop = fileSize
+	}
+	dataChunks, _, err := filer.ResolveChunkManifest(ctx, fh.wfs.LookupFn(), entry.GetEntry().Chunks, offset, readStop)
+	if err != nil {
+		return 0, 0, fmt.Errorf("resolve manifest: %w", err)
+	}
+	targetChunk, chunkOffset := findChunkContaining(dataChunks, offset)
+	if targetChunk == nil {
+		return 0, 0, fmt.Errorf("no leaf chunk for offset %d", offset)
+	}
+	// Reject reads that cross a chunk boundary. We only fetch one chunk
+	// here, and the caller (readFromChunks) maps a short non-error
+	// return to "success, zero-fill the rest" — which would silently
+	// corrupt reads that should actually span two chunks. Bail so the
+	// fallback ReadDataAt path can handle the multi-chunk case
+	// correctly. See weedfs_file_read.go short-read semantics.
+	chunkEnd := targetChunk.Offset + int64(targetChunk.Size)
+	if readStop > chunkEnd {
+		return 0, 0, errPeerReadSkipped
+	}
+
+	// Fail fast when the chunk is already cached locally: the fallback
+	// ReadDataAt path will satisfy the read from chunkCache with no RPCs,
+	// so dialing a peer (ChunkLookup + FetchChunk) would be pure overhead.
+	if fh.wfs.chunkCache != nil && fh.wfs.chunkCache.IsInCache(targetChunk.FileId, true) {
+		return 0, 0, errPeerReadSkipped
+	}
+
+	selfAddr := ""
+	if fh.wfs.peerGrpcServer != nil {
+		selfAddr = fh.wfs.peerGrpcServer.SelfAddr()
+	}
+
+	owner := fh.wfs.peerRegistrar.OwnerFor(targetChunk.FileId)
+	if owner == "" || owner == selfAddr {
+		return 0, 0, fmt.Errorf("no peer owner for fid %s", targetChunk.FileId)
+	}
+
+	holders, err := peerLookupHolders(ctx, fh.wfs.peerConnPool.Dialer(), owner, targetChunk.FileId)
+	if err != nil {
+		return 0, 0, fmt.Errorf("peer lookup: %w", err)
+	}
+	if len(holders) == 0 {
+		return 0, 0, fmt.Errorf("no peer holder for fid %s", targetChunk.FileId)
+	}
+
+	// Re-rank holders by locality (same rack > same DC > elsewhere), keeping
+	// the server's LRU order stable within each bucket. The server caps the
+	// list at maxLookupHolders, so this is always a small N.
+	sortHoldersByLocality(holders, fh.wfs.option.PeerDataCenter, fh.wfs.option.PeerRack)
+
+	dial := fh.wfs.peerConnPool.Dialer()
+	for _, h := range holders {
+		if h.addr == selfAddr {
+			continue
+		}
+		data, ferr := fetchChunkFromPeer(ctx, dial, h.addr, targetChunk.FileId, targetChunk.Size, targetChunk.ETag)
+		if ferr != nil {
+			glog.V(2).Infof("peer-fetch %s from %s: %v", targetChunk.FileId, h.addr, ferr)
+			continue
+		}
+		if fh.wfs.chunkCache != nil {
+			fh.wfs.chunkCache.SetChunk(targetChunk.FileId, data)
+		}
+		if fh.wfs.peerAnnouncer != nil {
+			fh.wfs.peerAnnouncer.EnqueueAnnounce(targetChunk.FileId)
+		}
+		if chunkOffset >= int64(len(data)) {
+			return 0, 0, fmt.Errorf("peer returned short chunk")
+		}
+		// Cap the copy to whichever is smaller: the caller's buffer, or
+		// the remaining bytes before logical EOF (readStop already
+		// clamped to fileSize). FileChunk.Size can legitimately exceed
+		// the logical file length when the last chunk is partially
+		// written and the filer stored the full padded buffer, so a
+		// naïve copy(buff, data[chunkOffset:]) would return bytes past
+		// EOF. The caller treats a short non-error return as success
+		// and zero-fills the tail, so returning too many bytes here is
+		// a correctness bug, not just wasted I/O.
+		remaining := readStop - offset
+		available := int64(len(data)) - chunkOffset
+		maxCopy := int64(len(buff))
+		if remaining < maxCopy {
+			maxCopy = remaining
+		}
+		if available < maxCopy {
+			maxCopy = available
+		}
+		copied := copy(buff[:maxCopy], data[chunkOffset:chunkOffset+maxCopy])
+		return int64(copied), targetChunk.ModifiedTsNs, nil
+	}
+	return 0, 0, fmt.Errorf("no peer served fid %s", targetChunk.FileId)
+}
+
+func findChunkContaining(chunks []*filer_pb.FileChunk, offset int64) (*filer_pb.FileChunk, int64) {
+	for _, c := range chunks {
+		start := c.Offset
+		stop := c.Offset + int64(c.Size)
+		if offset >= start && offset < stop {
+			return c, offset - start
+		}
+	}
+	return nil, 0
+}
+
+// peerHolder is a holder entry carried through the fetcher: addr for
+// dialing plus the DC/Rack labels the owner recorded at announce time, so
+// the fetcher can re-rank by its own locality before dialing.
+type peerHolder struct {
+	addr string
+	dc   string
+	rack string
+}
+
+// peerLookupHolders calls ChunkLookup on the given HRW owner and returns
+// the holders in the server-reported (LRU) order along with their locality
+// labels. The pooled dialer is passed in from the caller.
+func peerLookupHolders(ctx context.Context, dial MountPeerDialer, ownerAddr, fid string) ([]peerHolder, error) {
+	client, closeFn, err := dial(ctx, ownerAddr)
+	if err != nil {
+		return nil, err
+	}
+	defer closeFn()
+
+	callCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout)
+	defer cancel()
+	resp, err := client.ChunkLookup(callCtx, &mount_peer_pb.ChunkLookupRequest{FileIds: []string{fid}})
+	if err != nil {
+		return nil, err
+	}
+	set, ok := resp.PeersByFid[fid]
+	if !ok || set == nil {
+		return nil, nil
+	}
+	out := make([]peerHolder, 0, len(set.Peers))
+	for _, p := range set.Peers {
+		out = append(out, peerHolder{addr: p.PeerAddr, dc: p.DataCenter, rack: p.Rack})
+	}
+	return out, nil
+}
+
+// localityBucket scores how "close" a peer is to self. Lower is better:
+//
+//	0 = same rack in same DC      (shortest hop)
+//	1 = same DC, different rack   (cross-rack but still intra-DC)
+//	2 = different DC, or unknown  (anything else)
+//
+// Missing labels on either side fall to bucket 2 — we don't claim locality
+// we can't prove.
+func localityBucket(selfDC, selfRack, peerDC, peerRack string) int {
+	if selfDC == "" || peerDC == "" || selfDC != peerDC {
+		return 2
+	}
+	if selfRack != "" && peerRack != "" && selfRack == peerRack {
+		return 0
+	}
+	return 1
+}
+
+// sortHoldersByLocality stable-sorts holders so the most local peers are
+// tried first. The server-returned order is LRU (freshest holder first),
+// and sort.SliceStable preserves that ordering within each locality
+// bucket, so among equally-local peers we still prefer the freshest.
+func sortHoldersByLocality(holders []peerHolder, selfDC, selfRack string) {
+	if len(holders) < 2 {
+		return
+	}
+	sort.SliceStable(holders, func(i, j int) bool {
+		return localityBucket(selfDC, selfRack, holders[i].dc, holders[i].rack) <
+			localityBucket(selfDC, selfRack, holders[j].dc, holders[j].rack)
+	})
+}
+
+// fetchChunkFromPeer server-streams a chunk from the given peer and
+// verifies the MD5 of the assembled bytes against expectedETag.
+//
+// expectedSize (when > 0) is the authoritative chunk length from the
+// filer entry; we pre-allocate exactly that to avoid slice growth and
+// reject streams that overshoot it. A zero expectedSize falls back to
+// maxPeerFetchChunkBytes as a safety ceiling.
+func fetchChunkFromPeer(ctx context.Context, dial MountPeerDialer, peerAddr, fid string, expectedSize uint64, expectedETag string) ([]byte, error) {
+	client, closeFn, err := dial(ctx, peerAddr)
+	if err != nil {
+		return nil, err
+	}
+	defer closeFn()
+
+	callCtx, cancel := context.WithTimeout(ctx, peerFetchTimeout)
+	defer cancel()
+
+	stream, err := client.FetchChunk(callCtx, &mount_peer_pb.FetchChunkRequest{
+		FileId:       fid,
+		ExpectedEtag: expectedETag,
+		ExpectedSize: expectedSize,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// capHint pre-sizes the assembly buffer from the filer-reported chunk
+	// size. When the caller didn't know the size (expectedSize == 0), we
+	// let append grow the buffer rather than reserve the 64 MiB ceiling —
+	// typical chunks are a few MiB, and the maxPeerFetchChunkBytes check
+	// during Recv is the real safety ceiling.
+	capHint := expectedSize
+	if capHint > maxPeerFetchChunkBytes {
+		capHint = maxPeerFetchChunkBytes
+	}
+	buf := make([]byte, 0, capHint)
+
+	for {
+		resp, rerr := stream.Recv()
+		if rerr == io.EOF {
+			break
+		}
+		if rerr != nil {
+			if status.Code(rerr) == codes.NotFound {
+				return nil, fmt.Errorf("peer not cached")
+			}
+			return nil, rerr
+		}
+		if len(buf)+len(resp.Data) > int(maxPeerFetchChunkBytes) {
+			return nil, fmt.Errorf("peer response exceeds max chunk size %d", maxPeerFetchChunkBytes)
+		}
+		buf = append(buf, resp.Data...)
+	}
+
+	if expectedSize > 0 && uint64(len(buf)) != expectedSize {
+		return nil, fmt.Errorf("peer returned %d bytes, expected %d", len(buf), expectedSize)
+	}
+
+	// Per-chunk integrity check — the peer is treated as untrusted.
+	// FileChunk.ETag is UploadResult.ContentMd5, which is base64 of the
+	// raw 16-byte MD5 (see weed/operation/upload_content.go:64 and
+	// filer's ETagChunks which decodes via util.Base64Md5ToBytes).
+	// Compare raw digests so we don't reject every valid peer response
+	// because of an encoding mismatch.
+	if expectedETag != "" {
+		got := md5.Sum(buf)
+		if !etagMatchesMD5(expectedETag, got[:]) {
+			return nil, fmt.Errorf("etag mismatch: peer=%s got=%x want=%s", peerAddr, got, expectedETag)
+		}
+	}
+	return buf, nil
+}
+
+// etagMatchesMD5 compares a stored FileChunk.ETag string to a raw 16-byte
+// MD5 digest. FileChunk.ETag is produced as base64(md5) on upload, but
+// some older chunks (and tests) use hex(md5). Accept either.
+func etagMatchesMD5(etag string, rawMD5 []byte) bool {
+	if len(rawMD5) != md5.Size {
+		return false
+	}
+	if dec := util.Base64Md5ToBytes(etag); len(dec) == md5.Size {
+		return bytes.Equal(dec, rawMD5)
+	}
+	if dec, err := hex.DecodeString(etag); err == nil && len(dec) == md5.Size {
+		return bytes.Equal(dec, rawMD5)
+	}
+	return false
+}

weed/mount/peer_fetcher_test.go

@@ -0,0 +1,180 @@
+package mount
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/base64"
+	"encoding/hex"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// newTestFetchServer stands up a minimal MountPeer gRPC server that only
+// serves FetchChunk out of a fakeChunkCache. Goes through the same
+// PeerGrpcServer.Start path production uses (via pb.NewGrpcServer with
+// the standard keepalive + msg-size options) so future default server
+// options are exercised by these tests too.
+func newTestFetchServer(t *testing.T, cache *fakeChunkCache) (addr string, stop func()) {
+	t.Helper()
+	dir := NewPeerDirectory()
+	srv := NewPeerGrpcServer(cache, dir, nil, "")
+	if err := srv.Start("127.0.0.1:0"); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	return srv.Addr(), func() { srv.Stop() }
+}
+
+func etagOf(b []byte) string {
+	sum := md5.Sum(b)
+	return hex.EncodeToString(sum[:])
+}
+
+// etagOfBase64 mirrors how the real upload path produces FileChunk.ETag:
+// base64 of the raw 16-byte MD5 digest. Kept alongside the hex form so
+// we can test that both encodings work end-to-end.
+func etagOfBase64(b []byte) string {
+	sum := md5.Sum(b)
+	return base64.StdEncoding.EncodeToString(sum[:])
+}
+
+func TestFetchChunkFromPeer_Hit(t *testing.T) {
+	cache := newFakeChunkCache()
+	payload := []byte("hello from peer stream")
+	cache.Put("3,abc", payload)
+	addr, stop := newTestFetchServer(t, cache)
+	defer stop()
+
+	dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials()))
+	got, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,abc", uint64(len(payload)), etagOf(payload))
+	if err != nil {
+		t.Fatalf("fetch: %v", err)
+	}
+	if string(got) != string(payload) {
+		t.Errorf("bytes mismatch: got %q want %q", got, payload)
+	}
+}
+
+// TestFetchChunkFromPeer_Base64Etag mirrors the real upload path where
+// FileChunk.ETag is base64(md5) rather than hex(md5). A regression
+// there would reject every valid peer response in production.
+func TestFetchChunkFromPeer_Base64Etag(t *testing.T) {
+	cache := newFakeChunkCache()
+	payload := []byte("base64-etag peer bytes")
+	cache.Put("3,b64", payload)
+	addr, stop := newTestFetchServer(t, cache)
+	defer stop()
+
+	dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials()))
+	got, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,b64", uint64(len(payload)), etagOfBase64(payload))
+	if err != nil {
+		t.Fatalf("fetch: %v", err)
+	}
+	if string(got) != string(payload) {
+		t.Errorf("bytes mismatch: got %q want %q", got, payload)
+	}
+}
+
+func TestFetchChunkFromPeer_EtagMismatch(t *testing.T) {
+	cache := newFakeChunkCache()
+	payload := []byte("unexpected bytes")
+	cache.Put("3,abc", payload)
+	addr, stop := newTestFetchServer(t, cache)
+	defer stop()
+
+	dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials()))
+	_, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,abc", 0, "not-the-real-etag")
+	if err == nil {
+		t.Fatalf("expected etag mismatch error, got nil")
+	}
+}
+
+func TestFetchChunkFromPeer_NotFound(t *testing.T) {
+	cache := newFakeChunkCache()
+	addr, stop := newTestFetchServer(t, cache)
+	defer stop()
+
+	dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials()))
+	_, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,missing", 0, "")
+	if err == nil {
+		t.Errorf("expected error for missing fid, got nil")
+	}
+}
+
+// TestSortHoldersByLocality verifies that same-rack peers sort ahead of
+// same-DC-different-rack peers, which in turn sort ahead of cross-DC
+// peers, and that LRU order is preserved within each bucket.
+func TestSortHoldersByLocality(t *testing.T) {
+	selfDC, selfRack := "dc1", "r1"
+	// Input order mimics a server-side LRU list (newest first).
+	holders := []peerHolder{
+		{addr: "far-newest", dc: "dc2", rack: "r9"},        // bucket 2 (diff DC)
+		{addr: "mid-newer", dc: "dc1", rack: "r2"},         // bucket 1 (same DC, diff rack)
+		{addr: "local-newer", dc: "dc1", rack: "r1"},       // bucket 0 (same rack)
+		{addr: "far-older", dc: "dc2", rack: "r9"},         // bucket 2
+		{addr: "mid-older", dc: "dc1", rack: "r2"},         // bucket 1
+		{addr: "local-older", dc: "dc1", rack: "r1"},       // bucket 0
+		{addr: "unlabeled-older", dc: "", rack: ""},        // bucket 2 (unknown)
+	}
+
+	sortHoldersByLocality(holders, selfDC, selfRack)
+
+	want := []string{
+		"local-newer", "local-older",
+		"mid-newer", "mid-older",
+		"far-newest", "far-older", "unlabeled-older",
+	}
+	if len(holders) != len(want) {
+		t.Fatalf("len got %d want %d", len(holders), len(want))
+	}
+	for i, w := range want {
+		if holders[i].addr != w {
+			t.Errorf("pos %d: got %q want %q (full: %+v)", i, holders[i].addr, w, holders)
+		}
+	}
+}
+
+// TestSortHoldersByLocality_NoSelfLabels — when the caller has no DC/rack
+// labels, every peer falls to bucket 2 and the server-returned LRU order
+// must pass through unchanged.
+func TestSortHoldersByLocality_NoSelfLabels(t *testing.T) {
+	holders := []peerHolder{
+		{addr: "a", dc: "dc1", rack: "r1"},
+		{addr: "b", dc: "dc2", rack: "r2"},
+		{addr: "c", dc: "", rack: ""},
+	}
+	sortHoldersByLocality(holders, "", "")
+	want := []string{"a", "b", "c"}
+	for i, w := range want {
+		if holders[i].addr != w {
+			t.Errorf("pos %d: got %q want %q", i, holders[i].addr, w)
+		}
+	}
+}
+
+func TestFetchChunkFromPeer_MultiFrameChunkAssembledCorrectly(t *testing.T) {
+	cache := newFakeChunkCache()
+	// Just over 2× the stream frame size so we get at least three frames.
+	payload := make([]byte, fetchChunkStreamSize*2+42)
+	for i := range payload {
+		payload[i] = byte((i * 7) & 0xff)
+	}
+	cache.Put("3,large", payload)
+	addr, stop := newTestFetchServer(t, cache)
+	defer stop()
+
+	dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials()))
+	got, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,large", uint64(len(payload)), etagOf(payload))
+	if err != nil {
+		t.Fatalf("fetch: %v", err)
+	}
+	if len(got) != len(payload) {
+		t.Fatalf("len got %d want %d", len(got), len(payload))
+	}
+	for i := range payload {
+		if got[i] != payload[i] {
+			t.Fatalf("mismatch at offset %d", i)
+		}
+	}
+}

weed/mount/peer_grpc.go

@@ -108,6 +108,10 @@ func (s *PeerGrpcServer) Addr() string {
 	return s.listener.Addr().String()
 }
 
+// SelfAddr returns the advertise address this server was constructed with —
+// the identity the fetcher compares against to avoid dialing itself.
+func (s *PeerGrpcServer) SelfAddr() string { return s.selfAddr }
+
 // ChunkAnnounce accepts holder entries for fids this mount owns; rejects
 // others so the caller can retry against the correct owner.
 func (s *PeerGrpcServer) ChunkAnnounce(ctx context.Context, req *mount_peer_pb.ChunkAnnounceRequest) (*mount_peer_pb.ChunkAnnounceResponse, error) {