filer.remote.sync: do not upload an entry whose only content is the remote object (#11890)

* filer.remote.sync: do not upload an entry whose local copy was dropped remote.uncache and the filer's eviction under storage pressure drop a synced entry's chunks and clear LastLocalSyncTsNs; remote.meta.sync does the same when it pulls a newer remote version. When the remote mtime falls in the entry's mtime second, which is the common case for a small file synced promptly, shouldSendToRemote falls back to the cleared stamp and reads the entry as dirty. The dropped chunks make the update a content change, so the chunkless entry was uploaded and a zero-byte object replaced the remote one, the entry's only content. Skip an update that leaves the entry at the same key with no local data and a sized RemoteEntry: the object at that key already is the content. The stamp stays cleared, so remote.cache and eviction still see the entry as uncached. Renames keep going through completeRemoteOnlyRename, metadata-only edits of a remote-only entry still reach the remote, and a rewrite in the remote mtime's second holds local data and is uploaded. * filer.remote.sync: upload a truncate that leaves the entry remote-only A FUSE truncate of a cached, remote-backed file to zero (truncateEntry, or SetAttr with size 0) drops the chunks and keeps the sized RemoteEntry, so the entry reads as remote-only and the dropped-copy skip left the remote object with its old content. Skip only while the entry's size is still the remote object's. Uncache and eviction keep the size and remote.meta.sync sets it from the object, so those stay skipped; a truncate changes the size and is uploaded, in the remote mtime's second too.
2026-10-22 05:54:24 +02:00 · 2026-10-21 11:32:20 -07:00
1 parent f8cfea0dcf
commit 03065155db
@@ -384,6 +384,16 @@ func processUpdateEvent(
 		}
 		glog.V(0).Infof("never replicated, uploading %s", remote_storage.FormatLocation(dest))
 	}
+	if proto.Equal(oldDest, dest) && isDroppedLocalCopy(message.NewEntry) {
+		// The update dropped the local copy: remote.uncache, eviction under
+		// storage pressure, or remote.meta.sync pulling a newer remote
+		// version. Each clears the sync stamp, so shouldSendToRemote reads
+		// the entry as dirty when the remote mtime shares its second. The
+		// content is the object at this very key, and uploading the chunkless
+		// entry would replace it with zero bytes.
+		glog.V(2).Infof("skipping update of remote-only entry: %+v", resp)
+		return nil
+	}
 	glog.V(2).Infof("update: %+v", resp)
 	if !proto.Equal(oldDest, dest) {
 		// A renamed entry that holds no local data now (the snapshot was
@@ -488,6 +498,15 @@ func isRemoteOnly(entry *filer_pb.Entry) bool {
 	return entry != nil && !filer.HasData(entry) && entry.IsInRemoteOnly()
 }

+// isDroppedLocalCopy reports a remote-only entry whose size is still the
+// remote object's: its local copy was dropped and the object describes it.
+// Uncache and eviction keep the size, and remote.meta.sync sets it from the
+// object. A truncate drops the chunks too but changes the size, and the
+// truncated content has to reach the remote.
+func isDroppedLocalCopy(entry *filer_pb.Entry) bool {
+	return isRemoteOnly(entry) && entry.GetAttributes().GetFileSize() == uint64(entry.RemoteEntry.RemoteSize)
+}
+
 // completeRemoteOnlyRename finishes a rename whose content exists only on the
 // remote. When the destination object already holds what the entry's stamp
 // describes (the rename ran before, its offset was not persisted, and
@@ -1722,6 +1722,206 @@ func TestMetadataOnlyUpdateRestampsSyncTs(t *testing.T) {
 	}
 }

+// The droppedCopy fixtures are a small synced file under a mount: a Delta log
+// entry whose remote object was written in the second of its mtime, as a quick
+// sync of a small file is.
+const droppedCopyMtime = 1786096669
+
+var (
+	droppedCopyDir    = "/buckets/gold/sales/_delta_log"
+	droppedCopyName   = "00000000000000000007.json"
+	droppedCopyMount  = &remote_pb.RemoteStorageLocation{Name: "s3", Bucket: "gold", Path: "/"}
+	droppedCopyDest   = &remote_pb.RemoteStorageLocation{Name: "s3", Bucket: "gold", Path: "/sales/_delta_log/" + droppedCopyName}
+	droppedCopySynced = &filer_pb.RemoteEntry{StorageName: "s3", RemoteETag: `"abc"`, RemoteSize: 1024, RemoteMtime: droppedCopyMtime, LastLocalSyncTsNs: droppedCopyMtime*1e9 + 500}
+)
+
+// droppedCopyStamp is the synced stamp with LastLocalSyncTsNs cleared, as
+// remote.uncache and the eviction leave it.
+func droppedCopyStamp() *filer_pb.RemoteEntry {
+	stamp := proto.Clone(droppedCopySynced).(*filer_pb.RemoteEntry)
+	stamp.LastLocalSyncTsNs = 0
+	return stamp
+}
+
+// droppedCopyEntry is the fixture file with the given stamp and chunks, sized
+// as its remote object.
+func droppedCopyEntry(stamp *filer_pb.RemoteEntry, chunks ...*filer_pb.FileChunk) *filer_pb.Entry {
+	entry := entryWith(droppedCopyName, stamp, chunks...)
+	entry.Attributes.FileSize = uint64(droppedCopySynced.RemoteSize)
+	return entry
+}
+
+func droppedCopyUpdate(oldEntry, newEntry *filer_pb.Entry) *filer_pb.SubscribeMetadataResponse {
+	return &filer_pb.SubscribeMetadataResponse{
+		Directory:         droppedCopyDir,
+		EventNotification: &filer_pb.EventNotification{OldEntry: oldEntry, NewParentPath: droppedCopyDir, NewEntry: newEntry},
+	}
+}
+
+// TestDroppedLocalCopyIsNotUploaded: remote.uncache and the filer's eviction
+// under storage pressure drop a synced entry's chunks and clear its sync
+// stamp; remote.meta.sync does the same when it pulls a newer remote version.
+// With the remote mtime in the entry's second, shouldSendToRemote falls back
+// to the cleared stamp and reads the entry as dirty. The object at the same
+// key is the entry's only content, so uploading the chunkless entry would
+// replace it with a zero-byte object. The update is skipped, and the stamp
+// stays cleared so remote.cache and eviction still see the entry as uncached.
+func TestDroppedLocalCopyIsNotUploaded(t *testing.T) {
+	// the eviction's entry, built the way uncacheRemoteEntry builds it, and
+	// the event the way the filer's CreateEntry emits it
+	cached := &filer.Entry{
+		FullPath: util.NewFullPath(droppedCopyDir, droppedCopyName),
+		Attr:     filer.Attr{Mtime: time.Unix(droppedCopyMtime, 0), Mode: 0660, FileSize: 1024},
+		Chunks:   []*filer_pb.FileChunk{chunk("3,01", "e1")},
+		Remote:   droppedCopySynced,
+	}
+	evicted := cached.ShallowClone()
+	evicted.Chunks = nil
+	evicted.Remote = proto.Clone(cached.Remote).(*filer_pb.RemoteEntry)
+	evicted.Remote.LastLocalSyncTsNs = 0
+
+	// remote.meta.sync found a newer object: its mtime becomes the entry's
+	newer := &filer_pb.RemoteEntry{StorageName: "s3", RemoteETag: `"def"`, RemoteSize: 4096, RemoteMtime: droppedCopyMtime + 60}
+	pulled := &filer_pb.Entry{Name: droppedCopyName, Attributes: &filer_pb.FuseAttributes{Mtime: droppedCopyMtime + 60, FileSize: 4096}, RemoteEntry: newer}
+
+	tests := []struct {
+		name     string
+		oldEntry *filer_pb.Entry
+		newEntry *filer_pb.Entry
+	}{
+		{name: "remote.uncache", oldEntry: droppedCopyEntry(droppedCopySynced, chunk("3,01", "e1")), newEntry: droppedCopyEntry(droppedCopyStamp())},
+		{name: "eviction under storage pressure", oldEntry: cached.ToProtoEntry(), newEntry: evicted.ToProtoEntry()},
+		{name: "remote.meta.sync pulls a newer version", oldEntry: entryWith(droppedCopyName, droppedCopySynced, chunk("3,01", "e1")), newEntry: pulled},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			resp := droppedCopyUpdate(tt.oldEntry, tt.newEntry)
+			if !shouldSendToRemote(tt.newEntry) {
+				t.Fatal("precondition: the cleared stamp in the remote mtime's second reads as dirty, the bug's trigger")
+			}
+			if isMetadataOnlyUpdate(resp.Directory, resp.EventNotification) {
+				t.Fatal("precondition: the dropped chunks make the update a content change")
+			}
+
+			remote := &recordingRemote{}
+			filerClient := &stubFilerClient{entry: tt.newEntry}
+			if err := processUpdateEvent(filerClient, filerClient, "", remote, "/buckets/gold", droppedCopyMount, resp); err != nil {
+				t.Fatal(err)
+			}
+			if len(remote.writes) != 0 {
+				t.Errorf("writes = %+v, want none: the remote object is the entry's content and an upload would zero it", remote.writes)
+			}
+			if len(remote.deletes) != 0 || len(remote.metadataUpdates) != 0 {
+				t.Errorf("deletes = %+v, metadata updates = %+v, want the remote object left alone", remote.deletes, remote.metadataUpdates)
+			}
+			if filerClient.updates != 0 {
+				t.Errorf("stamped the entry %d times, want 0: a stamp would mark the uncached entry as cached", filerClient.updates)
+			}
+		})
+	}
+
+	t.Run("a metadata edit of a remote-only entry still reaches the remote", func(t *testing.T) {
+		tagged := droppedCopyEntry(droppedCopyStamp())
+		tagged.Extended = map[string][]byte{"X-Amz-Tagging-tier": []byte("gold")}
+		resp := droppedCopyUpdate(droppedCopyEntry(droppedCopyStamp()), tagged)
+		remote := &recordingRemote{}
+		filerClient := &stubFilerClient{entry: tagged}
+		if err := processUpdateEvent(filerClient, filerClient, "", remote, "/buckets/gold", droppedCopyMount, resp); err != nil {
+			t.Fatal(err)
+		}
+		if len(remote.metadataUpdates) != 1 || len(remote.writes) != 0 {
+			t.Errorf("metadata updates = %+v, writes = %+v, want one metadata update and no upload", remote.metadataUpdates, remote.writes)
+		}
+	})
+}
+
+// TestSameSecondRewriteIsStillUploaded: a rewrite that lands in the remote
+// mtime's second holds local data, so the skip for dropped local copies does
+// not apply and it is uploaded and restamped (#11715).
+func TestSameSecondRewriteIsStillUploaded(t *testing.T) {
+	stale := proto.Clone(droppedCopySynced).(*filer_pb.RemoteEntry)
+	stale.LastLocalSyncTsNs = droppedCopyMtime*1e9 + 100
+	rewrite := func(stamp *filer_pb.RemoteEntry) *filer_pb.Entry {
+		entry := entryWith(droppedCopyName, stamp, chunk("4,02", "e2"))
+		entry.Attributes.MtimeNs = 337
+		return entry
+	}
+
+	tests := []struct {
+		name     string
+		oldEntry *filer_pb.Entry
+		newEntry *filer_pb.Entry
+	}{
+		// Filer.UpdateEntry clears the stamp of a content change
+		{name: "synced entry, stamp cleared by the filer", oldEntry: entryWith(droppedCopyName, droppedCopySynced, chunk("3,01", "e1")), newEntry: rewrite(droppedCopyStamp())},
+		{name: "synced entry, stamp older than the rewrite", oldEntry: entryWith(droppedCopyName, stale, chunk("3,01", "e1")), newEntry: rewrite(stale)},
+		{name: "uncached entry", oldEntry: entryWith(droppedCopyName, droppedCopyStamp()), newEntry: rewrite(droppedCopyStamp())},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.newEntry.Attributes.Mtime != tt.newEntry.RemoteEntry.RemoteMtime {
+				t.Fatal("precondition: the rewrite lands in the remote mtime's second")
+			}
+			remote := &recordingRemote{}
+			filerClient := &stubFilerClient{entry: tt.newEntry}
+			if err := processUpdateEvent(filerClient, filerClient, "", remote, "/buckets/gold", droppedCopyMount, droppedCopyUpdate(tt.oldEntry, tt.newEntry)); err != nil {
+				t.Fatal(err)
+			}
+			if len(remote.writes) != 1 || !proto.Equal(remote.writes[0], droppedCopyDest) {
+				t.Fatalf("writes = %+v, want the rewrite uploaded to %s", remote.writes, remote_storage.FormatLocation(droppedCopyDest))
+			}
+			if !slices.Equal(remote.writtenChunks[0], []string{"4,02"}) {
+				t.Errorf("uploaded chunks %v, want the rewrite's [4,02]", remote.writtenChunks[0])
+			}
+			if filerClient.updates != 1 {
+				t.Errorf("stamped the entry %d times, want 1 after the upload", filerClient.updates)
+			}
+		})
+	}
+}
+
+// TestTruncateToZeroIsUploaded: a FUSE truncate of a cached file to zero
+// (truncateEntry, or SetAttr with size 0) drops the chunks and keeps the
+// sized RemoteEntry, so the entry looks remote-only. Its size is no longer
+// the remote object's, and the empty file has to replace the object.
+func TestTruncateToZeroIsUploaded(t *testing.T) {
+	truncated := func(mtime int64, mtimeNs int32) *filer_pb.Entry {
+		// the filer clears the stamp of a content change
+		entry := entryWith(droppedCopyName, droppedCopyStamp())
+		entry.Attributes.Mtime, entry.Attributes.MtimeNs = mtime, mtimeNs
+		return entry
+	}
+	tests := []struct {
+		name     string
+		newEntry *filer_pb.Entry
+	}{
+		{name: "in a later second", newEntry: truncated(droppedCopyMtime+5, 0)},
+		{name: "in the remote mtime's second", newEntry: truncated(droppedCopyMtime, 337)},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if !isRemoteOnly(tt.newEntry) {
+				t.Fatal("precondition: the truncated entry looks remote-only")
+			}
+			remote := &recordingRemote{}
+			filerClient := &stubFilerClient{entry: tt.newEntry}
+			resp := droppedCopyUpdate(droppedCopyEntry(droppedCopySynced, chunk("3,01", "e1")), tt.newEntry)
+			if err := processUpdateEvent(filerClient, filerClient, "", remote, "/buckets/gold", droppedCopyMount, resp); err != nil {
+				t.Fatal(err)
+			}
+			if len(remote.writes) != 1 || !proto.Equal(remote.writes[0], droppedCopyDest) {
+				t.Fatalf("writes = %+v, want the empty file uploaded to %s", remote.writes, remote_storage.FormatLocation(droppedCopyDest))
+			}
+			if len(remote.writtenChunks[0]) != 0 {
+				t.Errorf("uploaded chunks %v, want none: a zero-byte object", remote.writtenChunks[0])
+			}
+			if filerClient.updates != 1 {
+				t.Errorf("stamped the entry %d times, want 1 after the upload", filerClient.updates)
+			}
+		})
+	}
+}
+
 func TestCompleteRemoteOnlyRenameKeepsAReusedOldKey(t *testing.T) {
 	// A was renamed to B, then D to A before the sync got to the first
 	// rename: the object at A is now D's, and the only copy of it