From 6c7fe40305fdebe0902e44ae2a841ca4382b6a9b Mon Sep 17 00:00:00 2001
From: chrislu <chris.lu@gmail.com>
Date: Mon, 19 Dec 2022 11:31:58 -0800
Subject: [PATCH] filer sink retries reading file chunks, skipping missing
 chunks

if the file chunk is not available during replication time, the file is skipped
---
 weed/replication/sink/filersink/fetch_write.go | 16 ++++++++++------
 weed/replication/sink/filersink/filer_sink.go  |  8 +++++---
 2 files changed, 15 insertions(+), 9 deletions(-)

diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go
index cd961f147..7d526513e 100644
--- a/weed/replication/sink/filersink/fetch_write.go
+++ b/weed/replication/sink/filersink/fetch_write.go
@@ -27,12 +27,16 @@ func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path st
 		index, source := chunkIndex, sourceChunk
 		fs.executor.Execute(func() {
 			defer wg.Done()
-			replicatedChunk, e := fs.replicateOneChunk(source, path)
-			if e != nil {
-				err = e
-				return
-			}
-			replicatedChunks[index] = replicatedChunk
+			util.Retry("replicate chunks", func() error {
+				replicatedChunk, e := fs.replicateOneChunk(source, path)
+				if e != nil {
+					err = e
+					return e
+				}
+				replicatedChunks[index] = replicatedChunk
+				err = nil
+				return nil
+			})
 		})
 	}
 	wg.Wait()
diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go
index de5ff55cc..35b6ffa73 100644
--- a/weed/replication/sink/filersink/filer_sink.go
+++ b/weed/replication/sink/filersink/filer_sink.go
@@ -112,7 +112,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
 			Directory: dir,
 			Name:      name,
 		}
-		glog.V(1).Infof("lookup: %v", lookupRequest)
+		// glog.V(1).Infof("lookup: %v", lookupRequest)
 		if resp, err := filer_pb.LookupEntry(client, lookupRequest); err == nil {
 			if filer.ETag(resp.Entry) == filer.ETag(entry) {
 				glog.V(3).Infof("already replicated %s", key)
@@ -125,9 +125,10 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
 		if err != nil {
 			// only warning here since the source chunk may have been deleted already
 			glog.Warningf("replicate entry chunks %s: %v", key, err)
+			return nil
 		}
 
-		glog.V(4).Infof("replicated %s %+v ===> %+v", key, entry.GetChunks(), replicatedChunks)
+		// glog.V(4).Infof("replicated %s %+v ===> %+v", key, entry.GetChunks(), replicatedChunks)
 
 		request := &filer_pb.CreateEntryRequest{
 			Directory: dir,
@@ -205,7 +206,8 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 		// replicate the chunks that are new in the source
 		replicatedChunks, err := fs.replicateChunks(newChunks, key)
 		if err != nil {
-			return true, fmt.Errorf("replicate %s chunks error: %v", key, err)
+			glog.Warningf("replicate entry chunks %s: %v", key, err)
+			return true, nil
 		}
 		existingEntry.Chunks = append(existingEntry.GetChunks(), replicatedChunks...)
 		existingEntry.Attributes = newEntry.Attributes