remote sync: keep the event entry intact for IF_ENTRY_EQUAL

weed/command/filer_remote_sync_dir.go

@@ -43,14 +43,6 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour
 
 	var lastLogTsNs = time.Now().UnixNano()
 	processEventFnWithOffset := pb.AddOffsetFunc(func(resp *filer_pb.SubscribeMetadataResponse) error {
-		if resp.EventNotification.NewEntry != nil {
-			if *option.storageClass == "" {
-				delete(resp.EventNotification.NewEntry.Extended, s3_constants.AmzStorageClass)
-			} else {
-				resp.EventNotification.NewEntry.Extended[s3_constants.AmzStorageClass] = []byte(*option.storageClass)
-			}
-		}
-
 		processor.AddSyncJob(resp)
 		return nil
 	}, 3*time.Second, func(counter int64, lastTsNs int64) error {
@@ -177,10 +169,10 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem
 			dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(parentPath, entryName), remoteStorageMountLocation)
 			if message.NewEntry.IsDirectory {
 				glog.V(0).Infof("mkdir  %s", remote_storage.FormatLocation(dest))
-				return client.WriteDirectory(dest, message.NewEntry)
+				return client.WriteDirectory(dest, remoteWriteEntry(message.NewEntry, *option.storageClass))
 			}
 			glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest))
-			remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest)
+			remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, remoteWriteEntry(message.NewEntry, *option.storageClass), dest)
 			if errors.Is(writeErr, errSuperseded) {
 				glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr)
 				return nil
@@ -215,7 +207,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem
 			return client.DeleteFile(dest)
 		}
 		if message.OldEntry != nil && message.NewEntry != nil {
-			return processUpdateEvent(option, filerSource, client, mountedDir, remoteStorageMountLocation, resp)
+			return processUpdateEvent(option, filerSource, *option.storageClass, client, mountedDir, remoteStorageMountLocation, resp)
 		}
 
 		return nil
@@ -226,6 +218,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem
 func processUpdateEvent(
 	filerClient filer_pb.FilerClient,
 	filerSource filer_pb.FilerClient,
+	storageClass string,
 	client remote_storage.RemoteStorageClient,
 	mountedDir string,
 	remoteStorageMountLocation *remote_pb.RemoteStorageLocation,
@@ -246,7 +239,7 @@ func processUpdateEvent(
 		return nil
 	}
 	if message.NewEntry.IsDirectory {
-		return client.WriteDirectory(dest, message.NewEntry)
+		return client.WriteDirectory(dest, remoteWriteEntry(message.NewEntry, storageClass))
 	}
 	if isMetadataOnlyUpdate(resp.Directory, message) {
 		remoteEntry, err := liveRemoteEntry(filerClient, message.NewParentPath, message.NewEntry)
@@ -259,7 +252,7 @@ func processUpdateEvent(
 		}
 		if remoteEntry != nil {
 			glog.V(2).Infof("update meta: %+v", resp)
-			return client.UpdateFileMetadata(dest, message.OldEntry, message.NewEntry)
+			return client.UpdateFileMetadata(dest, message.OldEntry, remoteWriteEntry(message.NewEntry, storageClass))
 		}
 		glog.V(0).Infof("never replicated, uploading %s", remote_storage.FormatLocation(dest))
 	}
@@ -279,7 +272,7 @@ func processUpdateEvent(
 			}
 		}
 	}
-	remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest)
+	remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, remoteWriteEntry(message.NewEntry, storageClass), dest)
 	if errors.Is(writeErr, errSuperseded) {
 		glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr)
 		return nil
@@ -419,6 +412,23 @@ func shouldSendToRemote(entry *filer_pb.Entry) bool {
 	return false
 }
 
+// remoteWriteEntry returns the entry as remote storage should see it: the
+// storage class attribute is dropped, or overridden by -storageClass. The
+// event entry is left untouched so updateLocalEntry still compares the entry
+// the filer stored.
+func remoteWriteEntry(entry *filer_pb.Entry, storageClass string) *filer_pb.Entry {
+	clone := proto.Clone(entry).(*filer_pb.Entry)
+	if storageClass == "" {
+		delete(clone.Extended, s3_constants.AmzStorageClass)
+	} else {
+		if clone.Extended == nil {
+			clone.Extended = map[string][]byte{}
+		}
+		clone.Extended[s3_constants.AmzStorageClass] = []byte(storageClass)
+	}
+	return clone
+}
+
 // updateLocalEntry stamps the entry an event described with its RemoteEntry.
 // The write carries IF_ENTRY_EQUAL over the event's entry: the filer deletes
 // every stored chunk absent from an updated entry, so a snapshot older than

weed/command/filer_remote_sync_dir_test.go

@@ -686,7 +686,7 @@ func TestRenameWithInheritedRemoteEntryWritesNewKey(t *testing.T) {
 
 	remote := &recordingRemote{}
 	filerClient := &stubFilerClient{}
-	if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil {
+	if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil {
 		t.Fatal(err)
 	}
 
@@ -733,7 +733,7 @@ func TestRenameRemoteOnlyEntrySkipsEmptyUpload(t *testing.T) {
 
 	remote := &recordingRemote{}
 	filerClient := &stubFilerClient{}
-	if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil {
+	if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil {
 		t.Fatal(err)
 	}
 	if len(remote.writes) != 0 {
@@ -766,7 +766,7 @@ func TestRenameDeleteOldKeyFailureReturnsError(t *testing.T) {
 	deleteErr := errors.New("AccessDenied: Access Denied")
 	remote := &recordingRemote{deleteErr: deleteErr}
 	filerClient := &stubFilerClient{}
-	err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp)
+	err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp)
 	if !errors.Is(err, deleteErr) {
 		t.Errorf("err = %v, want the delete failure returned so the event is retried", err)
 	}
@@ -802,7 +802,7 @@ func TestRenameDeleteOldKeyNotFoundStillWrites(t *testing.T) {
 
 	remote := &recordingRemote{deleteErr: remote_storage.ErrRemoteObjectNotFound}
 	filerClient := &stubFilerClient{}
-	if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil {
+	if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil {
 		t.Fatalf("err = %v, want nil: an already-deleted old key must not block the write", err)
 	}
 	wantWrite := &remote_pb.RemoteStorageLocation{Name: "gcs", Bucket: "bucket", Path: "/b/dst/probe.bin"}
@@ -829,6 +829,38 @@ func TestIfEntryEqualCarriesTheEventEntry(t *testing.T) {
 	}
 }
 
+// The remote-bound entry loses (or gains) the storage class attribute while
+// the event entry keeps it, so IF_ENTRY_EQUAL still sees the entry the filer
+// stored — otherwise every stamp on an S3-written object reads as stale.
+func TestRemoteWriteEntryLeavesEventEntryUntouched(t *testing.T) {
+	entry := &filer_pb.Entry{
+		Name:       "f",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 1},
+		Extended:   map[string][]byte{s3_constants.AmzStorageClass: []byte("STANDARD")},
+	}
+
+	stripped := remoteWriteEntry(entry, "")
+	if _, ok := stripped.Extended[s3_constants.AmzStorageClass]; ok {
+		t.Fatalf("remote entry still carries %s", s3_constants.AmzStorageClass)
+	}
+	if string(entry.Extended[s3_constants.AmzStorageClass]) != "STANDARD" {
+		t.Fatalf("event entry Extended mutated: %v", entry.Extended)
+	}
+
+	overridden := remoteWriteEntry(entry, "GLACIER")
+	if got := string(overridden.Extended[s3_constants.AmzStorageClass]); got != "GLACIER" {
+		t.Fatalf("override = %q, want GLACIER", got)
+	}
+	if string(entry.Extended[s3_constants.AmzStorageClass]) != "STANDARD" {
+		t.Fatalf("event entry Extended mutated: %v", entry.Extended)
+	}
+
+	bare := &filer_pb.Entry{Name: "g", Attributes: &filer_pb.FuseAttributes{Mtime: 1}}
+	if got := string(remoteWriteEntry(bare, "GLACIER").Extended[s3_constants.AmzStorageClass]); got != "GLACIER" {
+		t.Fatalf("override on nil Extended = %q, want GLACIER", got)
+	}
+}
+
 func TestIsFailedPrecondition(t *testing.T) {
 	refused := status.Error(codes.FailedPrecondition, "precondition failed: /buckets/b/f")
 	cases := []struct {