1 files changed, 30 insertions, 10 deletions
diff --git a/weed/pb/filer_pb_tail.go b/weed/pb/filer_pb_tail.go
index b373588c9..2d6a7898b 100644
--- a/weed/pb/filer_pb_tail.go
+++ b/weed/pb/filer_pb_tail.go
@@ -5,19 +5,28 @@ import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 	"io"
 	"time"
 )
 
+type EventErrorType int
+
+const (
+	TrivialOnError EventErrorType = iota
+	FatalOnError
+	RetryForeverOnError
+)
+
 type ProcessMetadataFunc func(resp *filer_pb.SubscribeMetadataResponse) error
 
 func FollowMetadata(filerAddress ServerAddress, grpcDialOption grpc.DialOption, clientName string, clientId int32,
-	pathPrefix string, additionalPathPrefixes []string, lastTsNs int64, selfSignature int32,
-	processEventFn ProcessMetadataFunc, fatalOnError bool) error {
+	pathPrefix string, additionalPathPrefixes []string, lastTsNs int64, untilTsNs int64, selfSignature int32,
+	processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error {
 
 	err := WithFilerClient(true, filerAddress, grpcDialOption, makeSubscribeMetadataFunc(clientName, clientId,
-		pathPrefix, additionalPathPrefixes, &lastTsNs, selfSignature, processEventFn, fatalOnError))
+		pathPrefix, additionalPathPrefixes, &lastTsNs, untilTsNs, selfSignature, processEventFn, eventErrorType))
 	if err != nil {
 		return fmt.Errorf("subscribing filer meta change: %v", err)
 	}
@@ -25,11 +34,11 @@ func FollowMetadata(filerAddress ServerAddress, grpcDialOption grpc.DialOption,
 }
 
 func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient,
-	clientName string, clientId int32, pathPrefix string, lastTsNs *int64, selfSignature int32,
-	processEventFn ProcessMetadataFunc, fatalOnError bool) error {
+	clientName string, clientId int32, pathPrefix string, lastTsNs *int64, untilTsNs int64, selfSignature int32,
+	processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error {
 
 	err := filerClient.WithFilerClient(true, makeSubscribeMetadataFunc(clientName, clientId,
-		pathPrefix, nil, lastTsNs, selfSignature, processEventFn, fatalOnError))
+		pathPrefix, nil, lastTsNs, untilTsNs, selfSignature, processEventFn, eventErrorType))
 	if err != nil {
 		return fmt.Errorf("subscribing filer meta change: %v", err)
 	}
@@ -37,8 +46,8 @@ func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient,
 	return nil
 }
 
-func makeSubscribeMetadataFunc(clientName string, clientId int32, pathPrefix string, additionalPathPrefixes []string, lastTsNs *int64, selfSignature int32,
-	processEventFn ProcessMetadataFunc, fatalOnError bool) func(client filer_pb.SeaweedFilerClient) error {
+func makeSubscribeMetadataFunc(clientName string, clientId int32, pathPrefix string, additionalPathPrefixes []string, lastTsNs *int64, untilTsNs int64, selfSignature int32,
+	processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) func(client filer_pb.SeaweedFilerClient) error {
 	return func(client filer_pb.SeaweedFilerClient) error {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
@@ -49,6 +58,7 @@ func makeSubscribeMetadataFunc(clientName string, clientId int32, pathPrefix str
 			SinceNs:      *lastTsNs,
 			Signature:    selfSignature,
 			ClientId:     clientId,
+			UntilNs:      untilTsNs,
 		})
 		if err != nil {
 			return fmt.Errorf("subscribe: %v", err)
@@ -64,9 +74,19 @@ func makeSubscribeMetadataFunc(clientName string, clientId int32, pathPrefix str
 			}
 
 			if err := processEventFn(resp); err != nil {
-				if fatalOnError {
+				switch eventErrorType {
+				case TrivialOnError:
+					glog.Errorf("process %v: %v", resp, err)
+				case FatalOnError:
 					glog.Fatalf("process %v: %v", resp, err)
-				} else {
+				case RetryForeverOnError:
+					util.RetryForever("followMetaUpdates", func() error {
+						return processEventFn(resp)
+					}, func(err error) bool {
+						glog.Errorf("process %v: %v", resp, err)
+						return true
+					})
+				default:
 					glog.Errorf("process %v: %v", resp, err)
 				}
 			}