metadata subscription uses client epoch

1 files changed, 17 insertions, 12 deletions

diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go
index 82261ca51..8d6dd987f 100644
--- a/weed/server/filer_grpc_server_sub_meta.go
+++ b/weed/server/filer_grpc_server_sub_meta.go
@@ -24,11 +24,11 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 
 	peerAddress := findClientAddress(stream.Context(), 0)
 
-	alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId)
+	alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId, req.ClientEpoch)
 	if alreadyKnown {
 		return fmt.Errorf("duplicated subscription detected for client %s id %d", clientName, req.ClientId)
 	}
-	defer fs.deleteClient(clientName, req.ClientId)
+	defer fs.deleteClient(clientName, req.ClientId, req.ClientEpoch)
 
 	lastReadTime := time.Unix(0, req.SinceNs)
 	glog.V(0).Infof(" %v starts to subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
@@ -93,13 +93,13 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 	// use negative client id to differentiate from addClient()/deleteClient() used in SubscribeMetadata()
 	req.ClientId = -req.ClientId
 
-	alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId)
+	alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId, req.ClientEpoch)
 	if alreadyKnown {
 		return fmt.Errorf("duplicated local subscription detected for client %s clientId:%d", clientName, req.ClientId)
 	}
 	defer func() {
 		glog.V(0).Infof(" - %v local subscribe %s clientId:%d", clientName, req.PathPrefix, req.ClientId)
-		fs.deleteClient(clientName, req.ClientId)
+		fs.deleteClient(clientName, req.ClientId, req.ClientEpoch)
 	}()
 
 	lastReadTime := time.Unix(0, req.SinceNs)
@@ -263,25 +263,30 @@ func hasPrefixIn(text string, prefixes []string) bool {
 	return false
 }
 
-func (fs *FilerServer) addClient(clientType string, clientAddress string, clientId int32) (alreadyKnown bool, clientName string) {
+func (fs *FilerServer) addClient(clientType string, clientAddress string, clientId int32, clientEpoch int32) (alreadyKnown bool, clientName string) {
 	clientName = clientType + "@" + clientAddress
 	glog.V(0).Infof("+ listener %v", clientName)
 	if clientId != 0 {
 		fs.knownListenersLock.Lock()
-		_, alreadyKnown = fs.knownListeners[clientId]
-		if !alreadyKnown {
-			fs.knownListeners[clientId] = struct{}{}
+		defer fs.knownListenersLock.Unlock()
+		epoch, found := fs.knownListeners[clientId]
+		if !found || epoch < clientEpoch {
+			fs.knownListeners[clientId] = clientEpoch
+		} else {
+			alreadyKnown = true
 		}
-		fs.knownListenersLock.Unlock()
 	}
 	return
 }
 
-func (fs *FilerServer) deleteClient(clientName string, clientId int32) {
+func (fs *FilerServer) deleteClient(clientName string, clientId int32, clientEpoch int32) {
 	glog.V(0).Infof("- listener %v", clientName)
 	if clientId != 0 {
 		fs.knownListenersLock.Lock()
-		delete(fs.knownListeners, clientId)
-		fs.knownListenersLock.Unlock()
+		defer fs.knownListenersLock.Unlock()
+		epoch, found := fs.knownListeners[clientId]
+		if found && epoch <= clientEpoch {
+			delete(fs.knownListeners, clientId)
+		}
 	}
 }