administrator/bak-and-rest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Riduce falsi positivi del watchdog

Browse Source
This commit is contained in:
allebonvi
2026-07-02 09:59:49 +02:00
parent ffdbba2655
commit 2f0a5d53cf
4 changed files with 133 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
src/bakrest/registry.py
View File

@@ -13,6 +13,7 @@ class ChangeEvent:
event_type: str
old_path: str | None = None
size: int | None = None
mtime_ns: int | None = None
class ChangeRegistry:
@@ -26,19 +27,66 @@ class ChangeRegistry:
with closing(self._connect()) as connection:
connection.execute(
"""
INSERT INTO changes(path, event_type, old_path, size, status, created_at, updated_at)
VALUES (?, ?, ?, ?, 'pending', ?, ?)
INSERT INTO changes(path, event_type, old_path, size, mtime_ns, status, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, 'pending', ?, ?)
ON CONFLICT(path) DO UPDATE SET
event_type = excluded.event_type,
old_path = excluded.old_path,
size = excluded.size,
mtime_ns = excluded.mtime_ns,
status = 'pending',
updated_at = excluded.updated_at
""",
(event.path, event.event_type, event.old_path, event.size, timestamp, timestamp),
(
event.path,
event.event_type,
event.old_path,
event.size,
event.mtime_ns,
timestamp,
timestamp,
),
)
if event.event_type != "deleted":
self.update_snapshot(event.path, event.size, event.mtime_ns, connection)
connection.commit()
def has_snapshot_changed(self, path: str, size: int | None, mtime_ns: int | None) -> bool:
with closing(self._connect()) as connection:
row = connection.execute(
"SELECT size, mtime_ns FROM file_snapshots WHERE path = ?",
(path,),
).fetchone()
if row is None:
return True
return row[0] != size or row[1] != mtime_ns
def update_snapshot(
self,
path: str,
size: int | None,
mtime_ns: int | None,
connection: sqlite3.Connection | None = None,
) -> None:
timestamp = datetime.now(timezone.utc).isoformat()
if connection is None:
with closing(self._connect()) as local_connection:
self.update_snapshot(path, size, mtime_ns, local_connection)
local_connection.commit()
return
connection.execute(
"""
INSERT INTO file_snapshots(path, size, mtime_ns, updated_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(path) DO UPDATE SET
size = excluded.size,
mtime_ns = excluded.mtime_ns,
updated_at = excluded.updated_at
""",
(path, size, mtime_ns, timestamp),
)
def pending_count(self) -> int:
with closing(self._connect()) as connection:
row = connection.execute(
@@ -50,14 +98,14 @@ class ChangeRegistry:
with closing(self._connect()) as connection:
rows = connection.execute(
"""
SELECT path, event_type, old_path, size
SELECT path, event_type, old_path, size, mtime_ns
FROM changes
WHERE status = 'pending'
ORDER BY updated_at ASC
"""
).fetchall()
return [
ChangeEvent(path=row[0], event_type=row[1], old_path=row[2], size=row[3])
ChangeEvent(path=row[0], event_type=row[1], old_path=row[2], size=row[3], mtime_ns=row[4])
for row in rows
]
@@ -90,12 +138,24 @@ class ChangeRegistry:
event_type TEXT NOT NULL,
old_path TEXT,
size INTEGER,
mtime_ns INTEGER,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
_ensure_column(connection, "changes", "mtime_ns", "INTEGER")
connection.execute(
"""
CREATE TABLE IF NOT EXISTS file_snapshots (
path TEXT NOT NULL PRIMARY KEY,
size INTEGER,
mtime_ns INTEGER,
updated_at TEXT NOT NULL
)
"""
)
connection.execute(
"CREATE INDEX IF NOT EXISTS idx_changes_status ON changes(status)"
)
@@ -103,3 +163,12 @@ class ChangeRegistry:
def _connect(self) -> sqlite3.Connection:
return sqlite3.connect(self.db_path)
def _ensure_column(connection: sqlite3.Connection, table: str, column: str, definition: str) -> None:
columns = {
row[1]
for row in connection.execute(f"PRAGMA table_info({table})").fetchall()
}
if column not in columns:
connection.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}")

32
src/bakrest/watcher.py
View File

@@ -35,6 +35,7 @@ class WatchdogRuntime:
watched = 0
for include_dir in self.config.watch.include_dirs:
if include_dir.exists():
self._prime_snapshots(include_dir)
observer.schedule(Handler(), str(include_dir), recursive=True)
self.logger.info("Monitoring directory: %s", include_dir)
watched += 1
@@ -70,12 +71,21 @@ class WatchdogRuntime:
if not self.filter.should_track(target_path, is_directory=is_directory):
return
size = _safe_size(target_path)
size, mtime_ns = _safe_snapshot(target_path)
if event_type == "modified" and not self.registry.has_snapshot_changed(
str(target_path),
size,
mtime_ns,
):
self.logger.info("Ignored metadata-only modified event: %s", target_path)
return
change = ChangeEvent(
path=str(target_path),
event_type=event_type,
old_path=str(src_path) if dest_path_value else None,
size=size,
mtime_ns=mtime_ns,
)
self.registry.record(change)
self.logger.info("Recorded %s: %s", event_type, target_path)
@@ -84,9 +94,23 @@ class WatchdogRuntime:
while not self.stop_event.wait(60):
pass
def _prime_snapshots(self, root: Path) -> None:
self.logger.info("Priming file snapshots: %s", root)
count = 0
for path in root.rglob("*"):
if self.stop_event.is_set():
break
if not self.filter.should_track(path, is_directory=path.is_dir()):
continue
size, mtime_ns = _safe_snapshot(path)
self.registry.update_snapshot(str(path), size, mtime_ns)
count += 1
self.logger.info("Primed %s file snapshot(s): %s", count, root)
def _safe_size(path: Path) -> int | None:
def _safe_snapshot(path: Path) -> tuple[int | None, int | None]:
try:
return path.stat().st_size
stat = path.stat()
return stat.st_size, stat.st_mtime_ns
except OSError:
return None
return None, None