filer.sync: repair a destination shorter than the source (#9778)

When the destination's stored mtime is newer than the incoming source version, UpdateEntry skips the update (last-writer-wins). A copy left truncated by an earlier failed replication trips this: the source kept the file's original mtime while the partial copy was written recently, so it looks "newer" and is never corrected. When the destination is strictly shorter than the source, re-replicate the full source content and replace the chunk list instead of skipping. Same shorter-than-source bypass for CreateEntry.

weed/replication/sink/filersink/filer_sink.go

@@ -184,8 +184,14 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
 				return nil
 			}
 			if resp.Entry.Attributes != nil && resp.Entry.Attributes.Mtime >= entry.Attributes.Mtime {
-				glog.V(3).Infof("skip overwriting %s", key)
-				return nil
+				if filer.FileSize(resp.Entry) >= filer.FileSize(entry) {
+					glog.V(3).Infof("skip overwriting %s", key)
+					return nil
+				}
+				// destination is shorter despite a newer mtime: a truncated copy
+				// an earlier failed replication left behind; overwrite from source.
+				glog.Warningf("repair truncated %s: destination %d bytes < source %d bytes but has a newer mtime; overwriting from source",
+					key, filer.FileSize(resp.Entry), filer.FileSize(entry))
 			}
 		}
 
@@ -260,11 +266,32 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 
 	glog.V(4).Infof("oldEntry %+v, newEntry %+v, existingEntry: %+v", oldEntry, newEntry, existingEntry)
 
-	if existingEntry.Attributes.Mtime > newEntry.Attributes.Mtime {
-		// skip if already changed
-		// this usually happens when the messages are not ordered
+	switch chooseUpdateAction(existingEntry, newEntry) {
+	case updateSkip:
+		// a newer, complete version already landed; usually out-of-order messages.
+		// leave the destination untouched — no point rewriting the same entry.
 		glog.V(2).Infof("late updates %s", key)
-	} else {
+		return true, nil
+	case updateRepair:
+		glog.Warningf("repair truncated %s: destination %d bytes < source %d bytes but has a newer mtime; re-replicating full source content",
+			key, filer.FileSize(existingEntry), filer.FileSize(newEntry))
+		replicatedChunks, err := fs.replicateChunks(context.Background(), newEntry.GetChunks(), key, getEntryMtime(newEntry))
+		if err != nil {
+			if errors.Is(err, errChunkSizeMismatch) {
+				glog.Errorf("refuse to replicate entry with corrupt chunk %s: %v", key, err)
+				return true, err
+			}
+			glog.Warningf("replicate entry chunks %s: %v", key, err)
+			return true, nil
+		}
+		existingEntry.Chunks = replicatedChunks
+		existingEntry.Attributes = newEntry.Attributes
+		existingEntry.Extended = newEntry.Extended
+		existingEntry.HardLinkId = newEntry.HardLinkId
+		existingEntry.HardLinkCounter = newEntry.HardLinkCounter
+		existingEntry.Content = newEntry.Content
+		existingEntry.RemoteEntry = newEntry.RemoteEntry
+	default:
 		// source-side chunks resolve via source filer; sink volume IDs may collide.
 		deletedChunks, newChunks, err := compareChunks(context.Background(), filer.LookupFn(fs.filerSource), oldEntry, newEntry)
 		if err != nil {
@@ -339,3 +366,32 @@ func getEntryMtime(entry *filer_pb.Entry) int64 {
 	}
 	return entry.Attributes.Mtime
 }
+
+// updateAction decides how UpdateEntry should reconcile an incoming source
+// version with the destination's current entry.
+type updateAction int
+
+const (
+	// updateNormal applies the incremental source diff: the destination is at
+	// or behind the source version.
+	updateNormal updateAction = iota
+	// updateSkip leaves the destination untouched because a newer, complete
+	// version already landed out of order (last-writer-wins).
+	updateSkip
+	// updateRepair re-replicates the full source content over a destination
+	// that is strictly shorter than the source despite a newer mtime — a
+	// truncated copy an earlier failed replication left behind. Its mtime can
+	// look "newer" (the source preserved an old mtime while the partial copy
+	// was written recently), which would otherwise strand the corruption.
+	updateRepair
+)
+
+func chooseUpdateAction(existing, incoming *filer_pb.Entry) updateAction {
+	if getEntryMtime(existing) <= getEntryMtime(incoming) {
+		return updateNormal
+	}
+	if filer.FileSize(existing) < filer.FileSize(incoming) {
+		return updateRepair
+	}
+	return updateSkip
+}

weed/replication/sink/filersink/filer_sink_test.go

@@ -0,0 +1,41 @@
+package filersink
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+// entry builds a file entry with the given mtime and size for the update-action
+// decision, which only looks at those two attributes.
+func entry(mtime int64, size uint64) *filer_pb.Entry {
+	return &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: mtime, FileSize: size}}
+}
+
+// chooseUpdateAction decides skip vs repair vs normal. A destination left
+// truncated by an earlier failed replication can carry a newer mtime (the
+// source preserved an old mtime while the partial copy was written recently),
+// so a pure mtime guard would strand the corruption. A shorter destination with
+// a newer mtime must be repaired, not skipped.
+func TestChooseUpdateAction(t *testing.T) {
+	tests := []struct {
+		name     string
+		existing *filer_pb.Entry
+		incoming *filer_pb.Entry
+		want     updateAction
+	}{
+		{"nil existing entry", nil, entry(200, 100), updateNormal},
+		{"destination older, catching up", entry(100, 50), entry(200, 100), updateNormal},
+		{"same mtime", entry(200, 50), entry(200, 100), updateNormal},
+		{"destination newer and complete", entry(300, 100), entry(200, 100), updateSkip},
+		{"destination newer and larger", entry(300, 200), entry(200, 100), updateSkip},
+		{"destination newer but truncated", entry(300, 90), entry(200, 100), updateRepair},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := chooseUpdateAction(tc.existing, tc.incoming); got != tc.want {
+				t.Fatalf("chooseUpdateAction = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}