feast-dev
@@ -19,6 +19,7 @@
 from typing import Any, Dict, List, Optional, Union
 from urllib.parse import urlparse
 
+from google.protobuf.duration_pb2 import Duration
 from google.protobuf.internal.containers import RepeatedCompositeFieldContainer
 from google.protobuf.message import Message
 
@@ -580,20 +581,23 @@ def _update_metadata_fields(
             existing_proto.spec.version = getattr(updated_fv, "version")
 
         # Configuration fields (FeatureView / LabelView TTL)
-        if (
-            hasattr(existing_proto.spec, "ttl")
-            and hasattr(updated_fv, "ttl")
-            and updated_fv.ttl
-        ):
+        # Note: don't gate this on `updated_fv.ttl` being truthy -- None and
+        # timedelta(0) are both the documented way to express "no ttl", and
+        # are falsy, so that check would silently drop the update exactly
+        # when a user clears an existing ttl.
+        if hasattr(existing_proto.spec, "ttl") and hasattr(updated_fv, "ttl"):
             if isinstance(updated_fv, FeatureView):
                 ttl_duration = updated_fv.get_ttl_duration()
-                if ttl_duration:
-                    existing_proto.spec.ttl.CopyFrom(ttl_duration)
+                existing_proto.spec.ttl.CopyFrom(
+                    ttl_duration if ttl_duration is not None else Duration()
+                )
             elif isinstance(updated_fv, LabelView):
-                from google.protobuf.duration_pb2 import Duration
-
                 ttl_duration = Duration()
-                ttl_duration.FromTimedelta(updated_fv.ttl)
+                if updated_fv.ttl is not None:
-                ttl_duration = Duration()
-                ttl_duration.FromTimedelta(updated_fv.ttl)
-                if updated_fv.ttl is not None:
+ttl_duration = Duration()
+if updated_fv.ttl is not None:
+    try:
+        ttl_duration.FromTimedelta(updated_fv.ttl)
+    except (ValueError, OverflowError) as e:
+        raise ValueError(f"Invalid TTL value: {updated_fv.ttl}") from e
-                ttl_duration = Duration()
-                ttl_duration.FromTimedelta(updated_fv.ttl)
-                if updated_fv.ttl is not None:
+ttl_duration = Duration()
+if updated_fv.ttl is not None:
+    try:
+        ttl_duration.FromTimedelta(updated_fv.ttl)
+    except (ValueError, OverflowError) as e:
+        raise ValueError(f"Invalid TTL value: {updated_fv.ttl}") from e
+                    try:
+                        ttl_duration.FromTimedelta(updated_fv.ttl)
+                    except (ValueError, OverflowError) as e:
+                        raise ValueError(f"Invalid TTL value: {updated_fv.ttl}") from e
                 existing_proto.spec.ttl.CopyFrom(ttl_duration)
         if hasattr(existing_proto.spec, "online") and hasattr(updated_fv, "online"):
             existing_proto.spec.online = getattr(updated_fv, "online")

@@ -0,0 +1,53 @@
+"""Unit tests for Registry._update_metadata_fields TTL handling (issue #6703).
+
+Re-applying a FeatureView with its ttl cleared to ``None`` or
+``timedelta(0)`` (both the documented ways to express "no ttl") used to be
+silently dropped, because the update was gated on ``updated_fv.ttl`` being
+truthy -- and both of those values are falsy.
+
+``_update_metadata_fields`` does not use any instance state, so it is exercised
+directly via the class here rather than standing up a full registry backend.
+"""
+
+from datetime import timedelta
+
+import pytest
+
+from feast.entity import Entity
+from feast.feature_view import FeatureView
+from feast.field import Field
+from feast.infra.offline_stores.file_source import FileSource
+from feast.infra.registry.registry import Registry
+from feast.types import Float32
+
+
+def _feature_view(ttl):
+    return FeatureView(
+        name="fv",
+        entities=[Entity(name="e", join_keys=["e_id"])],
+        schema=[Field(name="f1", dtype=Float32)],
+        source=FileSource(path="file://feast/*", timestamp_field="ts_col"),
+        ttl=ttl,
+    )
+
+
+@pytest.mark.parametrize("cleared_ttl", [None, timedelta(0)])
+def test_update_metadata_fields_clears_ttl(cleared_ttl):
+    existing_proto = _feature_view(timedelta(days=10)).to_proto()
+    # sanity: the existing view starts with a finite ttl
+    assert existing_proto.spec.ttl.ToNanoseconds() != 0
+
+    updated_fv = _feature_view(cleared_ttl)
+    Registry._update_metadata_fields(None, existing_proto, updated_fv)
+
+    # the cleared ttl (None / timedelta(0)) must now be reflected as "no ttl"
+    assert existing_proto.spec.ttl.ToNanoseconds() == 0
+
+
+def test_update_metadata_fields_preserves_finite_ttl():
+    existing_proto = _feature_view(timedelta(days=10)).to_proto()
+
+    updated_fv = _feature_view(timedelta(days=3))
+    Registry._update_metadata_fields(None, existing_proto, updated_fv)
+
+    assert existing_proto.spec.ttl.ToTimedelta() == timedelta(days=3)