fix(sync): migrate to using epoch time for sync timestamp

Author: vicheey

samcli/commands/sync/sync_context.py

@@ -7,7 +7,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Dict, Optional, cast
+from typing import Dict, Optional, Union, cast
 
 import tomlkit
 from tomlkit.items import Item
@@ -65,46 +65,45 @@ def update_infra_sync_time(self) -> None:
         self.latest_infra_sync_time = datetime.now(timezone.utc)
 
 
-def _parse_datetime_from_toml(datetime_str: str) -> datetime:
+def _parse_time_from_toml(time_value: Union[str, int, float, datetime]) -> Optional[datetime]:
     """
-    Parse datetime string from TOML file and ensure it's timezone-aware UTC.
+    Parse time from TOML file - supports both epoch and ISO format.
 
-    Handles three formats:
-    1. With explicit timezone: "2024-05-08T15:16:43+00:00"
-    2. Without timezone: "2024-05-08T15:16:43"
-    3. With Z suffix (Zulu time): "2024-05-08T15:16:43Z"
-
-    Python 3.9/3.10 don't support 'Z' suffix in fromisoformat(), so we convert it.
+    Handles legacy ISO format strings as sam migrat to using epoch timestamps for backward compatibility.
+    New writes always use epoch format.
 
     Parameters
     ----------
-    datetime_str: str
-        ISO format datetime string from TOML file
+    time_value: Union[str, int, float, datetime]
+        Either epoch timestamp (new format), ISO format string (legacy), or datetime object
 
     Returns
     -------
-    datetime
-        Timezone-aware datetime in UTC
-
-    Raises
-    ------
-    ValueError
-        If datetime_str is not a valid ISO format datetime string
+    Optional[datetime]
+        Timezone-aware datetime in UTC, or None if parsing fails
     """
     try:
-        # Handle 'Z' suffix for UTC timezone (Python 3.9/3.10 compatibility)
-        if datetime_str.endswith("Z"):
-            datetime_str = datetime_str[:-1] + "+00:00"
+        if isinstance(time_value, datetime):
+            return time_value if time_value.tzinfo else time_value.replace(tzinfo=timezone.utc)
 
-        parsed_datetime = datetime.fromisoformat(datetime_str)
+        if isinstance(time_value, (int, float)):
+            return datetime.fromtimestamp(time_value, tz=timezone.utc)
 
-        # Ensure timezone-aware (handles old sync.toml files without timezone)
-        if parsed_datetime.tzinfo is None:
-            parsed_datetime = parsed_datetime.replace(tzinfo=timezone.utc)
+        if isinstance(time_value, str):
+            if time_value.endswith("Z"):
+                time_value = time_value[:-1] + "+00:00"
+            parsed = datetime.fromisoformat(time_value)
+            return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
 
-        return parsed_datetime
-    except (ValueError, AttributeError) as e:
-        raise ValueError(f"Invalid datetime format in sync.toml: '{datetime_str}'") from e
+        LOG.warning("Invalid time format in sync.toml: %s. Triggering full CloudFormation deployment.", time_value)
+        return None
+    except (ValueError, OSError) as e:
+        LOG.warning(
+            "Failed to parse timestamp from sync.toml: %s. Triggering full CloudFormation deployment. Error: %s",
+            time_value,
+            e,
+        )
+        return None
 
 
 def _sync_state_to_toml_document(sync_state: SyncState) -> TOMLDocument:
@@ -124,7 +123,7 @@ def _sync_state_to_toml_document(sync_state: SyncState) -> TOMLDocument:
     sync_state_toml_table = tomlkit.table()
     sync_state_toml_table[DEPENDENCY_LAYER] = sync_state.dependency_layer
     if sync_state.latest_infra_sync_time:
-        sync_state_toml_table[LATEST_INFRA_SYNC_TIME] = sync_state.latest_infra_sync_time.isoformat()
+        sync_state_toml_table[LATEST_INFRA_SYNC_TIME] = sync_state.latest_infra_sync_time.timestamp()
 
     resource_sync_states_toml_table = tomlkit.table()
     for resource_id in sync_state.resource_sync_states:
@@ -133,7 +132,7 @@ def _sync_state_to_toml_document(sync_state: SyncState) -> TOMLDocument:
         resource_sync_state_toml_table = tomlkit.table()
 
         resource_sync_state_toml_table[HASH] = resource_sync_state.hash_value
-        resource_sync_state_toml_table[SYNC_TIME] = resource_sync_state.sync_time.isoformat()
+        resource_sync_state_toml_table[SYNC_TIME] = resource_sync_state.sync_time.timestamp()
 
         # For Nested stack resources, replace "/" with "-"
         resource_id_toml = resource_id.replace("/", "-")
@@ -173,7 +172,10 @@ def _toml_document_to_sync_state(toml_document: Dict) -> Optional[SyncState]:
             resource_sync_state_toml_table = resource_sync_states_toml_table.get(resource_id)
             sync_time_str = resource_sync_state_toml_table.get(SYNC_TIME)
             # Parse datetime and ensure it's timezone-aware UTC (consistent with how we write)
-            sync_time = _parse_datetime_from_toml(sync_time_str)
+            sync_time = _parse_time_from_toml(sync_time_str)
+            if sync_time is None:
+                # Skip this resource if timestamp is invalid - resource will be re-synced on next sam sync
+                continue
             resource_sync_state = ResourceSyncState(
                 resource_sync_state_toml_table.get(HASH),
                 sync_time,
@@ -189,8 +191,7 @@ def _toml_document_to_sync_state(toml_document: Dict) -> Optional[SyncState]:
         dependency_layer = sync_state_toml_table.get(DEPENDENCY_LAYER)
         latest_infra_sync_time_str = sync_state_toml_table.get(LATEST_INFRA_SYNC_TIME)
         if latest_infra_sync_time_str:
-            # Parse datetime and ensure it's timezone-aware UTC (consistent with how we write)
-            latest_infra_sync_time = _parse_datetime_from_toml(str(latest_infra_sync_time_str))
+            latest_infra_sync_time = _parse_time_from_toml(latest_infra_sync_time_str)
     sync_state = SyncState(dependency_layer, resource_sync_states, latest_infra_sync_time)
 
     return sync_state