From 6e4a46c04b1685dff762e5a2aaa8d3300da6fc2d Mon Sep 17 00:00:00 2001 From: Abhi <171412961+iapoorv01@users.noreply.github.com> Date: Tue, 21 Jul 2026 12:17:45 +0530 Subject: [PATCH 1/2] connectors/metadata: harden watermarks with sub-second precision and S3 ETags (fixes #225) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Problem ------- The POSIX-like object tracker was detecting changes by comparing a watermark tuple of (mtime_seconds, size, owner). This is insufficient in at least two well-documented failure modes: 1. Sub-second double-writes on local filesystems: if a file is rewritten twice within the same wall-clock second with an identical size and owner (e.g. log rotation, test pipelines, high-frequency file generators), the second write is invisible to the tracker. The change is silently lost. 2. S3 object replacement: S3 objects can be atomically swapped to a new version with the same last_modified epoch-second and byte size. Without comparing the ETag or VersionId, such swaps go undetected. Solution -------- This commit implements the improvements requested in issue #225 across three independent areas: **1. Sub-second timestamps + inode tie-breaker (POSIX / local fs)** FileLikeMetadata::from_fs_meta now extracts: - mtime_ns — full nanosecond-precision modification time sourced directly from SystemTime::duration_since(UNIX_EPOCH). Falls back to mtime_seconds * 1e9 on platforms where sub-second resolution is unavailable. - inode — st_ino from std::os::unix::fs::MetadataExt. Files sharing the same mtime_ns and size but on different inodes are treated as changed (handles rename-into-place write patterns). - ctime_ns — st_ctime with nanosecond precision. ctime advances on any metadata mutation (chmod, chown, xattr write) even when mtime and size stay constant, acting as a final tie-breaker. All three fields are gated behind #[cfg(unix)] and fall back to None on non-Unix targets (Windows, WASM) so the cross-platform build is unaffected. **2. S3 ETag comparison** FileLikeMetadata::from_s3_object now captures object.e_tag (an Option) alongside the existing last_modified and size. The ETag is an MD5 or multipart checksum computed by S3 itself; it changes whenever the object content changes, regardless of whether the last_modified timestamp or size happens to coincide with the previous version. **3. In-RAM ScannerTag redesigned as a discriminated enum** The old ScannerTag was a flat 24-byte struct holding (modified_at, size, owner_id, has_mtime). It could not represent the new fields without significant bloat. The new ScannerTag is a memory-efficient enum with two variants: - ScannerTag::Posix { mtime_ns, size, identity_id, inode, ctime_ns } - ScannerTag::S3 { last_modified, size, identity_id } The tag variant is chosen at tagging time based on whether the metadata looks like an S3 object (etag is Some, or path starts with s3://). This keeps the hot polling path branchless per-object and avoids any heap allocation. The size assertion is updated from 24 to 40 bytes to document the new conscious tradeoff. OwnerInterner has been renamed IdentityInterner and generalised to intern any string identifier (owner for POSIX, ETag for S3), preserving the O(1) RAM comparison semantics of the original. **4. Backward-compatible persistence layer** CachedObjectStorage serialises EventsBatch blobs with bincode, which is positional — adding fields to FileLikeMetadata or EventType would cause a panic when reading an existing cache written by an older binary. To satisfy the requirement that old cache entries are treated as safe cache misses (not panics), two shadow V1 structs are introduced at the bottom of cached_object_storage.rs: - FileLikeMetadataV1 mirrors the previous six-field struct - EventsBatchV1 / MetadataEventV1 / EventTypeV1 mirror the previous event log format The deserialization callsites (deserialize_metadata and the batch loading loop in load_state) now attempt to decode into the current format first and, on failure, fall back to the V1 format. Decoded V1 values are up-converted by setting all new Option fields to None. When such an up-converted entry is later compared against a freshly polled object (which will have Some(mtime_ns), Some(inode), etc.), the None vs Some mismatch causes is_changed to return true — the entry is re-ingested as if the cache had no entry for that path. This is the correct safe default: we may re-read a file we already have, but we will never silently skip a changed file. Files changed ------------- src/connectors/metadata/file_like.rs - FileLikeMetadata: +mtime_ns, +etag, +inode, +ctime_ns fields - from_fs_meta: extract mtime_ns, inode, ctime_ns via MetadataExt - from_s3_object: capture e_tag - is_changed: incorporate all four new fields - ScannerTag: struct -> enum (Posix | S3) - OwnerInterner -> IdentityInterner (generalised interning) - Unit tests updated to use IdentityInterner src/persistence/cached_object_storage.rs - Import IdentityInterner instead of OwnerInterner - CachedObjectStorage: owner_interner -> identity_interner - deserialize_metadata: V1 fallback deserialization - load_state batch loop: EventsBatchV1 fallback deserialization - Added FileLikeMetadataV1, EventsBatchV1, MetadataEventV1, EventTypeV1 shadow structs for backward compatibility python/pathway/tests/test_common.py - Fix flake8 E231: add missing whitespace after comma in tuple[int,int] type expressions in error-message match strings Testing ------- - cargo fmt --check: passes - isort --check, black --check, flake8: all pass on changed files - cargo test and cargo clippy require the MSVC linker (unavailable in this local Windows environment); full validation deferred to the ubuntu-latest GitHub Actions runners on PR open --- CHANGELOG.md | 2 + python/pathway/tests/test_common.py | 4 +- src/connectors/metadata/file_like.rs | 199 +++++++++++++----- src/persistence/cached_object_storage.rs | 101 ++++++++- .../integration/test_cached_object_storage.rs | 8 +- 5 files changed, 250 insertions(+), 64 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 51cb3f2ba..eb5acab85 100644 --- a/CHANGELOG.md +++ b/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 (`pw.io.s3.read`, `pw.io.minio.read`, etc.) 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 three axes: on POSIX/local filesystems the integer-second `mtime` is replaced with a nanosecond-precision timestamp together with the file's inode and `ctime_ns` as tie-breakers, so two writes within the same second are distinguishable even when the file size stays the same; on S3 and S3-compatible stores the object ETag is now included in the watermark, so an atomic object replacement that preserves `last_modified` and `size` is still detected. 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 diff --git a/python/pathway/tests/test_common.py b/python/pathway/tests/test_common.py index 69f269433..8edfbc4d1 100644 --- a/python/pathway/tests/test_common.py +++ b/python/pathway/tests/test_common.py @@ -5424,7 +5424,7 @@ def test_sequence_get_unchecked_fixed_length_errors(): with pytest.raises( IndexError, match=( - re.escape(f"Index 2 out of range for a tuple of type {tuple[int,int]}.") + re.escape(f"Index 2 out of range for a tuple of type {tuple[int, int]}.") ), ): t2.select(i=pw.this.tup[2]) @@ -5452,7 +5452,7 @@ def test_sequence_get_checked_fixed_length_errors(): with pytest.warns( match=( "(?s)" # make dot match newlines - + re.escape(f"Index 2 out of range for a tuple of type {tuple[int,int]}. ") + + re.escape(f"Index 2 out of range for a tuple of type {tuple[int, int]}. ") + ".*" + re.escape("Consider using just the default value without .get().") ), diff --git a/src/connectors/metadata/file_like.rs b/src/connectors/metadata/file_like.rs index 826d43e63..bc76f7fb2 100644 --- a/src/connectors/metadata/file_like.rs +++ b/src/connectors/metadata/file_like.rs @@ -34,6 +34,11 @@ 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, + pub etag: Option, + pub inode: Option, + pub ctime_ns: Option, } impl FileLikeMetadata { @@ -42,6 +47,28 @@ 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)); + + #[cfg(unix)] + let (inode, ctime_ns) = { + use std::os::unix::fs::MetadataExt; + let ctime_seconds = u64::try_from(meta.ctime()).unwrap_or(0); + let ctime_nanoseconds = u64::try_from(meta.ctime_nsec()).unwrap_or(0); + let ctime_ns = if meta.ctime() >= 0 { + Some(ctime_seconds * 1_000_000_000 + ctime_nanoseconds) + } else { + None + }; + (Some(meta.ino()), ctime_ns) + }; + #[cfg(not(unix))] + let (inode, ctime_ns) = (None, None); + Self { created_at, modified_at, @@ -49,6 +76,10 @@ impl FileLikeMetadata { path: path.to_string_lossy().to_string(), size: meta.len(), seen_at: current_unix_timestamp_secs(), + mtime_ns, + etag: None, + inode, + ctime_ns, } } @@ -78,6 +109,34 @@ impl FileLikeMetadata { path: object.key.clone(), size: object.size, seen_at: current_unix_timestamp_secs(), + mtime_ns: None, + etag: object.e_tag.clone(), + inode: None, + ctime_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, + modified_at: Option, + owner: Option, + path: String, + size: u64, + seen_at: u64, + ) -> Self { + Self { + created_at, + modified_at, + owner, + path, + size, + seen_at, + mtime_ns: None, + etag: None, + inode: None, + ctime_ns: None, } } @@ -89,82 +148,118 @@ impl FileLikeMetadata { self.modified_at != other.modified_at || self.size != other.size || self.owner != other.owner + || self.mtime_ns != other.mtime_ns + || self.etag != other.etag + || self.inode != other.inode + || self.ctime_ns != other.ctime_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` has no niche and would inflate the -/// struct from 24 to 32 bytes. +/// string identifiers (owner/etag) are replaced by an id interned in +/// `IdentityInterner`. #[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 enum ScannerTag { + Posix { + mtime_ns: u64, // Fallback to modified_at * 1_000_000_000 if absent + size: u64, + identity_id: u32, + inode: u64, + ctime_ns: u64, + }, + S3 { + last_modified: u64, + size: u64, + identity_id: u32, + }, } -// 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::() == 24); - -impl ScannerTag { - fn modified_at(&self) -> Option { - self.has_mtime.then_some(self.modified_at) - } -} +const _: () = assert!(std::mem::size_of::() == 40); -const NO_OWNER_ID: u32 = 0; +const NO_IDENTITY_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 identity strings (owner or etag) to compact ids for `ScannerTag`. #[derive(Debug, Default)] -pub struct OwnerInterner { - owner_ids: HashMap, +pub struct IdentityInterner { + identity_ids: HashMap, } -impl OwnerInterner { - /// Builds the in-RAM tag for a metadata entry, interning its owner. +impl IdentityInterner { + /// Builds the in-RAM tag for a metadata entry, interning its identity. pub fn tag(&mut self, metadata: &FileLikeMetadata) -> ScannerTag { - ScannerTag { - modified_at: metadata.modified_at.unwrap_or(0), - has_mtime: metadata.modified_at.is_some(), - size: metadata.size, - owner_id: self.intern_owner(metadata.owner.as_deref()), + if metadata.path.starts_with("s3://") + || metadata.path.starts_with("s3a://") + || metadata.etag.is_some() + { + ScannerTag::S3 { + last_modified: metadata.modified_at.unwrap_or(0), + size: metadata.size, + identity_id: self.intern_identity(metadata.etag.as_deref()), + } + } else { + let mtime_ns = metadata + .mtime_ns + .unwrap_or_else(|| metadata.modified_at.unwrap_or(0) * 1_000_000_000); + ScannerTag::Posix { + mtime_ns, + size: metadata.size, + identity_id: self.intern_identity(metadata.owner.as_deref()), + inode: metadata.inode.unwrap_or(0), + ctime_ns: metadata.ctime_ns.unwrap_or(0), + } } } /// 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 - || stored.size != actual.size - || self.lookup_owner(actual.owner.as_deref()) != Some(stored.owner_id) + match stored { + ScannerTag::Posix { + mtime_ns, + size, + identity_id, + inode, + ctime_ns, + } => { + let actual_mtime_ns = actual + .mtime_ns + .unwrap_or_else(|| actual.modified_at.unwrap_or(0) * 1_000_000_000); + *mtime_ns != actual_mtime_ns + || *size != actual.size + || *inode != actual.inode.unwrap_or(0) + || *ctime_ns != actual.ctime_ns.unwrap_or(0) + || self.lookup_identity(actual.owner.as_deref()) != Some(*identity_id) + } + ScannerTag::S3 { + last_modified, + size, + identity_id, + } => { + *last_modified != actual.modified_at.unwrap_or(0) + || *size != actual.size + || self.lookup_identity(actual.etag.as_deref()) != Some(*identity_id) + } + } } - fn intern_owner(&mut self, owner: Option<&str>) -> u32 { - let Some(owner) = owner else { - return NO_OWNER_ID; + fn intern_identity(&mut self, identity: Option<&str>) -> u32 { + let Some(identity) = identity else { + return NO_IDENTITY_ID; }; - if let Some(id) = self.owner_ids.get(owner) { + if let Some(id) = self.identity_ids.get(identity) { return *id; } - let id = u32::try_from(self.owner_ids.len() + 1).expect("too many distinct owners"); - self.owner_ids.insert(owner.to_string(), id); + let id = u32::try_from(self.identity_ids.len() + 1).expect("too many distinct identities"); + self.identity_ids.insert(identity.to_string(), id); id } - fn lookup_owner(&self, owner: Option<&str>) -> Option { - match owner { - None => Some(NO_OWNER_ID), - Some(owner) => self.owner_ids.get(owner).copied(), + fn lookup_identity(&self, identity: Option<&str>) -> Option { + match identity { + None => Some(NO_IDENTITY_ID), + Some(identity) => self.identity_ids.get(identity).copied(), } } } @@ -367,12 +462,16 @@ mod tests { path: "/data/file.txt".to_string(), size, seen_at: 0, + mtime_ns: None, + etag: None, + inode: None, + ctime_ns: None, } } #[test] fn test_scanner_tag_mirrors_is_changed() { - let mut interner = OwnerInterner::default(); + let mut interner = IdentityInterner::default(); let cases = [ metadata(Some(10), 4, Some("alice")), metadata(Some(10), 4, Some("bob")), @@ -398,7 +497,7 @@ mod tests { #[test] fn test_scanner_tag_unknown_owner_counts_as_changed() { - let mut interner = OwnerInterner::default(); + let mut interner = IdentityInterner::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. diff --git a/src/persistence/cached_object_storage.rs b/src/persistence/cached_object_storage.rs index 8cdfdb232..26da03119 100644 --- a/src/persistence/cached_object_storage.rs +++ b/src/persistence/cached_object_storage.rs @@ -14,7 +14,7 @@ use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; use tempfile::TempDir; -use crate::connectors::metadata::file_like::{OwnerInterner, ScannerTag}; +use crate::connectors::metadata::file_like::{IdentityInterner, ScannerTag}; use crate::connectors::metadata::FileLikeMetadata; use crate::persistence::backends::{Error as PersistenceError, PersistenceBackend}; @@ -765,7 +765,20 @@ fn serialize_metadata(metadata: &FileLikeMetadata) -> Result, Persistenc } fn deserialize_metadata(serialized: &[u8]) -> Result { - bincode::deserialize(serialized).map_err(|err| PersistenceError::Bincode(*err)) + bincode::deserialize(serialized) + .or_else(|_| { + bincode::deserialize::(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 { @@ -779,7 +792,7 @@ pub struct CachedObjectStorage { // bytes per key header). The full metadata is kept on disk, in // `objects_snapshot`, next to the object contents. metadata_snapshot: HashMap, ScannerTag>, - owner_interner: OwnerInterner, + identity_interner: IdentityInterner, objects_snapshot: SqliteObjectsSnapshot, current_version: CachedObjectVersion, @@ -793,7 +806,7 @@ impl CachedObjectStorage { EMPTY_STORAGE_BATCH_ID + 1, ))), metadata_snapshot: HashMap::new(), - owner_interner: OwnerInterner::default(), + identity_interner: IdentityInterner::default(), objects_snapshot: SqliteObjectsSnapshot::new()?, current_version: EMPTY_STORAGE_VERSION + 1, }) @@ -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::(&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: @@ -938,7 +954,7 @@ impl CachedObjectStorage { /// Mirrors `FileLikeMetadata::is_changed` for a stored tag and the /// actual metadata of the corresponding object. pub fn is_changed(&self, stored: &ScannerTag, actual: &FileLikeMetadata) -> bool { - self.owner_interner.is_changed(stored, actual) + self.identity_interner.is_changed(stored, actual) } /// Reads the full stored copy of an object — contents and metadata — @@ -977,7 +993,7 @@ impl CachedObjectStorage { EventType::Update(_) => { let batch_id = event.batch_id; let blob_segment = event.into_blob_segment(); - let tag = self.owner_interner.tag(&blob_segment.metadata); + let tag = self.identity_interner.tag(&blob_segment.metadata); self.metadata_snapshot .insert(Box::from(blob_segment.uri.as_slice()), tag); segments_for_download @@ -1143,7 +1159,7 @@ impl CachedObjectStorage { ) -> Result<(), PersistenceError> { match event.type_ { EventType::Update(metadata) => { - let tag = self.owner_interner.tag(&metadata); + let tag = self.identity_interner.tag(&metadata); self.objects_snapshot .insert(&event.uri, contents, &metadata)?; self.metadata_snapshot @@ -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, + modified_at: Option, + owner: Option, + path: String, + size: u64, + seen_at: u64, +} + +#[derive(Deserialize)] +struct EventsBatchV1 { + batch_id: CachedObjectsBatchId, + events: Vec, + #[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, + } + } +} diff --git a/tests/integration/test_cached_object_storage.rs b/tests/integration/test_cached_object_storage.rs index 9a5b3da3e..ff59611fe 100644 --- a/tests/integration/test_cached_object_storage.rs +++ b/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`. From 87cbf481495d737e55c85f916f051087baeb3c64 Mon Sep 17 00:00:00 2001 From: Abhi <171412961+iapoorv01@users.noreply.github.com> Date: Thu, 23 Jul 2026 11:21:51 +0530 Subject: [PATCH 2/2] connectors/metadata: harden POSIX watermarks with mtime_ns (fixes #225) --- CHANGELOG.md | 2 +- python/pathway/tests/test_common.py | 4 +- src/connectors/metadata/file_like.rs | 221 ++++++++--------------- src/persistence/cached_object_storage.rs | 12 +- 4 files changed, 83 insertions(+), 156 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eb5acab85..0a042e126 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,7 +44,7 @@ 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 (`pw.io.s3.read`, `pw.io.minio.read`, etc.) 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 three axes: on POSIX/local filesystems the integer-second `mtime` is replaced with a nanosecond-precision timestamp together with the file's inode and `ctime_ns` as tie-breakers, so two writes within the same second are distinguishable even when the file size stays the same; on S3 and S3-compatible stores the object ETag is now included in the watermark, so an atomic object replacement that preserves `last_modified` and `size` is still detected. 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. +- `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 diff --git a/python/pathway/tests/test_common.py b/python/pathway/tests/test_common.py index 8edfbc4d1..69f269433 100644 --- a/python/pathway/tests/test_common.py +++ b/python/pathway/tests/test_common.py @@ -5424,7 +5424,7 @@ def test_sequence_get_unchecked_fixed_length_errors(): with pytest.raises( IndexError, match=( - re.escape(f"Index 2 out of range for a tuple of type {tuple[int, int]}.") + re.escape(f"Index 2 out of range for a tuple of type {tuple[int,int]}.") ), ): t2.select(i=pw.this.tup[2]) @@ -5452,7 +5452,7 @@ def test_sequence_get_checked_fixed_length_errors(): with pytest.warns( match=( "(?s)" # make dot match newlines - + re.escape(f"Index 2 out of range for a tuple of type {tuple[int, int]}. ") + + re.escape(f"Index 2 out of range for a tuple of type {tuple[int,int]}. ") + ".*" + re.escape("Consider using just the default value without .get().") ), diff --git a/src/connectors/metadata/file_like.rs b/src/connectors/metadata/file_like.rs index bc76f7fb2..2243b7539 100644 --- a/src/connectors/metadata/file_like.rs +++ b/src/connectors/metadata/file_like.rs @@ -36,9 +36,6 @@ pub struct FileLikeMetadata { seen_at: u64, pub mtime_ns: Option, - pub etag: Option, - pub inode: Option, - pub ctime_ns: Option, } impl FileLikeMetadata { @@ -54,21 +51,6 @@ impl FileLikeMetadata { .map(|d| u64::try_from(d.as_nanos()).unwrap_or(u64::MAX)) .or_else(|| modified_at.map(|m| m * 1_000_000_000)); - #[cfg(unix)] - let (inode, ctime_ns) = { - use std::os::unix::fs::MetadataExt; - let ctime_seconds = u64::try_from(meta.ctime()).unwrap_or(0); - let ctime_nanoseconds = u64::try_from(meta.ctime_nsec()).unwrap_or(0); - let ctime_ns = if meta.ctime() >= 0 { - Some(ctime_seconds * 1_000_000_000 + ctime_nanoseconds) - } else { - None - }; - (Some(meta.ino()), ctime_ns) - }; - #[cfg(not(unix))] - let (inode, ctime_ns) = (None, None); - Self { created_at, modified_at, @@ -77,9 +59,6 @@ impl FileLikeMetadata { size: meta.len(), seen_at: current_unix_timestamp_secs(), mtime_ns, - etag: None, - inode, - ctime_ns, } } @@ -110,9 +89,6 @@ impl FileLikeMetadata { size: object.size, seen_at: current_unix_timestamp_secs(), mtime_ns: None, - etag: object.e_tag.clone(), - inode: None, - ctime_ns: None, } } @@ -134,9 +110,6 @@ impl FileLikeMetadata { size, seen_at, mtime_ns: None, - etag: None, - inode: None, - ctime_ns: None, } } @@ -149,117 +122,76 @@ impl FileLikeMetadata { || self.size != other.size || self.owner != other.owner || self.mtime_ns != other.mtime_ns - || self.etag != other.etag - || self.inode != other.inode - || self.ctime_ns != other.ctime_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 -/// string identifiers (owner/etag) are replaced by an id interned in -/// `IdentityInterner`. +/// owner string is replaced by an id interned in `OwnerInterner`. #[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub enum ScannerTag { - Posix { - mtime_ns: u64, // Fallback to modified_at * 1_000_000_000 if absent - size: u64, - identity_id: u32, - inode: u64, - ctime_ns: u64, - }, - S3 { - last_modified: u64, - size: u64, - identity_id: u32, - }, +pub struct ScannerTag { + 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, } -const _: () = assert!(std::mem::size_of::() == 40); +const _: () = assert!(std::mem::size_of::() == 24); -const NO_IDENTITY_ID: u32 = 0; +const NO_OWNER_ID: u32 = 0; -/// Maps identity strings (owner or etag) to compact ids for `ScannerTag`. +/// Maps string owners to compact ids for `ScannerTag`. #[derive(Debug, Default)] -pub struct IdentityInterner { - identity_ids: HashMap, +pub struct OwnerInterner { + owner_ids: HashMap, } -impl IdentityInterner { - /// Builds the in-RAM tag for a metadata entry, interning its identity. +impl OwnerInterner { + /// Builds the in-RAM tag for a metadata entry, interning its owner. pub fn tag(&mut self, metadata: &FileLikeMetadata) -> ScannerTag { - if metadata.path.starts_with("s3://") - || metadata.path.starts_with("s3a://") - || metadata.etag.is_some() - { - ScannerTag::S3 { - last_modified: metadata.modified_at.unwrap_or(0), - size: metadata.size, - identity_id: self.intern_identity(metadata.etag.as_deref()), - } - } else { - let mtime_ns = metadata - .mtime_ns - .unwrap_or_else(|| metadata.modified_at.unwrap_or(0) * 1_000_000_000); - ScannerTag::Posix { - mtime_ns, - size: metadata.size, - identity_id: self.intern_identity(metadata.owner.as_deref()), - inode: metadata.inode.unwrap_or(0), - ctime_ns: metadata.ctime_ns.unwrap_or(0), - } + 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 { + 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. pub fn is_changed(&self, stored: &ScannerTag, actual: &FileLikeMetadata) -> bool { - match stored { - ScannerTag::Posix { - mtime_ns, - size, - identity_id, - inode, - ctime_ns, - } => { - let actual_mtime_ns = actual - .mtime_ns - .unwrap_or_else(|| actual.modified_at.unwrap_or(0) * 1_000_000_000); - *mtime_ns != actual_mtime_ns - || *size != actual.size - || *inode != actual.inode.unwrap_or(0) - || *ctime_ns != actual.ctime_ns.unwrap_or(0) - || self.lookup_identity(actual.owner.as_deref()) != Some(*identity_id) - } - ScannerTag::S3 { - last_modified, - size, - identity_id, - } => { - *last_modified != actual.modified_at.unwrap_or(0) - || *size != actual.size - || self.lookup_identity(actual.etag.as_deref()) != Some(*identity_id) - } - } + 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) } - fn intern_identity(&mut self, identity: Option<&str>) -> u32 { - let Some(identity) = identity else { - return NO_IDENTITY_ID; + fn intern_owner(&mut self, owner: Option<&str>) -> u32 { + let Some(owner) = owner else { + return NO_OWNER_ID; }; - if let Some(id) = self.identity_ids.get(identity) { + if let Some(id) = self.owner_ids.get(owner) { return *id; } - let id = u32::try_from(self.identity_ids.len() + 1).expect("too many distinct identities"); - self.identity_ids.insert(identity.to_string(), id); + let id = u32::try_from(self.owner_ids.len() + 1).expect("too many distinct owners"); + self.owner_ids.insert(owner.to_string(), id); id } - fn lookup_identity(&self, identity: Option<&str>) -> Option { - match identity { - None => Some(NO_IDENTITY_ID), - Some(identity) => self.identity_ids.get(identity).copied(), + fn lookup_owner(&self, owner: Option<&str>) -> Option { + match owner { + None => Some(NO_OWNER_ID), + Some(owner) => self.owner_ids.get(owner).copied(), } } } @@ -463,51 +395,46 @@ mod tests { size, seen_at: 0, mtime_ns: None, - etag: None, - inode: None, - ctime_ns: None, } } - #[test] fn test_scanner_tag_mirrors_is_changed() { - let mut interner = IdentityInterner::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 mut interner = OwnerInterner::default(); + + 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 = IdentityInterner::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 mut interner = OwnerInterner::default(); + + 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)); } } diff --git a/src/persistence/cached_object_storage.rs b/src/persistence/cached_object_storage.rs index 26da03119..b9ffcbc4e 100644 --- a/src/persistence/cached_object_storage.rs +++ b/src/persistence/cached_object_storage.rs @@ -14,7 +14,7 @@ use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; use tempfile::TempDir; -use crate::connectors::metadata::file_like::{IdentityInterner, ScannerTag}; +use crate::connectors::metadata::file_like::{OwnerInterner, ScannerTag}; use crate::connectors::metadata::FileLikeMetadata; use crate::persistence::backends::{Error as PersistenceError, PersistenceBackend}; @@ -792,7 +792,7 @@ pub struct CachedObjectStorage { // bytes per key header). The full metadata is kept on disk, in // `objects_snapshot`, next to the object contents. metadata_snapshot: HashMap, ScannerTag>, - identity_interner: IdentityInterner, + owner_interner: OwnerInterner, objects_snapshot: SqliteObjectsSnapshot, current_version: CachedObjectVersion, @@ -806,7 +806,7 @@ impl CachedObjectStorage { EMPTY_STORAGE_BATCH_ID + 1, ))), metadata_snapshot: HashMap::new(), - identity_interner: IdentityInterner::default(), + owner_interner: OwnerInterner::default(), objects_snapshot: SqliteObjectsSnapshot::new()?, current_version: EMPTY_STORAGE_VERSION + 1, }) @@ -954,7 +954,7 @@ impl CachedObjectStorage { /// Mirrors `FileLikeMetadata::is_changed` for a stored tag and the /// actual metadata of the corresponding object. pub fn is_changed(&self, stored: &ScannerTag, actual: &FileLikeMetadata) -> bool { - self.identity_interner.is_changed(stored, actual) + self.owner_interner.is_changed(stored, actual) } /// Reads the full stored copy of an object — contents and metadata — @@ -993,7 +993,7 @@ impl CachedObjectStorage { EventType::Update(_) => { let batch_id = event.batch_id; let blob_segment = event.into_blob_segment(); - let tag = self.identity_interner.tag(&blob_segment.metadata); + let tag = self.owner_interner.tag(&blob_segment.metadata); self.metadata_snapshot .insert(Box::from(blob_segment.uri.as_slice()), tag); segments_for_download @@ -1159,7 +1159,7 @@ impl CachedObjectStorage { ) -> Result<(), PersistenceError> { match event.type_ { EventType::Update(metadata) => { - let tag = self.identity_interner.tag(&metadata); + let tag = self.owner_interner.tag(&metadata); self.objects_snapshot .insert(&event.uri, contents, &metadata)?; self.metadata_snapshot