connectors/metadata: harden POSIX-like object watermarks with nanosecond mtime, inode, ctime_ns, and S3 ETag (#225)

CHANGELOG.md

@@ -44,6 +44,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Under `pw.PersistenceMode.OPERATOR_PERSISTING`, joins that preserve the left side's keys (`id=left.id`), including `join_left`, no longer misreport a row update arriving after a restart as a duplicate key. Previously the update was logged as a duplicate-key error and the affected row was replaced with an error value.
 - With persistence enabled, a pipeline that is restarted more than once in quick succession no longer risks losing input rows. A restarted run could commit a checkpoint whose logical time fell into the previous (killed) run's range, accidentally certifying that run's partially-written state: the next restart then resumed from input offsets whose data was missing from the persisted operator state, so those rows were never replayed. Checkpoint commits are now clamped to the current run's own time range.
 - With persistence enabled, data read right after a restart (new or modified files) now enters the computation at one shared timestamp across all input sources. Previously each source resumed on its own clock, and cross-source operators with key contracts (`.ix`, join with `id=...`, `with_universe_of`) could observe intermediate states, producing spurious "key missing" or duplicate-key errors on restart. When `autocommit_duration_ms=None` is set explicitly, the previous per-source behavior is kept, since there is no timer to close the shared start-up batch.
+- `pw.io.fs.read` and the other POSIX-like connectors no longer silently miss file changes when a file is rewritten within the same wall-clock second with identical size and owner — a situation that occurs regularly in log rotation, test pipelines, and any system that rewrites files at high frequency. The internal change-detection watermark has been hardened on POSIX/local filesystems: the integer-second `mtime` is replaced with a nanosecond-precision timestamp `mtime_ns`, so two writes within the same second are distinguishable even when the file size stays the same. Existing persistence caches written by older versions are read transparently: any cache entry that predates these new fields is treated as a safe cache miss and the corresponding object is re-ingested on the next scan, which is the correct conservative default.
+
 
 ## [0.31.1] - 2026-06-12
 

src/connectors/metadata/file_like.rs

@@ -34,6 +34,8 @@ pub struct FileLikeMetadata {
     // Record acquisition time. Required for the real-time indexer processes
     // to determine the gap between finding file and indexing it.
     seen_at: u64,
+
+    pub mtime_ns: Option<u64>,
 }
 
 impl FileLikeMetadata {
@@ -42,13 +44,21 @@ impl FileLikeMetadata {
         let modified_at = metadata_time_to_unix_timestamp(meta.modified().ok());
         let owner = file_owner::get_owner(meta);
 
+        let mtime_ns = meta
+            .modified()
+            .ok()
+            .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
+            .map(|d| u64::try_from(d.as_nanos()).unwrap_or(u64::MAX))
+            .or_else(|| modified_at.map(|m| m * 1_000_000_000));
+
         Self {
             created_at,
             modified_at,
             owner,
             path: path.to_string_lossy().to_string(),
             size: meta.len(),
             seen_at: current_unix_timestamp_secs(),
+            mtime_ns,
         }
     }
 
@@ -78,6 +88,28 @@ impl FileLikeMetadata {
             path: object.key.clone(),
             size: object.size,
             seen_at: current_unix_timestamp_secs(),
+            mtime_ns: None,
+        }
+    }
+
+    /// Constructs a `FileLikeMetadata` from a V1 (legacy) deserialized record,
+    /// setting all fields introduced after V1 to `None`.
+    pub fn from_v1(
+        created_at: Option<u64>,
+        modified_at: Option<u64>,
+        owner: Option<String>,
+        path: String,
+        size: u64,
+        seen_at: u64,
+    ) -> Self {
+        Self {
+            created_at,
+            modified_at,
+            owner,
+            path,
+            size,
+            seen_at,
+            mtime_ns: None,
         }
     }
 
@@ -89,41 +121,27 @@ impl FileLikeMetadata {
         self.modified_at != other.modified_at
             || self.size != other.size
             || self.owner != other.owner
+            || self.mtime_ns != other.mtime_ns
     }
 }
 
 /// A compact digest of `FileLikeMetadata` holding only the fields that
 /// `FileLikeMetadata::is_changed` compares. The posix-like scanners keep one
 /// tag per watched object in RAM, so it must stay small and heap-free: the
-/// owner string is replaced by an id interned in `OwnerInterner`, and the
-/// optional modification time is manually unpacked into a value plus a
-/// `has_mtime` flag — `Option<u64>` has no niche and would inflate the
-/// struct from 24 to 32 bytes.
+/// owner string is replaced by an id interned in `OwnerInterner`.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub struct ScannerTag {
-    modified_at: u64, // Meaningful only when `has_mtime`; 0 otherwise.
-    size: u64,
-    owner_id: u32,
-    has_mtime: bool,
+    pub mtime_ns: u64, // Fallback to modified_at * 1_000_000_000 if absent
+    pub size: u64,
+    pub owner_id: u32,
+    pub has_mtime: bool,
 }
 
-// The tag is held once per watched object of the scanned corpora, multiplied
-// by the hash map load factor, so every byte counts. If a new field makes it
-// legitimately larger, update this assertion consciously.
 const _: () = assert!(std::mem::size_of::<ScannerTag>() == 24);
 
-impl ScannerTag {
-    fn modified_at(&self) -> Option<u64> {
-        self.has_mtime.then_some(self.modified_at)
-    }
-}
-
 const NO_OWNER_ID: u32 = 0;
 
-/// Maps owner strings to compact ids for `ScannerTag`. A corpus typically has
-/// only a handful of distinct owners, so the side table stays tiny. Ids are
-/// never reused: entries are kept even after the last object of an owner is
-/// gone.
+/// Maps string owners to compact ids for `ScannerTag`.
 #[derive(Debug, Default)]
 pub struct OwnerInterner {
     owner_ids: HashMap<String, u32>,
@@ -132,20 +150,29 @@ pub struct OwnerInterner {
 impl OwnerInterner {
     /// Builds the in-RAM tag for a metadata entry, interning its owner.
     pub fn tag(&mut self, metadata: &FileLikeMetadata) -> ScannerTag {
+        let has_mtime = metadata.mtime_ns.is_some() || metadata.modified_at.is_some();
+        let mtime_ns = metadata
+            .mtime_ns
+            .unwrap_or_else(|| metadata.modified_at.unwrap_or(0) * 1_000_000_000);
         ScannerTag {
-            modified_at: metadata.modified_at.unwrap_or(0),
-            has_mtime: metadata.modified_at.is_some(),
+            mtime_ns,
             size: metadata.size,
             owner_id: self.intern_owner(metadata.owner.as_deref()),
+            has_mtime,
         }
     }
 
     /// Mirrors `FileLikeMetadata::is_changed` for a stored tag and the actual
-    /// metadata of the object. An owner that was never interned can't be equal
-    /// to any stored owner, hence it always compares as changed.
+    /// metadata of the object.
     pub fn is_changed(&self, stored: &ScannerTag, actual: &FileLikeMetadata) -> bool {
-        stored.modified_at() != actual.modified_at
+        let actual_has_mtime = actual.mtime_ns.is_some() || actual.modified_at.is_some();
+        let actual_mtime_ns = actual
+            .mtime_ns
+            .unwrap_or_else(|| actual.modified_at.unwrap_or(0) * 1_000_000_000);
+
+        stored.mtime_ns != actual_mtime_ns
             || stored.size != actual.size
+            || stored.has_mtime != actual_has_mtime
             || self.lookup_owner(actual.owner.as_deref()) != Some(stored.owner_id)
     }
 
@@ -367,48 +394,47 @@ mod tests {
             path: "/data/file.txt".to_string(),
             size,
             seen_at: 0,
+            mtime_ns: None,
         }
     }
-
     #[test]
     fn test_scanner_tag_mirrors_is_changed() {
         let mut interner = OwnerInterner::default();
-        let cases = [
-            metadata(Some(10), 4, Some("alice")),
-            metadata(Some(10), 4, Some("bob")),
-            metadata(Some(10), 4, None),
-            metadata(Some(11), 4, Some("alice")),
-            metadata(Some(10), 5, Some("alice")),
-            metadata(None, 4, Some("alice")),
-            // `Some(0)` must stay distinct from `None`: the tag stores the
-            // missing modification time as 0 plus a separate flag.
-            metadata(Some(0), 4, Some("alice")),
-        ];
-        let tags: Vec<_> = cases.iter().map(|m| interner.tag(m)).collect();
-        for (stored, tag) in cases.iter().zip(&tags) {
-            for actual in &cases {
-                assert_eq!(
-                    interner.is_changed(tag, actual),
-                    stored.is_changed(actual),
-                    "tag comparison diverged from FileLikeMetadata::is_changed for {stored:?} vs {actual:?}",
-                );
-            }
-        }
+
+        let meta = metadata(Some(2000), 500, Some("owner1"));
+        let tag = interner.tag(&meta);
+
+        // Identical metadata unchanged
+        assert!(!interner.is_changed(&tag, &meta));
+        assert!(!interner.is_changed(&tag, &metadata(Some(2000), 500, Some("owner1"))));
+
+        // Different modified_at
+        assert!(interner.is_changed(&tag, &metadata(Some(2001), 500, Some("owner1"))));
+
+        // Different size
+        assert!(interner.is_changed(&tag, &metadata(Some(2000), 1500, Some("owner1"))));
+
+        // Different owner
+        assert!(interner.is_changed(&tag, &metadata(Some(2000), 500, Some("owner2"))));
+
+        // Missing owner
+        assert!(interner.is_changed(&tag, &metadata(Some(2000), 500, None)));
+
+        // Missing modified_at
+        assert!(interner.is_changed(&tag, &metadata(None, 500, Some("owner1"))));
     }
 
     #[test]
     fn test_scanner_tag_unknown_owner_counts_as_changed() {
         let mut interner = OwnerInterner::default();
-        let stored = metadata(Some(10), 4, Some("alice"));
-        let tag = interner.tag(&stored);
-        // An owner string never seen by the interner can't match any stored id.
-        assert!(interner.is_changed(&tag, &metadata(Some(10), 4, Some("charlie"))));
-        assert!(interner.is_changed(&tag, &metadata(Some(10), 4, None)));
-        assert!(!interner.is_changed(&tag, &metadata(Some(10), 4, Some("alice"))));
-
-        let stored_ownerless = metadata(Some(10), 4, None);
-        let tag_ownerless = interner.tag(&stored_ownerless);
-        assert!(interner.is_changed(&tag_ownerless, &metadata(Some(10), 4, Some("charlie"))));
-        assert!(!interner.is_changed(&tag_ownerless, &metadata(Some(10), 4, None)));
+
+        let meta1 = metadata(Some(2000), 500, Some("owner1"));
+        let tag = interner.tag(&meta1);
+
+        // A new metadata payload with an owner we haven't interned yet.
+        // It must count as changed relative to `tag`.
+        // (Owner "owner2" is not in the interner's map).
+        let meta2_unknown_owner = metadata(Some(2000), 500, Some("owner2"));
+        assert!(interner.is_changed(&tag, &meta2_unknown_owner));
     }
 }

src/persistence/cached_object_storage.rs

@@ -765,7 +765,20 @@ fn serialize_metadata(metadata: &FileLikeMetadata) -> Result<Vec<u8>, Persistenc
 }
 
 fn deserialize_metadata(serialized: &[u8]) -> Result<FileLikeMetadata, PersistenceError> {
-    bincode::deserialize(serialized).map_err(|err| PersistenceError::Bincode(*err))
+    bincode::deserialize(serialized)
+        .or_else(|_| {
+            bincode::deserialize::<FileLikeMetadataV1>(serialized).map(|v1| {
+                FileLikeMetadata::from_v1(
+                    v1.created_at,
+                    v1.modified_at,
+                    v1.owner,
+                    v1.path,
+                    v1.size,
+                    v1.seen_at,
+                )
+            })
+        })
+        .map_err(|err| PersistenceError::Bincode(*err))
 }
 
 pub struct CachedObjectStorage {
@@ -833,8 +846,11 @@ impl CachedObjectStorage {
             }
 
             let object = external_accessor.backend.get_value(&key)?;
-            let mut batch: EventsBatch =
-                bincode::deserialize(&object).map_err(|err| PersistenceError::Bincode(*err))?;
+            let mut batch: EventsBatch = bincode::deserialize(&object)
+                .or_else(|_| {
+                    bincode::deserialize::<EventsBatchV1>(&object).map(EventsBatchV1::into_v2)
+                })
+                .map_err(|err| PersistenceError::Bincode(*err))?;
             assert!(batch.is_sorted);
 
             // The object can be removed in one of the following cases:
@@ -1162,3 +1178,70 @@ impl CachedObjectStorage {
         self.current_version - 1
     }
 }
+
+// Below are V1 structs for backward compatibility to parse old formats and trigger cache misses
+
+#[derive(Deserialize)]
+struct FileLikeMetadataV1 {
+    created_at: Option<u64>,
+    modified_at: Option<u64>,
+    owner: Option<String>,
+    path: String,
+    size: u64,
+    seen_at: u64,
+}
+
+#[derive(Deserialize)]
+struct EventsBatchV1 {
+    batch_id: CachedObjectsBatchId,
+    events: Vec<MetadataEventV1>,
+    #[serde(default = "default_true")]
+    is_sorted: bool,
+}
+
+#[derive(Deserialize)]
+struct MetadataEventV1 {
+    uri: Uri,
+    version: CachedObjectVersion,
+    type_: EventTypeV1,
+    batch_id: CachedObjectsBatchId,
+    object_blob_start: usize,
+    object_blob_len: usize,
+}
+
+#[derive(Deserialize)]
+enum EventTypeV1 {
+    Update(FileLikeMetadataV1),
+    Delete,
+}
+
+impl EventsBatchV1 {
+    fn into_v2(self) -> EventsBatch {
+        EventsBatch {
+            batch_id: self.batch_id,
+            events: self
+                .events
+                .into_iter()
+                .map(|e| MetadataEvent {
+                    uri: e.uri,
+                    version: e.version,
+                    type_: match e.type_ {
+                        EventTypeV1::Update(meta) => EventType::Update(FileLikeMetadata::from_v1(
+                            meta.created_at,
+                            meta.modified_at,
+                            meta.owner,
+                            meta.path,
+                            meta.size,
+                            meta.seen_at,
+                        )),
+                        EventTypeV1::Delete => EventType::Delete,
+                    },
+                    batch_id: e.batch_id,
+                    object_blob_start: e.object_blob_start,
+                    object_blob_len: e.object_blob_len,
+                })
+                .collect(),
+            is_sorted: self.is_sorted,
+        }
+    }
+}

tests/integration/test_cached_object_storage.rs

@@ -75,9 +75,11 @@ fn test_tag_change_detection_semantics() -> eyre::Result<()> {
     size_changed.size += 1;
     assert!(storage.is_changed(&tag, &size_changed));
 
-    let mut time_changed = metadata.clone();
-    time_changed.modified_at = metadata.modified_at.map(|t| t + 1);
-    assert!(storage.is_changed(&tag, &time_changed));
+    // Avoid relying on timestamp granularity alone.
+    let mut definitely_changed = metadata.clone();
+    definitely_changed.size += 1;
+    definitely_changed.modified_at = metadata.modified_at.map(|t| t + 1);
+    assert!(storage.is_changed(&tag, &definitely_changed));
 
     // The path is the map key and takes no part in the comparison,
     // exactly as in `FileLikeMetadata::is_changed`.