Recover from a dead volume server on the mount read path (#10798)

* mount: re-resolve volume locations after a failed chunk read NewChunkGroup passed nil as the ReaderCache's CacheInvalidator, so retryFetchAfterCacheInvalidation was dead code on the FUSE read path. A mount that cached a volume's locations while one server was down kept retrying that server after it died, then returned EIO, even though the master and filer both resolved the live replica. The S3 gateway already passes its filerClient; do the same for the mount. * test: FUSE integration tests for volume server failover One mount appends while a second tails, and a volume server is killed, started or restarted mid-stream against a 001-replicated cluster of three volume servers. Automates the scenario matrix reported for Docker Swarm mounts, including the large-file variant and a no-chaos control. * test: report the filer's own view when append content mismatches A mismatch between what the writer wrote and what the reader sees can come from either side's cache. Read the file back through the filer's HTTP handler as well, and let the mount verbosity be raised from the environment, so a failing run says which layer lost the data. * test: wait for the reader mount to converge before comparing A mount caches metadata for about a second, so reading the file the instant the writer's last close returned can legitimately come back short. Poll the reader until it matches or the timeout expires; content that is wrong rather than merely late never converges and still fails, now with the writer's mount and the filer's own view alongside it. * test: detect a failover cluster child that exited at startup Signal(0) succeeds for a zombie and nothing reaped these children until shutdown, so a process that died on startup looked alive until the readiness timeout expired. Reap each child as it is started and consult the result. * test: read a file the killed volume server actually holds Placement decides which two of three servers back each volume, so killing volume N and reading readfile-N could pass without the victim ever holding a replica of it. Resolve each file's volumes through the filer and the master, and pick one the victim backs, preferring a file the reader has not cached. * ci: stop persisting checkout credentials in the failover workflow The job does not use the token after cloning. Also tag the README's command block as bash and match the timeout the workflow actually uses. * test: discard the ignored errors errcheck flags in the failover harness * test: resolve manifests when mapping a file to its volumes A manifest chunk's own fid names the volume holding the manifest, not the volumes holding the data, so a large enough file would point the failover victim at the wrong server. * test: pin the stale-location recovery path with a primed reader Reading a file for the first time after a server dies proves nothing: the lookup is fresh and returns the survivor. Kill one holder and wait for the master to drop it, read a file on that volume so the reader caches the lone survivor, restart the first server, then kill the survivor. The reader's only cached location is now dead while the data is live elsewhere, which is the case the invalidator exists for: EIO without it, recovery with it.
2026-08-18 13:17:08 +00:00 · 2026-08-17 17:30:33 -07:00
parent 6fda8c67f3
commit 1ddec72707
8 changed files with 1082 additions and 5 deletions
@@ -0,0 +1,75 @@
+name: "FUSE Volume Server Failover Tests"
+
+on:
+  pull_request:
+    paths:
+      - 'weed/command/mount*.go'
+      - 'weed/mount/**'
+      - 'weed/filer/**'
+      - 'weed/wdclient/**'
+      - 'weed/operation/upload_content.go'
+      - 'test/fuse_failover/**'
+      - '.github/workflows/fuse-failover.yml'
+      - '.github/actions/fix-fusermount-setuid/**'
+  push:
+    branches: [master]
+    paths:
+      - 'weed/command/mount*.go'
+      - 'weed/mount/**'
+      - 'weed/filer/**'
+      - 'weed/wdclient/**'
+      - 'weed/operation/upload_content.go'
+      - 'test/fuse_failover/**'
+      - '.github/workflows/fuse-failover.yml'
+      - '.github/actions/fix-fusermount-setuid/**'
+
+concurrency:
+  group: ${{ github.head_ref || github.ref }}/fuse-failover
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  fuse-failover:
+    name: FUSE Volume Server Failover
+    runs-on: ubuntu-22.04
+    timeout-minutes: 40
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v7
+        with:
+          persist-credentials: false
+
+      - name: Set up Go
+        uses: actions/setup-go@v7
+        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: Repair the fusermount3 setuid bit
+        uses: ./.github/actions/fix-fusermount-setuid
+
+      - name: Build SeaweedFS
+        run: go build -o weed/weed -buildvcs=false ./weed
+
+      - name: Run failover integration tests
+        timeout-minutes: 35
+        env:
+          WEED_BINARY: ${{ github.workspace }}/weed/weed
+        run: go test -v -count=1 -timeout=30m ./test/fuse_failover/...
+
+      - name: Upload logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: fuse-failover-test-logs
+          path: /tmp/seaweedfs-fuse-failover-logs/
+          retention-days: 3
@@ -0,0 +1,35 @@
+# FUSE volume server failover tests
+
+Integration tests for what happens to FUSE mounts when a volume server goes
+away, comes back, or restarts underneath in-flight IO. They automate the manual
+matrix reported in
+[discussion #10206](https://github.com/seaweedfs/seaweedfs/discussions/10206):
+one mount appends to a file while a second mount tails it, and a volume server
+is stopped, started or restarted mid-stream.
+
+The cluster is 1 master (`-defaultReplication=001`), 3 volume servers, 1 filer
+and 2 mounts, all as local processes. With 001 every chunk has a copy on two of
+the three servers, so losing any single server must be invisible to both mounts.
+
+| Reported scenario | Test |
+| --- | --- |
+| control: append + tail with nothing failing | `TestAppendWithoutChaos` |
+| read a file while one volume server is down | `TestReadWithVolumeServerDown` |
+| "STOP volumes": append + tail, kill a server mid-stream | `TestAppendWhileVolumeServerStops` |
+| "Start volumes": append + tail with a server down, start it mid-stream | `TestAppendWhileVolumeServerStarts` |
+| "Re-start volumes": append + tail, restart a server mid-stream | `TestAppendWhileVolumeServerRestarts` |
+| part 2: large file copy instead of small appends | `TestLargeWriteWhileVolumeServerStops` |
+
+Volume servers are dropped with SIGKILL, the closest local equivalent of a Swarm
+task disappearing from the overlay network: no deregistration, and the address
+stops answering.
+
+## Running
+
+```bash
+go build -o weed/weed ./weed
+WEED_BINARY=$PWD/weed/weed go test -v -count=1 -timeout=30m ./test/fuse_failover/...
+```
+
+Needs FUSE and, on Linux, a working `/dev/fuse`. Logs from a failed run are
+copied to `/tmp/seaweedfs-fuse-failover-logs/`.
@@ -0,0 +1,537 @@
+//go:build linux || darwin
+
+package fuse_failover
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/stretchr/testify/require"
+)
+
+// failoverCluster runs 1 master, N volume servers, 1 filer and M FUSE mounts,
+// with the master defaulting to 001 replication so every chunk lands on two
+// distinct volume servers. Individual volume servers can be stopped and started
+// while IO is in flight, which is what the Docker Swarm reports in discussion
+// 10206 exercise: a volume server disappears mid-append and the mounts must
+// keep reading from the surviving replica and keep writing on another volume.
+type failoverCluster struct {
+	t          testing.TB
+	baseDir    string
+	weedBinary string
+
+	masterPort     int
+	masterGrpcPort int
+	filerPort      int
+	filerGrpcPort  int
+	volumePorts    []int
+	volumeGrpcPort []int
+	volumeDirs     []string
+
+	masterCmd   *exec.Cmd
+	filerCmd    *exec.Cmd
+	volumeCmds  []*exec.Cmd
+	mountCmds   []*exec.Cmd
+	mountPoints []string
+	logFiles    []*os.File
+
+	mu          sync.Mutex
+	waits       map[*exec.Cmd]chan error
+	cleanupOnce sync.Once
+}
+
+func startFailoverCluster(t testing.TB, numVolumes, numMounts int) *failoverCluster {
+	require.GreaterOrEqual(t, numVolumes, 2, "001 replication needs at least 2 volume servers")
+	require.GreaterOrEqual(t, numMounts, 1)
+
+	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_failover_")
+	require.NoError(t, err)
+
+	c := &failoverCluster{
+		t:              t,
+		baseDir:        baseDir,
+		weedBinary:     binary,
+		volumePorts:    make([]int, numVolumes),
+		volumeGrpcPort: make([]int, numVolumes),
+		volumeDirs:     make([]string, numVolumes),
+		volumeCmds:     make([]*exec.Cmd, numVolumes),
+		mountCmds:      make([]*exec.Cmd, numMounts),
+		mountPoints:    make([]string, numMounts),
+	}
+	t.Cleanup(c.Stop)
+
+	ports, err := testutil.AllocatePorts(4 + 2*numVolumes)
+	require.NoError(t, err)
+	c.masterPort, c.masterGrpcPort = ports[0], ports[1]
+	c.filerPort, c.filerGrpcPort = ports[2], ports[3]
+	for i := 0; i < numVolumes; i++ {
+		c.volumePorts[i] = ports[4+2*i]
+		c.volumeGrpcPort[i] = ports[5+2*i]
+		c.volumeDirs[i] = filepath.Join(baseDir, fmt.Sprintf("volume%d", i))
+		require.NoError(t, os.MkdirAll(c.volumeDirs[i], 0755))
+	}
+
+	require.NoError(t, c.startMaster())
+	require.NoError(t, c.waitForTCP(c.masterCmd, "master",
+		fmt.Sprintf("127.0.0.1:%d", c.masterPort), 30*time.Second))
+
+	for i := 0; i < numVolumes; i++ {
+		require.NoError(t, c.StartVolume(i))
+	}
+
+	require.NoError(t, c.startFiler())
+	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))
+		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 *failoverCluster) MountDir(i int) string { return c.mountPoints[i] }
+
+func (c *failoverCluster) Stop() {
+	if c == nil {
+		return
+	}
+	c.cleanupOnce.Do(func() {
+		for i := len(c.mountCmds) - 1; i >= 0; i-- {
+			c.stopCmd(c.mountCmds[i], syscall.SIGTERM)
+			_ = exec.Command("fusermount3", "-u", c.mountPoints[i]).Run()
+			_ = exec.Command("fusermount", "-u", c.mountPoints[i]).Run()
+		}
+		c.stopCmd(c.filerCmd, syscall.SIGTERM)
+		for i := len(c.volumeCmds) - 1; i >= 0; i-- {
+			c.stopCmd(c.volumeCmds[i], syscall.SIGTERM)
+		}
+		c.stopCmd(c.masterCmd, syscall.SIGTERM)
+
+		c.mu.Lock()
+		for _, f := range c.logFiles {
+			_ = f.Close()
+		}
+		c.mu.Unlock()
+		c.copyLogsForCI()
+		if !c.t.Failed() {
+			os.RemoveAll(c.baseDir)
+		}
+	})
+}
+
+// KillVolume drops a volume server without letting it deregister, the closest
+// local equivalent of a Swarm task vanishing from the overlay network.
+func (c *failoverCluster) KillVolume(i int) {
+	c.stopCmd(c.volumeCmds[i], syscall.SIGKILL)
+	c.volumeCmds[i] = nil
+}
+
+// StartVolume (re)starts volume server i on its original ports and data dir.
+func (c *failoverCluster) StartVolume(i int) error {
+	cmd := 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.volumePorts[i]),
+		"-port.grpc="+strconv.Itoa(c.volumeGrpcPort[i]),
+		"-master="+c.masterAddress(),
+		"-dir="+c.volumeDirs[i],
+		"-dataCenter=dc1",
+		"-rack=rack1",
+		"-max=10",
+	)
+	c.volumeCmds[i] = cmd
+	if err := c.startCmd(cmd, fmt.Sprintf("volume%d", i)); err != nil {
+		return err
+	}
+	return c.waitForTCP(cmd, fmt.Sprintf("volume%d", i),
+		fmt.Sprintf("127.0.0.1:%d", c.volumePorts[i]), 30*time.Second)
+}
+
+func (c *failoverCluster) startMaster() 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"),
+		"-defaultReplication=001",
+		"-volumeSizeLimitMB=64",
+	)
+	return c.startCmd(c.masterCmd, "master")
+}
+
+func (c *failoverCluster) startFiler() 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(),
+		"-defaultReplicaPlacement=001",
+		"-defaultStoreDir="+filerDir,
+	)
+	return c.startCmd(c.filerCmd, "filer")
+}
+
+func (c *failoverCluster) startMount(idx int) 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)
+	}
+	// Chunk-level detail needs -v=4; keep CI at -v=2 so the logs stay small.
+	verbosity := os.Getenv("FUSE_FAILOVER_MOUNT_V")
+	if verbosity == "" {
+		verbosity = "2"
+	}
+	c.mountCmds[idx] = exec.Command(c.weedBinary,
+		"-logdir="+filepath.Join(c.baseDir, "logs"),
+		"-v="+verbosity,
+		"mount",
+		"-filer="+c.filerAddress(),
+		"-dir="+c.mountPoints[idx],
+		"-filer.path=/",
+		"-dirAutoCreate",
+		"-allowOthers=false",
+		"-replication=001",
+		"-cacheDir="+cacheDir,
+	)
+	return c.startCmd(c.mountCmds[idx], fmt.Sprintf("mount%d", idx))
+}
+
+// MasterGet fetches a master HTTP endpoint, e.g. "/dir/status?pretty=y" or
+// "/dir/lookup?volumeId=6", so a failing test can show where the replicas are.
+func (c *failoverCluster) MasterGet(path string) string {
+	resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d%s", c.masterPort, path))
+	if err != nil {
+		return fmt.Sprintf("(master %s failed: %v)", path, err)
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Sprintf("(master %s read failed: %v)", path, err)
+	}
+	return string(body)
+}
+
+// FilerGet reads a file back through the filer's own HTTP handler: a view of
+// the chunk list that neither mount's cache can colour.
+func (c *failoverCluster) FilerGet(path string) ([]byte, error) {
+	resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d%s", c.filerPort, path))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("filer %s: %s", path, resp.Status)
+	}
+	return body, nil
+}
+
+// VolumeServerAddress is the address volume server i registers with the master.
+func (c *failoverCluster) VolumeServerAddress(i int) string {
+	return fmt.Sprintf("127.0.0.1:%d", c.volumePorts[i])
+}
+
+// FileVolumeIds returns the volume ids backing a file, read from the filer's
+// own entry rather than inferred, so a test can tell which servers a given
+// file actually depends on. Manifests are resolved first: a manifest chunk's
+// own fid names the volume holding the manifest, not the data.
+func (c *failoverCluster) FileVolumeIds(path string) ([]uint32, error) {
+	body, err := c.FilerGet(path + "?metadata=true&resolveManifest=true")
+	if err != nil {
+		return nil, err
+	}
+	var entry struct {
+		Chunks []struct {
+			FileId string `json:"file_id"`
+			Fid    struct {
+				VolumeId uint32 `json:"volume_id"`
+			} `json:"fid"`
+		} `json:"chunks"`
+	}
+	if err = json.Unmarshal(body, &entry); err != nil {
+		return nil, fmt.Errorf("decode entry %s: %w", path, err)
+	}
+	seen := make(map[uint32]bool)
+	var vids []uint32
+	for _, chunk := range entry.Chunks {
+		vid := chunk.Fid.VolumeId
+		if vid == 0 && chunk.FileId != "" {
+			parsed, parseErr := strconv.ParseUint(strings.SplitN(chunk.FileId, ",", 2)[0], 10, 32)
+			if parseErr != nil {
+				return nil, fmt.Errorf("parse file id %s: %w", chunk.FileId, parseErr)
+			}
+			vid = uint32(parsed)
+		}
+		if !seen[vid] {
+			seen[vid] = true
+			vids = append(vids, vid)
+		}
+	}
+	return vids, nil
+}
+
+// VolumeHolders returns the volume server addresses the master currently lists
+// for a volume id.
+func (c *failoverCluster) VolumeHolders(vid uint32) ([]string, error) {
+	var lookup struct {
+		Locations []struct {
+			Url string `json:"url"`
+		} `json:"locations"`
+	}
+	body := c.MasterGet(fmt.Sprintf("/dir/lookup?volumeId=%d", vid))
+	if err := json.Unmarshal([]byte(body), &lookup); err != nil {
+		return nil, fmt.Errorf("decode lookup for volume %d: %w (%s)", vid, err, body)
+	}
+	holders := make([]string, 0, len(lookup.Locations))
+	for _, loc := range lookup.Locations {
+		holders = append(holders, loc.Url)
+	}
+	return holders, nil
+}
+
+// WaitForHolders polls the master until it lists exactly count servers for a
+// volume. The master only drops a dead node after three missed heartbeats, so a
+// test that depends on the cluster's view having caught up has to wait for it.
+func (c *failoverCluster) WaitForHolders(vid uint32, count int, timeout time.Duration) ([]string, error) {
+	deadline := time.Now().Add(timeout)
+	for {
+		holders, err := c.VolumeHolders(vid)
+		if err == nil && len(holders) == count {
+			return holders, nil
+		}
+		if time.Now().After(deadline) {
+			return holders, fmt.Errorf("volume %d still has %d holders (%v), want %d", vid, len(holders), holders, count)
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+}
+
+// volumeIndexOf maps a server address back to its index, or -1.
+func (c *failoverCluster) volumeIndexOf(address string) int {
+	for i := range c.volumePorts {
+		if c.VolumeServerAddress(i) == address {
+			return i
+		}
+	}
+	return -1
+}
+
+// FileIsOn reports whether any of path's chunks live on the given volume
+// server, i.e. whether taking that server down actually costs this file a
+// replica.
+func (c *failoverCluster) FileIsOn(path, serverAddress string) (bool, error) {
+	vids, err := c.FileVolumeIds(path)
+	if err != nil {
+		return false, err
+	}
+	for _, vid := range vids {
+		holders, holdersErr := c.VolumeHolders(vid)
+		if holdersErr != nil {
+			return false, holdersErr
+		}
+		for _, holder := range holders {
+			if holder == serverAddress {
+				return true, nil
+			}
+		}
+	}
+	return false, nil
+}
+
+func (c *failoverCluster) masterAddress() string {
+	return string(pb.NewServerAddress("127.0.0.1", c.masterPort, c.masterGrpcPort))
+}
+
+func (c *failoverCluster) filerAddress() string {
+	return string(pb.NewServerAddress("127.0.0.1", c.filerPort, c.filerGrpcPort))
+}
+
+func (c *failoverCluster) 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)
+	}
+	// Append so a restarted volume server keeps the log of its earlier run.
+	logFile, err := os.OpenFile(filepath.Join(logPath, name+".log"),
+		os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
+	if err != nil {
+		return err
+	}
+	c.mu.Lock()
+	c.logFiles = append(c.logFiles, logFile)
+	c.mu.Unlock()
+	cmd.Stdout = logFile
+	cmd.Stderr = logFile
+	if err := cmd.Start(); err != nil {
+		return err
+	}
+	// Reap in the background and publish the result: Signal(0) succeeds for a
+	// zombie, so an unreaped child that died at startup would otherwise look
+	// alive until the readiness timeout expired.
+	ch := make(chan error, 1)
+	c.mu.Lock()
+	if c.waits == nil {
+		c.waits = make(map[*exec.Cmd]chan error)
+	}
+	c.waits[cmd] = ch
+	c.mu.Unlock()
+	go func() {
+		ch <- cmd.Wait()
+		close(ch)
+	}()
+	return nil
+}
+
+// waitChan returns the channel carrying cmd's exit, or nil if it was never
+// started through startCmd. It stays readable after the exit is consumed.
+func (c *failoverCluster) waitChan(cmd *exec.Cmd) chan error {
+	if cmd == nil {
+		return nil
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.waits[cmd]
+}
+
+func (c *failoverCluster) 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 *failoverCluster) copyLogsForCI() {
+	// One directory per test: subtests share a log dir name otherwise, and the
+	// last one to finish would overwrite the logs of the one that failed.
+	ciLogDir := filepath.Join("/tmp/seaweedfs-fuse-failover-logs",
+		strings.ReplaceAll(c.t.Name(), "/", "_"))
+	os.MkdirAll(ciLogDir, 0755)
+	entries, err := os.ReadDir(filepath.Join(c.baseDir, "logs"))
+	if err != nil {
+		return
+	}
+	for _, e := range entries {
+		data, err := os.ReadFile(filepath.Join(c.baseDir, "logs", e.Name()))
+		if err != nil {
+			continue
+		}
+		os.WriteFile(filepath.Join(ciLogDir, e.Name()), data, 0644)
+	}
+}
+
+func (c *failoverCluster) 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 ch := c.waitChan(cmd); ch != nil {
+			select {
+			case waitErr := <-ch:
+				return fmt.Errorf("%s exited before listening on %s: %v\n%s",
+					name, addr, waitErr, c.tailLog(name))
+			default:
+			}
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	return fmt.Errorf("service at %s not ready within timeout\n%s", addr, c.tailLog(name))
+}
+
+func (c *failoverCluster) 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
+		}
+		if parentStat.Sys().(*syscall.Stat_t).Dev != mountStat.Sys().(*syscall.Stat_t).Dev {
+			return nil
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	return fmt.Errorf("mount point %s not ready within timeout (FUSE not detected)", mountPoint)
+}
+
+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 signals cmd and waits for the reaper goroutine started by startCmd to
+// report its exit, escalating to SIGKILL if it does not go quietly.
+func (c *failoverCluster) stopCmd(cmd *exec.Cmd, sig syscall.Signal) {
+	if cmd == nil || cmd.Process == nil {
+		return
+	}
+	_ = cmd.Process.Signal(sig)
+	done := c.waitChan(cmd)
+	if done == nil {
+		return
+	}
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		_ = cmd.Process.Signal(syscall.SIGKILL)
+		<-done
+	}
+}
@@ -0,0 +1,419 @@
+//go:build linux || darwin
+
+package fuse_failover
+
+import (
+	"bytes"
+	"crypto/rand"
+	"crypto/sha256"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// The scenarios below come from the Docker Swarm report in discussion 10206:
+// one mount appends to a file while a second mount reads it, and a volume
+// server is stopped, started or restarted underneath. With 001 replication a
+// single volume server loss must never surface as EIO on either side.
+
+const (
+	appendLines      = 200
+	maxAppendLatency = 10 * time.Second
+	convergeTimeout  = 30 * time.Second
+)
+
+type appendResult struct {
+	errs       []error
+	maxLatency time.Duration
+	slowest    int
+}
+
+// appendLoop mimics `for i in ...; do echo $i >> file; done`: every line is its
+// own open/write/close, so every line forces a flush and a chunk upload.
+func appendLoop(path string, lines int, onLine func(i int)) *appendResult {
+	res := &appendResult{}
+	for i := 0; i < lines; i++ {
+		start := time.Now()
+		err := appendOnce(path, fmt.Sprintf("%07d\n", i))
+		elapsed := time.Since(start)
+		if elapsed > res.maxLatency {
+			res.maxLatency, res.slowest = elapsed, i
+		}
+		if err != nil {
+			res.errs = append(res.errs, fmt.Errorf("line %d: %w", i, err))
+		}
+		if onLine != nil {
+			onLine(i)
+		}
+	}
+	return res
+}
+
+func appendOnce(path, line string) error {
+	f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0644)
+	if err != nil {
+		return fmt.Errorf("open: %w", err)
+	}
+	if _, err = f.WriteString(line); err != nil {
+		f.Close()
+		return fmt.Errorf("write: %w", err)
+	}
+	if err = f.Close(); err != nil {
+		return fmt.Errorf("close: %w", err)
+	}
+	return nil
+}
+
+// reader is the `tail -f` side: it keeps re-reading the whole file from the
+// other mount and records every failure that is not "not created yet".
+type reader struct {
+	stop chan struct{}
+	done chan struct{}
+	errs []error
+}
+
+func startReader(path string) *reader {
+	r := &reader{stop: make(chan struct{}), done: make(chan struct{})}
+	go func() {
+		defer close(r.done)
+		for {
+			select {
+			case <-r.stop:
+				return
+			default:
+			}
+			if _, err := os.ReadFile(path); err != nil && !os.IsNotExist(err) {
+				r.errs = append(r.errs, err)
+			}
+			time.Sleep(200 * time.Millisecond)
+		}
+	}()
+	return r
+}
+
+func (r *reader) Stop() []error {
+	close(r.stop)
+	<-r.done
+	return r.errs
+}
+
+// waitForContent re-reads path until it matches want or the timeout passes.
+// A mount caches metadata for about a second, so a read taken the instant the
+// writer's last close returned can legitimately still be behind; content that
+// is wrong rather than merely late never converges and still fails.
+func waitForContent(path string, want []byte, timeout time.Duration) (got []byte, ok bool) {
+	deadline := time.Now().Add(timeout)
+	for {
+		data, err := os.ReadFile(path)
+		if err == nil {
+			got = data
+			if bytes.Equal(got, want) {
+				return got, true
+			}
+		}
+		if time.Now().After(deadline) {
+			return got, false
+		}
+		time.Sleep(250 * time.Millisecond)
+	}
+}
+
+func expectedAppendContent(lines int) []byte {
+	var buf []byte
+	for i := 0; i < lines; i++ {
+		buf = append(buf, []byte(fmt.Sprintf("%07d\n", i))...)
+	}
+	return buf
+}
+
+// TestReadWithVolumeServerDown covers the original report: a file written while
+// everything was healthy must stay readable from a second mount after any
+// single volume server goes away.
+func TestReadWithVolumeServerDown(t *testing.T) {
+	c := startFailoverCluster(t, 3, 2)
+
+	const fileSize = 8 << 20
+	payloads := make(map[string][]byte)
+	for i := 0; i < 3; i++ {
+		name := fmt.Sprintf("readfile-%d", i)
+		data := make([]byte, fileSize)
+		_, err := rand.Read(data)
+		require.NoError(t, err)
+		require.NoError(t, os.WriteFile(filepath.Join(c.MountDir(0), name), data, 0644))
+		payloads[name] = data
+	}
+	// Let the filer commit the last chunks before anything is torn down.
+	time.Sleep(2 * time.Second)
+
+	t.Logf("topology before chaos:\n%s", c.MasterGet("/dir/status?pretty=y"))
+
+	alreadyRead := make(map[string]bool)
+	for victim := 0; victim < 3; victim++ {
+		// Placement decides which two of the three servers hold each volume, so
+		// pick a file this victim actually backs: reading one it never held
+		// would pass without exercising recovery at all. Prefer a file no
+		// earlier iteration has read, whose chunks the reader has not cached.
+		name := ""
+		for _, allowRead := range []bool{false, true} {
+			for i := 0; i < 3 && name == ""; i++ {
+				candidate := fmt.Sprintf("readfile-%d", i)
+				if alreadyRead[candidate] && !allowRead {
+					continue
+				}
+				onVictim, err := c.FileIsOn("/"+candidate, c.VolumeServerAddress(victim))
+				require.NoError(t, err, "resolve placement of %s", candidate)
+				if onVictim {
+					name = candidate
+				}
+			}
+		}
+		require.NotEmpty(t, name, "no test file has a replica on volume%d, nothing to fail over from\n%s",
+			victim, c.MasterGet("/dir/status?pretty=y"))
+		alreadyRead[name] = true
+
+		c.KillVolume(victim)
+
+		start := time.Now()
+		got, err := os.ReadFile(filepath.Join(c.MountDir(1), name))
+		elapsed := time.Since(start)
+		if err != nil {
+			t.Logf("topology with volume%d down:\n%s", victim, c.MasterGet("/dir/status?pretty=y"))
+		}
+		require.NoError(t, err, "read %s with volume %d down\n%s", name, victim, c.tailLog("mount1"))
+		require.Equal(t, sha256.Sum256(payloads[name]), sha256.Sum256(got),
+			"content mismatch for %s with volume %d down", name, victim)
+		t.Logf("read %s (held by volume%d) with volume%d down in %v", name, victim, victim, elapsed)
+
+		require.NoError(t, c.StartVolume(victim))
+		time.Sleep(3 * time.Second) // let the master see the heartbeat again
+	}
+}
+
+// TestReadAfterCachedLocationDies pins the reason the read path needs a cache
+// invalidator at all. Reading a file for the first time after a server dies
+// proves nothing: the lookup is fresh and simply returns the survivor. The
+// damage needs a reader whose cached location list is both stale and useless,
+// which is what this sequence builds:
+//
+//  1. kill one of the two servers holding a volume and wait for the master to
+//     drop it, so a lookup now resolves to the survivor alone;
+//  2. read a file on that volume, which caches exactly that one location;
+//  3. bring the first server back, which the reader never hears about;
+//  4. kill the survivor, and read another file on the same volume.
+//
+// The reader's only cached location is now dead while the data is live on the
+// restarted server. Without the invalidator the mount retries the corpse until
+// it gives up with EIO.
+func TestReadAfterCachedLocationDies(t *testing.T) {
+	c := startFailoverCluster(t, 3, 2)
+
+	// Small files so each is a single chunk on a single volume, which keeps the
+	// mapping from file to server unambiguous.
+	const fileSize = 256 << 10
+	payloads := make(map[string][]byte)
+	for i := 0; i < 6; i++ {
+		name := fmt.Sprintf("smallfile-%d", i)
+		data := make([]byte, fileSize)
+		_, err := rand.Read(data)
+		require.NoError(t, err)
+		require.NoError(t, os.WriteFile(filepath.Join(c.MountDir(0), name), data, 0644))
+		payloads[name] = data
+	}
+	time.Sleep(2 * time.Second)
+
+	// Two files on one volume: one to prime the reader's cache, one to probe
+	// with afterwards. The probe has to be a file the reader has never read, or
+	// its chunk would come from the local cache without a lookup at all.
+	filesByVolume := make(map[uint32][]string)
+	for name := range payloads {
+		vids, err := c.FileVolumeIds("/" + name)
+		require.NoError(t, err, "resolve volumes of %s", name)
+		require.Len(t, vids, 1, "%s should be a single chunk", name)
+		filesByVolume[vids[0]] = append(filesByVolume[vids[0]], name)
+	}
+	var vid uint32
+	var prime, probe string
+	for candidate, names := range filesByVolume {
+		if len(names) >= 2 {
+			vid, prime, probe = candidate, names[0], names[1]
+			break
+		}
+	}
+	require.NotEmpty(t, prime, "no volume holds two of the test files: %v", filesByVolume)
+
+	holders, err := c.VolumeHolders(vid)
+	require.NoError(t, err)
+	require.Len(t, holders, 2, "001 replication should put volume %d on two servers", vid)
+	first, second := c.volumeIndexOf(holders[0]), c.volumeIndexOf(holders[1])
+	require.NotEqual(t, -1, first)
+	require.NotEqual(t, -1, second)
+
+	c.KillVolume(first)
+	_, err = c.WaitForHolders(vid, 1, 90*time.Second)
+	require.NoError(t, err, "master did not drop the dead server")
+
+	got, err := os.ReadFile(filepath.Join(c.MountDir(1), prime))
+	require.NoError(t, err, "priming read\n%s", c.tailLog("mount1"))
+	require.Equal(t, sha256.Sum256(payloads[prime]), sha256.Sum256(got))
+
+	require.NoError(t, c.StartVolume(first))
+	_, err = c.WaitForHolders(vid, 2, 90*time.Second)
+	require.NoError(t, err, "restarted server did not re-register")
+
+	c.KillVolume(second)
+	start := time.Now()
+	got, err = os.ReadFile(filepath.Join(c.MountDir(1), probe))
+	require.NoError(t, err, "probe read after the cached location died\n%s", c.tailLog("mount1"))
+	require.Equal(t, sha256.Sum256(payloads[probe]), sha256.Sum256(got))
+	t.Logf("recovered %s from the restarted server in %v", probe, time.Since(start))
+}
+
+// TestAppendWithoutChaos is the control for the chaos runs below: the same
+// append-and-tail workload with nothing being stopped or started.
+func TestAppendWithoutChaos(t *testing.T) {
+	c := startFailoverCluster(t, 3, 2)
+	runChaosAppend(t, c, "no-chaos", func() {})
+}
+
+// TestAppendWhileVolumeServerStops is scenario "STOP volumes": append from
+// mount0 and tail from mount1 while one volume server is killed mid-stream.
+func TestAppendWhileVolumeServerStops(t *testing.T) {
+	for victim := 0; victim < 3; victim++ {
+		t.Run(fmt.Sprintf("volume%d", victim), func(t *testing.T) {
+			c := startFailoverCluster(t, 3, 2)
+			runChaosAppend(t, c, fmt.Sprintf("stop-%d", victim), func() {
+				c.KillVolume(victim)
+			})
+		})
+	}
+}
+
+// TestAppendWhileVolumeServerStarts is scenario "Start volumes": the cluster is
+// already one server down when the writer starts, and that server comes back
+// mid-stream. This is where the reporter saw 12-38 s stalls per append.
+func TestAppendWhileVolumeServerStarts(t *testing.T) {
+	for victim := 0; victim < 3; victim++ {
+		t.Run(fmt.Sprintf("volume%d", victim), func(t *testing.T) {
+			c := startFailoverCluster(t, 3, 2)
+			c.KillVolume(victim)
+			time.Sleep(3 * time.Second)
+			runChaosAppend(t, c, fmt.Sprintf("start-%d", victim), func() {
+				require.NoError(t, c.StartVolume(victim))
+			})
+		})
+	}
+}
+
+// TestAppendWhileVolumeServerRestarts is scenario "Re-start volumes".
+func TestAppendWhileVolumeServerRestarts(t *testing.T) {
+	for victim := 0; victim < 3; victim++ {
+		t.Run(fmt.Sprintf("volume%d", victim), func(t *testing.T) {
+			c := startFailoverCluster(t, 3, 2)
+			runChaosAppend(t, c, fmt.Sprintf("restart-%d", victim), func() {
+				c.KillVolume(victim)
+				time.Sleep(2 * time.Second)
+				require.NoError(t, c.StartVolume(victim))
+			})
+		})
+	}
+}
+
+// TestLargeWriteWhileVolumeServerStops is the part-2 variant: a single large
+// copy instead of many small appends, checksum-verified from the other mount.
+func TestLargeWriteWhileVolumeServerStops(t *testing.T) {
+	c := startFailoverCluster(t, 3, 2)
+
+	const size = 64 << 20
+	data := make([]byte, size)
+	_, err := rand.Read(data)
+	require.NoError(t, err)
+
+	path := filepath.Join(c.MountDir(0), "largefile")
+	f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
+	require.NoError(t, err)
+
+	killed := false
+	const block = 1 << 20
+	for off := 0; off < size; off += block {
+		if !killed && off >= size/3 {
+			c.KillVolume(2)
+			killed = true
+		}
+		_, werr := f.Write(data[off : off+block])
+		require.NoError(t, werr, "write at offset %d with volume2 down\n%s", off, c.tailLog("mount0"))
+	}
+	require.NoError(t, f.Close())
+
+	got, converged := waitForContent(filepath.Join(c.MountDir(1), "largefile"), data, convergeTimeout)
+	require.True(t, converged, "large file did not converge on mount1: want %d bytes %x, got %d bytes %x\n%s",
+		len(data), sha256.Sum256(data), len(got), sha256.Sum256(got), c.tailLog("mount1"))
+}
+
+// runChaosAppend appends from mount0 with a reader tailing on mount1, firing
+// chaos once the writer is a quarter of the way in.
+func runChaosAppend(t *testing.T, c *failoverCluster, name string, chaos func()) {
+	t.Helper()
+
+	writePath := filepath.Join(c.MountDir(0), name)
+	readPath := filepath.Join(c.MountDir(1), name)
+
+	r := startReader(readPath)
+
+	var once sync.Once
+	res := appendLoop(writePath, appendLines, func(i int) {
+		if i == appendLines/4 {
+			once.Do(chaos)
+		}
+	})
+	readErrs := r.Stop()
+
+	require.Empty(t, res.errs, "append errors\n%s", c.tailLog("mount0"))
+	require.Empty(t, readErrs, "reader errors\n%s", c.tailLog("mount1"))
+	t.Logf("%s: slowest append was line %d at %v", name, res.slowest, res.maxLatency)
+	// A healthy append stays well under a second; the reported regression parked
+	// single appends at 12-38 s while a volume server was coming back.
+	require.Less(t, res.maxLatency, maxAppendLatency,
+		"append %d stalled for %v\n%s", res.slowest, res.maxLatency, c.tailLog("mount0"))
+
+	wantBytes := expectedAppendContent(appendLines)
+	want := string(wantBytes)
+	gotBytes, converged := waitForContent(readPath, wantBytes, convergeTimeout)
+	got := string(gotBytes)
+	if converged {
+		return
+	}
+
+	// The reader never caught up. Show where it diverges and what the writer's
+	// own mount and the filer make of the same file, which says whether the
+	// data was lost on the way in or is only invisible from this side.
+	d := firstDiff(want, got)
+	fromWriter, _ := os.ReadFile(writePath)
+	viaFiler, filerErr := c.FilerGet("/" + name)
+	require.Failf(t, "final content mismatch",
+		"%s: first difference at offset %d (want %d bytes, got %d)\nwant %q\ngot  %q\nmount0 matches=%v filer matches=%v (err %v)\n%s",
+		name, d, len(want), len(got), window(want, d), window(got, d),
+		string(fromWriter) == want, string(viaFiler) == want, filerErr,
+		c.tailLog("mount0"))
+}
+
+// firstDiff returns the offset of the first differing byte, or -1 when equal.
+func firstDiff(want, got string) int {
+	for i := 0; i < len(want) && i < len(got); i++ {
+		if want[i] != got[i] {
+			return i
+		}
+	}
+	if len(want) != len(got) {
+		return min(len(want), len(got))
+	}
+	return -1
+}
+
+func window(s string, at int) string {
+	return s[max(0, at-24):min(len(s), at+24)]
+}
@@ -26,7 +26,7 @@ type ChunkGroup struct {
 // - Read-ahead prefetch parallelism
 // - Number of concurrent section reads for large files
 // If concurrentReaders <= 0, defaults to 16.
-func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_cache.ChunkCache, chunks []*filer_pb.FileChunk, concurrentReaders int) (*ChunkGroup, error) {
+func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_cache.ChunkCache, chunks []*filer_pb.FileChunk, concurrentReaders int, cacheInvalidator CacheInvalidator) (*ChunkGroup, error) {
 	if concurrentReaders <= 0 {
 		concurrentReaders = 16
 	}
@@ -41,7 +41,7 @@ func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_
 	group := &ChunkGroup{
 		lookupFn:          lookupFn,
 		sections:          make(map[SectionIndex]*FileChunkSection),
-		readerCache:       NewReaderCache(readerCacheLimit, chunkCache, lookupFn, nil),
+		readerCache:       NewReaderCache(readerCacheLimit, chunkCache, lookupFn, cacheInvalidator),
 		concurrentReaders: concurrentReaders,
 	}

@@ -130,7 +130,7 @@ func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) {
 		fileSize := filer.FileSize(entry)
 		entry.Attributes.FileSize = fileSize
 		var resolveManifestErr error
-		fh.entryChunkGroup, resolveManifestErr = filer.NewChunkGroup(fh.wfs.LookupFn(), fh.wfs.chunkCache, entry.Chunks, fh.wfs.option.ConcurrentReaders)
+		fh.entryChunkGroup, resolveManifestErr = filer.NewChunkGroup(fh.wfs.LookupFn(), fh.wfs.chunkCache, entry.Chunks, fh.wfs.option.ConcurrentReaders, fh.wfs.CacheInvalidator())
 		if resolveManifestErr != nil {
 			glog.Warningf("failed to resolve manifest chunks in %+v", entry)
 		}
@@ -962,6 +962,17 @@ func (wfs *WFS) LookupFn() wdclient.LookupFileIdFunctionType {
 	return wfs.filerClient.GetLookupFileIdFunction()
 }

+// CacheInvalidator lets a chunk read that failed against every cached location
+// drop that entry and look the volume up again, so a mount does not keep
+// hammering a volume server that has since moved or died. Nil under filerProxy,
+// where there is no filerClient and LookupFn never consults a location cache.
+func (wfs *WFS) CacheInvalidator() filer.CacheInvalidator {
+	if wfs.filerClient == nil {
+		return nil
+	}
+	return wfs.filerClient
+}
+
 func (wfs *WFS) getCurrentFiler() pb.ServerAddress {
 	i := atomic.LoadInt32(&wfs.option.filerIndex)
 	return wfs.option.FilerAddresses[i]
@@ -34,7 +34,7 @@ func TestAttrChunkRace(t *testing.T) {
 		Name:       "sample.txt",
 		Attributes: &filer_pb.FuseAttributes{FileMode: 0644},
 	}
-	chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1)
+	chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil)
 	if err != nil {
 		t.Fatalf("NewChunkGroup: %v", err)
 	}
@@ -114,7 +114,7 @@ func TestReadFromChunksRace(t *testing.T) {
 		Name:       "sample.txt",
 		Attributes: &filer_pb.FuseAttributes{FileMode: 0644},
 	}
-	chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1)
+	chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil)
 	if err != nil {
 		t.Fatalf("NewChunkGroup: %v", err)
 	}