filer: rebuild peer metadata subscriptions after a master reconnect (#10648)

* filer: keep the existing peer subscription on a repeated add

A cluster node add for a peer that is already followed restarted the
subscription, dropping the metadata events between the two runs.

* master: tell a connecting client the current cluster membership

Cluster node updates are only broadcast to the clients connected at that
moment. A filer that lost its master stream while a peer came back never
learned about the peer, and stopped replicating its metadata for good.

* test: a filer joining the master learns about the filers already there

* test: a filer resubscribes to a peer that registered while it was disconnected

Runs the reported sequence against real processes: filer2 leaves, filer1
is paused and its master stream is broken, filer2 registers again, and
filer1 has to replicate from it after reconnecting.

.github/workflows/metadata-subscribe-tests.yml

@@ -8,6 +8,7 @@ on:
       - 'weed/pb/filer_pb/**'
       - 'weed/util/log_buffer/**'
       - 'weed/server/filer_grpc_server_sub_meta.go'
+      - 'weed/server/master_grpc_server.go'
       - 'weed/command/filer_backup.go'
       - 'test/metadata_subscribe/**'
       - '.github/workflows/metadata-subscribe-tests.yml'
@@ -18,6 +19,7 @@ on:
       - 'weed/pb/filer_pb/**'
       - 'weed/util/log_buffer/**'
       - 'weed/server/filer_grpc_server_sub_meta.go'
+      - 'weed/server/master_grpc_server.go'
       - 'weed/command/filer_backup.go'
       - 'test/metadata_subscribe/**'
       - '.github/workflows/metadata-subscribe-tests.yml'

test/metadata_subscribe/peer_resubscribe_test.go

@@ -0,0 +1,248 @@
+//go:build !windows
+
+package metadata_subscribe
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/test/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+)
+
+// A filer drops the metadata subscription to a peer that leaves, and only an
+// add from the master brings it back. Those updates are broadcast to the
+// clients connected at that moment, so a filer whose master stream broke while
+// the peer came back used to stay unsubscribed for good, and metadata written
+// on the peer never reached it again.
+func TestFilerResubscribesToPeerAfterMasterReconnect(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	weedBinary := findWeedBinary()
+	require.NotEmpty(t, weedBinary, "weed binary not found")
+
+	testDir, err := os.MkdirTemp("", "seaweedfs_peer_resubscribe_")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if t.Failed() {
+			t.Logf("logs kept at %s", testDir)
+			return
+		}
+		os.RemoveAll(testDir)
+	})
+
+	ports, err := testutil.AllocateMiniPorts(3)
+	require.NoError(t, err)
+	masterPort, filer1Port, filer2Port := ports[0], ports[1], ports[2]
+
+	master := pb.ServerAddress(fmt.Sprintf("127.0.0.1:%d", masterPort))
+	filer1Address := fmt.Sprintf("127.0.0.1:%d", filer1Port)
+	filer2Address := fmt.Sprintf("127.0.0.1:%d", filer2Port)
+	filer1Log := filepath.Join(testDir, "filer1.log")
+
+	masterArgs := []string{"master",
+		"-ip=127.0.0.1",
+		"-port=" + strconv.Itoa(masterPort),
+		"-mdir=" + mkdir(t, testDir, "master"),
+		"-peers=none"}
+	masterProcess := startProcess(t, weedBinary, filepath.Join(testDir, "master.log"), masterArgs...)
+	require.NoError(t, waitForLeader(masterPort, 60*time.Second))
+
+	// one at a time: a filer bootstraps from the peers the master already knows,
+	// and gives up if one of them is registered but not yet listening
+	filer1 := startFiler(t, weedBinary, testDir, "filer1", filer1Port, masterPort)
+	require.NoError(t, waitForHTTPServer(fmt.Sprintf("http://127.0.0.1:%d/", filer1Port), 30*time.Second))
+	filer2 := startFiler(t, weedBinary, testDir, "filer2", filer2Port, masterPort)
+	require.NoError(t, waitForHTTPServer(fmt.Sprintf("http://127.0.0.1:%d/", filer2Port), 30*time.Second))
+
+	// the peer subscription works to begin with
+	createPeerEntry(t, filer2Address, "baseline")
+	require.NoError(t, waitForPeerEntry(filer1Address, "baseline", 60*time.Second),
+		"filer1 never replicated the baseline entry from filer2")
+
+	// filer2 leaves, and filer1 drops the subscription
+	stopProcess(filer2)
+	require.NoError(t, waitForLog(filer1Log, "stop subscribing peer "+filer2Address, 60*time.Second),
+		"filer1 never dropped the subscription to filer2")
+
+	// filer1 stops reading its master stream, and restarting the master breaks
+	// it, so filer1 hears nothing until it reconnects
+	require.NoError(t, filer1.Process.Signal(syscall.SIGSTOP))
+	stopProcess(masterProcess)
+	startProcess(t, weedBinary, filepath.Join(testDir, "master.log"), masterArgs...)
+	require.NoError(t, waitForLeader(masterPort, 60*time.Second))
+
+	// filer2 comes back and registers while filer1 cannot hear about it
+	filer2 = startFiler(t, weedBinary, testDir, "filer2", filer2Port, masterPort)
+	require.NoError(t, waitForHTTPServer(fmt.Sprintf("http://127.0.0.1:%d/", filer2Port), 30*time.Second))
+	require.NoError(t, waitForClusterNode(master, filer2Address, 60*time.Second),
+		"the master never registered filer2 again")
+
+	require.NoError(t, filer1.Process.Signal(syscall.SIGCONT))
+	require.NoError(t, waitForClusterNode(master, filer1Address, 60*time.Second),
+		"filer1 never reconnected to the master")
+
+	createPeerEntry(t, filer2Address, "after-reconnect")
+	require.NoError(t, waitForPeerEntry(filer1Address, "after-reconnect", 90*time.Second),
+		"filer1 reconnected to the master but never resubscribed to filer2")
+}
+
+const peerEntryDir = "/peer-resubscribe"
+
+func createPeerEntry(t *testing.T, filerAddress, name string) {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	err := pb.WithFilerClient(false, 0, pb.ServerAddress(filerAddress), grpc.WithTransportCredentials(insecure.NewCredentials()), func(client filer_pb.SeaweedFilerClient) error {
+		_, err := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{
+			Directory: peerEntryDir,
+			Entry: &filer_pb.Entry{
+				Name: name,
+				Attributes: &filer_pb.FuseAttributes{
+					Mtime:    time.Now().Unix(),
+					FileMode: 0644,
+				},
+			},
+		})
+		return err
+	})
+	require.NoError(t, err, "create %s/%s on %s", peerEntryDir, name, filerAddress)
+}
+
+func waitForPeerEntry(filerAddress, name string, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	var lastErr error
+	for time.Now().Before(deadline) {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		lastErr = pb.WithFilerClient(false, 0, pb.ServerAddress(filerAddress), grpc.WithTransportCredentials(insecure.NewCredentials()), func(client filer_pb.SeaweedFilerClient) error {
+			_, err := client.LookupDirectoryEntry(ctx, &filer_pb.LookupDirectoryEntryRequest{
+				Directory: peerEntryDir,
+				Name:      name,
+			})
+			return err
+		})
+		cancel()
+		if lastErr == nil {
+			return nil
+		}
+		time.Sleep(time.Second)
+	}
+	return fmt.Errorf("%s/%s not on %s within %v: %w", peerEntryDir, name, filerAddress, timeout, lastErr)
+}
+
+func waitForClusterNode(master pb.ServerAddress, address string, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		found := false
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		err := pb.WithMasterClient(ctx, false, master, grpc.WithTransportCredentials(insecure.NewCredentials()), false, func(client master_pb.SeaweedClient) error {
+			resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ClientType: cluster.FilerType})
+			if err != nil {
+				return err
+			}
+			for _, node := range resp.ClusterNodes {
+				// the master reports the grpc port too, as "host:port.grpcPort"
+				if pb.ServerAddress(node.Address).Equals(pb.ServerAddress(address)) {
+					found = true
+				}
+			}
+			return nil
+		})
+		cancel()
+		if err == nil && found {
+			return nil
+		}
+		time.Sleep(time.Second)
+	}
+	return fmt.Errorf("%s not registered within %v", address, timeout)
+}
+
+func waitForLeader(masterPort int, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	url := fmt.Sprintf("http://127.0.0.1:%d/cluster/status", masterPort)
+	client := &http.Client{Timeout: 2 * time.Second}
+	for time.Now().Before(deadline) {
+		if resp, err := client.Get(url); err == nil {
+			body, _ := io.ReadAll(resp.Body)
+			resp.Body.Close()
+			if strings.Contains(string(body), `"IsLeader":true`) {
+				return nil
+			}
+		}
+		time.Sleep(time.Second)
+	}
+	return fmt.Errorf("master on %d has no leader within %v", masterPort, timeout)
+}
+
+func waitForLog(logFile, message string, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		content, err := os.ReadFile(logFile)
+		if err == nil && strings.Contains(string(content), message) {
+			return nil
+		}
+		time.Sleep(time.Second)
+	}
+	return fmt.Errorf("%q not in %s within %v", message, logFile, timeout)
+}
+
+func mkdir(t *testing.T, dir, name string) string {
+	t.Helper()
+	path := filepath.Join(dir, name)
+	require.NoError(t, os.MkdirAll(path, 0755))
+	return path
+}
+
+func startFiler(t *testing.T, weedBinary, testDir, name string, port, masterPort int) *exec.Cmd {
+	t.Helper()
+	return startProcess(t, weedBinary, filepath.Join(testDir, name+".log"), "filer",
+		"-ip=127.0.0.1",
+		"-port="+strconv.Itoa(port),
+		"-master=127.0.0.1:"+strconv.Itoa(masterPort),
+		"-defaultStoreDir="+mkdir(t, testDir, name))
+}
+
+func startProcess(t *testing.T, weedBinary, logFile string, args ...string) *exec.Cmd {
+	t.Helper()
+	log, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
+	require.NoError(t, err)
+
+	cmd := exec.Command(weedBinary, args...)
+	cmd.Stdout = log
+	cmd.Stderr = log
+	require.NoError(t, cmd.Start())
+	t.Cleanup(func() {
+		cmd.Process.Signal(syscall.SIGCONT)
+		stopProcess(cmd)
+	})
+	return cmd
+}
+
+// stopProcess kills the process outright: a filer takes seconds to shut down
+// gracefully, long enough to register with the master again on the way out.
+func stopProcess(cmd *exec.Cmd) {
+	if cmd == nil || cmd.Process == nil {
+		return
+	}
+	cmd.Process.Kill()
+	cmd.Wait()
+}

test/multi_master/cluster_membership_test.go

@@ -0,0 +1,103 @@
+package multi_master
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+)
+
+// A filer only learns about its peers from the cluster node updates on its
+// KeepConnected stream, and those are broadcast to whoever is connected at that
+// moment. A filer that reconnects has to be told the membership again, or it
+// never subscribes to the peers that registered while it was away.
+func TestKeepConnectedSendsExistingFilers(t *testing.T) {
+	mc := StartMasterCluster(t)
+
+	leaderIdx, leaderAddr := mc.FindLeader()
+	if leaderIdx < 0 {
+		t.Fatal("no leader")
+	}
+	master := pb.ServerAddress(leaderAddr)
+	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+
+	const existingFiler = "127.0.0.1:18888"
+	const joiningFiler = "127.0.0.1:18889"
+
+	ctx, cancel := context.WithTimeout(context.Background(), waitTimeout)
+	defer cancel()
+
+	err := pb.WithMasterClient(ctx, true, master, dialOption, false, func(client master_pb.SeaweedClient) error {
+		stream, err := client.KeepConnected(ctx)
+		if err != nil {
+			return err
+		}
+		if err := stream.Send(&master_pb.KeepConnectedRequest{
+			ClientType:    cluster.FilerType,
+			ClientAddress: existingFiler,
+		}); err != nil {
+			return err
+		}
+		if err := waitForClusterNode(ctx, client, existingFiler); err != nil {
+			return err
+		}
+
+		return pb.WithMasterClient(ctx, true, master, dialOption, false, func(joining master_pb.SeaweedClient) error {
+			joiningCtx, cancelJoining := context.WithTimeout(ctx, waitTimeout)
+			defer cancelJoining()
+
+			joiningStream, err := joining.KeepConnected(joiningCtx)
+			if err != nil {
+				return err
+			}
+			if err := joiningStream.Send(&master_pb.KeepConnectedRequest{
+				ClientType:    cluster.FilerType,
+				ClientAddress: joiningFiler,
+			}); err != nil {
+				return err
+			}
+			for i := 0; ; i++ {
+				resp, err := joiningStream.Recv()
+				if err != nil {
+					return err
+				}
+				// a client only reads the volume locations out of the first
+				// message, an update sent ahead of them would be dropped
+				if i == 0 && resp.VolumeLocation == nil {
+					return fmt.Errorf("first message is not a volume location: %+v", resp)
+				}
+				if update := resp.ClusterNodeUpdate; update != nil && update.IsAdd && update.Address == existingFiler {
+					return nil
+				}
+			}
+		})
+	})
+	if err != nil {
+		mc.DumpLogs()
+		t.Fatalf("a joining filer was not told about %s: %v", existingFiler, err)
+	}
+}
+
+func waitForClusterNode(ctx context.Context, client master_pb.SeaweedClient, address string) error {
+	deadline := time.Now().Add(waitTimeout)
+	for time.Now().Before(deadline) {
+		resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ClientType: cluster.FilerType})
+		if err != nil {
+			return err
+		}
+		for _, node := range resp.ClusterNodes {
+			if node.Address == address {
+				return nil
+			}
+		}
+		time.Sleep(waitTick)
+	}
+	return context.DeadlineExceeded
+}

weed/cluster/cluster.go

@@ -138,6 +138,16 @@ func (cluster *Cluster) ListClusterNode(filerGroup FilerGroupName, nodeType stri
 	return
 }
 
+// ListClusterNodeUpdates reports the current members as add updates, so a
+// client that just connected can rebuild the membership it missed while it was
+// away.
+func (cluster *Cluster) ListClusterNodeUpdates(filerGroup FilerGroupName, nodeType string) (updates []*master_pb.KeepConnectedResponse) {
+	for _, node := range cluster.ListClusterNode(filerGroup, nodeType) {
+		updates = append(updates, buildClusterNodeUpdateMessage(true, filerGroup, nodeType, node.Address)...)
+	}
+	return
+}
+
 // IsKnownNode reports whether address is currently registered under nodeType
 // in any filer group. The lookup is intentionally group-agnostic because callers
 // (e.g. Ping admission) only know the target address, not the group it joined.

weed/cluster/cluster_test.go

@@ -40,6 +40,27 @@ func TestConcurrentAddRemoveNodes(t *testing.T) {
 	wg.Wait()
 }
 
+func TestListClusterNodeUpdates(t *testing.T) {
+	c := NewCluster()
+	filer := pb.ServerAddress("10.0.0.20:8888")
+	c.AddClusterNode("group", FilerType, "dc1", "rack1", filer, "test")
+	c.AddClusterNode("group", BrokerType, "dc1", "rack1", pb.ServerAddress("10.0.0.20:17777"), "test")
+
+	updates := c.ListClusterNodeUpdates("group", FilerType)
+	if len(updates) != 1 {
+		t.Fatalf("expecting one filer update, got %d", len(updates))
+	}
+	update := updates[0].ClusterNodeUpdate
+	if update.Address != string(filer) || !update.IsAdd || update.FilerGroup != "group" {
+		t.Fatalf("unexpected update %+v", update)
+	}
+
+	c.RemoveClusterNode("group", FilerType, filer)
+	if updates := c.ListClusterNodeUpdates("group", FilerType); len(updates) != 0 {
+		t.Fatalf("expecting no update for a removed filer, got %d", len(updates))
+	}
+}
+
 func TestIsKnownNode(t *testing.T) {
 	c := NewCluster()
 	filer := pb.ServerAddress("10.0.0.20:8888")

weed/filer/meta_aggregator.go

@@ -53,9 +53,10 @@ func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, star
 
 	address := pb.ServerAddress(update.Address)
 	if update.IsAdd {
-		// cancel previous subscription if any
-		if prevChan, found := ma.peerChans[address]; found {
-			close(prevChan)
+		// the peer is already followed, restarting would only lose the events
+		// in between
+		if _, found := ma.peerChans[address]; found {
+			return
 		}
 		stopChan := make(chan struct{})
 		ma.peerChans[address] = stopChan

weed/filer/meta_aggregator_test.go

@@ -0,0 +1,36 @@
+package filer
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+)
+
+func TestOnPeerUpdateRepeatedAdd(t *testing.T) {
+	peer := pb.ServerAddress("127.0.0.1:1")
+	ma := NewMetaAggregator(nil, pb.ServerAddress("127.0.0.1:2"), nil)
+
+	add := &master_pb.ClusterNodeUpdate{Address: string(peer), IsAdd: true}
+	ma.OnPeerUpdate(add, time.Now())
+	first, found := ma.peerChans[peer]
+	if !found {
+		t.Fatal("expecting a subscription after the first add")
+	}
+
+	ma.OnPeerUpdate(add, time.Now())
+	if ma.peerChans[peer] != first {
+		t.Fatal("expecting the same subscription after a repeated add")
+	}
+	select {
+	case <-first:
+		t.Fatal("expecting the subscription to stay alive after a repeated add")
+	default:
+	}
+
+	ma.OnPeerUpdate(&master_pb.ClusterNodeUpdate{Address: string(peer)}, time.Now())
+	if _, found := ma.peerChans[peer]; found {
+		t.Fatal("expecting the subscription to be removed")
+	}
+}

weed/server/master_grpc_server.go

@@ -437,6 +437,14 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
 		}
 	}
 
+	// Cluster node changes are only broadcast to the clients connected at that
+	// moment, so a client that reconnects has to be told who is around now.
+	for _, update := range ms.Cluster.ListClusterNodeUpdates(cluster.FilerGroupName(req.FilerGroup), cluster.FilerType) {
+		if sendErr := stream.Send(update); sendErr != nil {
+			return sendErr
+		}
+	}
+
 	if initialLockRingUpdate := ms.initialLockRingUpdate(req.ClientType, req.FilerGroup); initialLockRingUpdate != nil {
 		if sendErr := stream.Send(initialLockRingUpdate); sendErr != nil {
 			return sendErr