Read metadata log chunks the way the mount reads every other chunk (#11018)

* Replay metadata log chunks the way the mount reads every other chunk

The subscription's log-chunk replay built its own lookup, which always
resolves volume server addresses. A mount started with
-volumeServerAccess=filerProxy cannot reach those, so every fresh
subscription failed on the previous minute's persisted segment and
resubscribed a second later, forever. Take the lookup from the caller
instead; the mount hands over the one it uses for file reads, which also
keeps publicUrl and the bounded location cache in play.

Claude-Session: https://claude.ai/code/session_01NGqrxYj7cHUpSrL249n3Z6

* Keep a log chunk read failure off the filer connection

A metadata subscriber reads persisted log chunks over HTTP from volume
servers and hands whatever went wrong back as the subscription's error.
"connection refused" from a volume server then matched the transport
patterns that decide a gRPC channel is dead, so every failed replay
closed the shared filer ClientConn and cancelled the assign and upload
RPCs riding on it with "the client connection is closing". Mark those
read failures so they are judged for what they are.

Claude-Session: https://claude.ai/code/session_01NGqrxYj7cHUpSrL249n3Z6

weed/filer/meta_aggregator.go

@@ -501,7 +501,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 						return processOne(event)
 					})
 				if readErr != nil {
-					return fmt.Errorf("read log file refs from %s: %w", peer, readErr)
+					return fmt.Errorf("%w from %s: %w", pb.ErrLogFileRead, peer, readErr)
 				}
 				if lastTs > 0 {
 					lastTsNs = lastTs

weed/mount/meta_cache/meta_cache_subscribe.go

@@ -10,6 +10,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 )
 
 type MetadataFollower struct {
@@ -44,7 +45,7 @@ func mergeProcessors(mainProcessor func(resp *filer_pb.SubscribeMetadataResponse
 	}
 }
 
-func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, dir string, lastTsNs int64, skipSelfEvents bool, onRetry func(lastTsNs int64, err error), followers ...*MetadataFollower) error {
+func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, lookupFn wdclient.LookupFileIdFunctionType, dir string, lastTsNs int64, skipSelfEvents bool, onRetry func(lastTsNs int64, err error), followers ...*MetadataFollower) error {
 
 	var prefixes []string
 	for _, follower := range followers {
@@ -68,9 +69,9 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
 		prefix = prefix + "/"
 	}
 
-	// Read persisted log chunks directly from volume servers, keeping the replay
-	// cost off the filer's heap (see LogFileReaderFn below).
-	lookupFn := filer.LookupFn(client)
+	// Replaying the persisted log chunks here keeps the cost off the filer's heap.
+	// The caller's lookup says where to read them from, so a mount that cannot
+	// reach volume servers directly still replays through its filer.
 	metadataFollowOption := &pb.MetadataFollowOption{
 		ClientName:             "mount",
 		ClientId:               selfSignature,

weed/mount/weedfs.go

@@ -493,7 +493,7 @@ func (wfs *WFS) StartBackgroundTasks() error {
 	}
 
 	startTime := time.Now()
-	go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano(), wfs.option.WritebackCache, func(lastTsNs int64, err error) {
+	go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.LookupFn(), wfs.option.FilerMountRootPath, startTime.UnixNano(), wfs.option.WritebackCache, func(lastTsNs int64, err error) {
 		glog.Warningf("meta events follow retry from %v: %v", time.Unix(0, lastTsNs), err)
 		// A subscription gap may have dropped events, so distrust every cached
 		// listing. Reset the flags first (safe — it never deletes entries), then

weed/mount/weedfs_meta_subscribe_test.go

@@ -0,0 +1,184 @@
+package mount
+
+import (
+	"bytes"
+	"context"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// proxyFilerServer replays the previous minute's metadata as a persisted log
+// chunk, the way a filer answers every fresh subscription, and points volume
+// lookups at an address nothing listens on: the mount host's view of a volume
+// server it is meant to reach only through -volumeServerAccess=filerProxy.
+type proxyFilerServer struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	chunk         *filer_pb.FileChunk
+	subscriptions atomic.Int32
+	volumeLookups atomic.Int32
+	stop          chan struct{}
+}
+
+func (s *proxyFilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer) error {
+	if s.subscriptions.Add(1) > 1 {
+		<-s.stop
+		return nil
+	}
+	return stream.Send(&filer_pb.SubscribeMetadataResponse{
+		LogFileRefs: []*filer_pb.LogFileChunkRef{{
+			FilerId:  "5319c0e8",
+			FileTsNs: time.Now().UnixNano(),
+			Chunks:   []*filer_pb.FileChunk{s.chunk},
+		}},
+	})
+}
+
+func (s *proxyFilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
+	s.volumeLookups.Add(1)
+	locations := make(map[string]*filer_pb.Locations)
+	for _, vid := range req.VolumeIds {
+		locations[vid] = &filer_pb.Locations{Locations: []*filer_pb.Location{{Url: "127.0.0.1:1", PublicUrl: "127.0.0.1:1"}}}
+	}
+	return &filer_pb.LookupVolumeResponse{LocationsMap: locations}, nil
+}
+
+// buildLogFileData writes the on-disk log format: [4-byte size | LogEntry].
+func buildLogFileData(events ...*filer_pb.SubscribeMetadataResponse) []byte {
+	var buf bytes.Buffer
+	for _, event := range events {
+		eventData, _ := proto.Marshal(event)
+		entryData, _ := proto.Marshal(&filer_pb.LogEntry{TsNs: event.TsNs, Data: eventData, Key: []byte(event.Directory)})
+		sizeBuf := make([]byte, 4)
+		util.Uint32toBytes(sizeBuf, uint32(len(entryData)))
+		buf.Write(sizeBuf)
+		buf.Write(entryData)
+	}
+	return buf.Bytes()
+}
+
+// TestSubscribeMetaEventsReplaysThroughFilerProxy pins the replay to the lookup
+// the mount was given: resolving volume servers here fails every subscription
+// of a filerProxy mount, which then resubscribes a second later, forever.
+func TestSubscribeMetaEventsReplaysThroughFilerProxy(t *testing.T) {
+	logData := buildLogFileData(&filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      time.Now().UnixNano(),
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: "file", Attributes: &filer_pb.FuseAttributes{FileSize: 7}},
+		},
+	})
+
+	var proxiedChunkIds atomic.Value
+	proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fileId := r.URL.Query().Get("proxyChunkId")
+		if fileId == "" {
+			http.Error(w, "not a filer proxy read", http.StatusBadRequest)
+			return
+		}
+		proxiedChunkIds.Store(fileId)
+		w.Write(logData)
+	}))
+	t.Cleanup(proxy.Close)
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	t.Cleanup(func() { _ = listener.Close() })
+
+	testServer := &proxyFilerServer{
+		chunk: &filer_pb.FileChunk{FileId: "4471,ce598bfff8416e", Size: uint64(len(logData))},
+		stop:  make(chan struct{}),
+	}
+	server := pb.NewGrpcServer()
+	filer_pb.RegisterSeaweedFilerServer(server, testServer)
+	go server.Serve(listener)
+	t.Cleanup(func() {
+		close(testServer.stop)
+		server.Stop()
+	})
+
+	proxyUrl, err := url.Parse(proxy.URL)
+	if err != nil {
+		t.Fatalf("parse proxy url: %v", err)
+	}
+
+	uidGidMapper, err := meta_cache.NewUidGidMapper("", "")
+	if err != nil {
+		t.Fatalf("create uid/gid mapper: %v", err)
+	}
+	root := util.FullPath("/")
+	wfs := &WFS{
+		signature:         1,
+		inodeToPath:       NewInodeToPath(root, 0),
+		fhMap:             NewFileHandleToInode(),
+		fhLockTable:       util.NewLockTable[FileHandleId](),
+		hardLinkLockTable: util.NewLockTable[string](),
+		option: &Option{
+			ChunkSizeLimit:     1024,
+			ConcurrentReaders:  1,
+			VolumeServerAccess: "filerProxy",
+			FilerMountRootPath: "/",
+			FilerAddresses: []pb.ServerAddress{
+				pb.NewServerAddressWithGrpcPort(proxyUrl.Host, listener.Addr().(*net.TCPAddr).Port),
+			},
+			GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		},
+	}
+	wfs.metaCache = meta_cache.NewMetaCache(
+		filepath.Join(t.TempDir(), "meta"),
+		uidGidMapper,
+		root,
+		false,
+		func(path util.FullPath) { wfs.inodeToPath.MarkChildrenCached(path) },
+		func(path util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(path) },
+		func(meta_cache.EntryInvalidation) {},
+		nil,
+	)
+	t.Cleanup(wfs.metaCache.Shutdown)
+	wfs.inodeToPath.MarkChildrenCached(root)
+	wfs.inodeToPath.Lookup(util.FullPath("/dir"), time.Now().Unix(), true, false, 0, false)
+	wfs.inodeToPath.MarkChildrenCached(util.FullPath("/dir"))
+
+	followed := make(chan struct{})
+	go func() {
+		defer close(followed)
+		meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.LookupFn(), wfs.option.FilerMountRootPath, 0, false, nil)
+	}()
+
+	select {
+	case <-followed:
+	case <-time.After(20 * time.Second):
+		t.Fatal("the log chunk replay never succeeded")
+	}
+
+	if got := testServer.volumeLookups.Load(); got != 0 {
+		t.Errorf("filerProxy mount resolved %d volume locations, want 0", got)
+	}
+	if got, _ := proxiedChunkIds.Load().(string); got != testServer.chunk.FileId {
+		t.Errorf("filer proxy served chunk %q, want %q", got, testServer.chunk.FileId)
+	}
+
+	entry, _, err := wfs.metaCache.FindEntry(context.Background(), util.FullPath("/dir/file"))
+	if err != nil {
+		t.Fatalf("replayed entry: %v", err)
+	}
+	if entry.Attr.FileSize != 7 {
+		t.Errorf("replayed file size %d, want 7", entry.Attr.FileSize)
+	}
+}

weed/pb/filer_pb_direct_read.go

@@ -25,6 +25,11 @@ const logEntryChannelSize = 512
 // prefix, mirroring the filer package's unexported constant.
 const maxLogEntrySize = 1 << 30
 
+// ErrLogFileRead marks a failure to read persisted log chunks from volume
+// servers. It surfaces as the metadata stream's error, but says nothing about
+// the filer connection that carried the stream.
+var ErrLogFileRead = errors.New("read log file refs")
+
 // errReaderStopped signals that the entry consumer asked to stop (the merge
 // loop aborted or the caller hit a processing error). It is not a read failure.
 var errReaderStopped = errors.New("log entry reader stopped")

weed/pb/filer_pb_tail.go

@@ -154,7 +154,7 @@ func makeSubscribeMetadataFunc(option *MetadataFollowOption, processEventFn Proc
 				},
 				processEventFn)
 			if readErr != nil {
-				return fmt.Errorf("read log file refs: %w", readErr)
+				return fmt.Errorf("%w: %w", ErrLogFileRead, readErr)
 			}
 			if lastTs > 0 {
 				option.StartTsNs = lastTs

weed/pb/grpc_client_cascade_test.go

@@ -2,9 +2,13 @@ package pb
 
 import (
 	"context"
+	"errors"
+	"io"
 	"net"
+	"net/http"
 	"sync"
 	"testing"
+	"time"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -104,7 +108,16 @@ func TestWithGrpcClient_AbandonedRequestKeepsSharedConnection(t *testing.T) {
 }
 
 func (s *cascadeFilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer) error {
-	return nil
+	if !req.ClientSupportsMetadataChunks {
+		return nil
+	}
+	return stream.Send(&filer_pb.SubscribeMetadataResponse{
+		LogFileRefs: []*filer_pb.LogFileChunkRef{{
+			FilerId:  "5319c0e8",
+			FileTsNs: time.Now().UnixNano(),
+			Chunks:   []*filer_pb.FileChunk{{FileId: "4471,ce598bfff8416e", Size: 64}},
+		}},
+	})
 }
 
 // TestWithGrpcClient_EndedStreamKeepsSharedConnection covers the other half of
@@ -145,3 +158,52 @@ func TestWithGrpcClient_EndedStreamKeepsSharedConnection(t *testing.T) {
 		t.Fatalf("concurrent caller must survive an unrelated stream ending: %v", concurrentErr)
 	}
 }
+
+// TestWithGrpcClient_LogChunkReadFailureKeepsSharedConnection covers a
+// subscriber replaying persisted log chunks over HTTP: an unreachable volume
+// server reports "connection refused" as the subscription's error, which used
+// to drop the shared filer ClientConn and cancel the assign and upload RPCs
+// riding on it.
+func TestWithGrpcClient_LogChunkReadFailureKeepsSharedConnection(t *testing.T) {
+	fake, address, dialOption := startCascadeFiler(t)
+
+	var concurrentErr error
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		concurrentErr = WithGrpcClient(context.Background(), false, 0, func(connection *grpc.ClientConn) error {
+			_, err := filer_pb.NewSeaweedFilerClient(connection).LookupDirectoryEntry(context.Background(),
+				&filer_pb.LookupDirectoryEntryRequest{Directory: "/buckets/b", Name: "slow"})
+			return err
+		}, address, false, dialOption)
+	}()
+	<-fake.inFlight
+
+	option := &MetadataFollowOption{
+		ClientName:     "mount",
+		PathPrefix:     "/",
+		EventErrorType: DontLogError,
+		LogFileReaderFn: func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) {
+			// An unreachable volume server, as a filerProxy mount sees it.
+			response, err := http.Get("http://127.0.0.1:1/" + chunks[0].FileId)
+			if err != nil {
+				return nil, err
+			}
+			return response.Body, nil
+		},
+	}
+	subscribe := makeSubscribeMetadataFunc(option, func(resp *filer_pb.SubscribeMetadataResponse) error { return nil })
+	err := WithGrpcClient(context.Background(), true, 0, func(connection *grpc.ClientConn) error {
+		return subscribe(filer_pb.NewSeaweedFilerClient(connection))
+	}, address, false, dialOption)
+	if err == nil || !errors.Is(err, ErrLogFileRead) {
+		t.Fatalf("the replay should have failed reading the log chunk: %v", err)
+	}
+
+	close(fake.release)
+	wg.Wait()
+	if concurrentErr != nil {
+		t.Fatalf("concurrent caller must survive a log chunk read failure: %v", concurrentErr)
+	}
+}

weed/pb/grpc_client_server.go

@@ -401,6 +401,13 @@ func shouldInvalidateConnection(ctx context.Context, err error) bool {
 		return false
 	}
 
+	// A metadata subscriber reads log chunks over HTTP from volume servers and
+	// returns what went wrong there through the stream. Those failures read like
+	// transport failures below, and would drop a filer channel that is fine.
+	if errors.Is(err, ErrLogFileRead) {
+		return false
+	}
+
 	// gRPC raises this locally, before the RPC reaches the wire, when this
 	// process already closed the ClientConn. The caller is a bystander of
 	// someone else's teardown, and knows nothing about the peer. Its message