add periodic full-tree reconcile sweep as inotify backstop
Root cause of the R.12.22 mkdir going unnoticed was never pinned down -- kernel-level inotifywait on the exact same directory caught the same mkdir instantly, ruling out mount/fs issues, but watchdog's own internal bookkeeping missed it with zero log line. Rather than keep chasing a specific inotify failure mode, make the daemon resilient regardless of cause: a reconcile() sweep walks the whole /data and /vault tree every 3 minutes (plus once at startup) and calls the same idempotent handle_data_created/handle_vault_created used for live events. Both are check-target-state-first by construction, so this is a no-op on an already-synced tree and only acts on genuine misses. Does not attempt rename reconstruction -- a missed rename looks identical to a delete+create after the fact, which is genuinely ambiguous. Live on_moved handling remains the only rename path; reconcile's job is strictly to guarantee creates are never lost.
This commit is contained in:
@@ -493,6 +493,59 @@ def handle_data_created(path):
|
||||
)
|
||||
|
||||
|
||||
def handle_data_moved(src_path, dest_path):
|
||||
"""
|
||||
A folder in /data was renamed/moved (e.g. via an SMB client, GUI file
|
||||
manager, or any tool that stages-then-renames rather than mkdir'ing
|
||||
directly in place). DataHandler previously had NO on_moved at all, so
|
||||
this fired zero log line and zero ntfy -- confirmed root cause of
|
||||
Ryan's 2026-09-22 report ("created a folder in /data, nothing happened
|
||||
on ntfy or in /vault": the folder actually arrived via a rename, not a
|
||||
bare create, and there was no handler to catch it).
|
||||
|
||||
Mirror rule: if /vault already has a note at the OLD relative path,
|
||||
rename it to match (same as the forward vault->data rename mirror).
|
||||
If there's no /vault counterpart yet at all, treat this exactly like
|
||||
a fresh data_created at the NEW path (covers the same "staged then
|
||||
renamed into place" arrival pattern as a plain create).
|
||||
"""
|
||||
if _in_skip(src_path, DATA) or _in_skip(dest_path, DATA):
|
||||
return
|
||||
rel_src = os.path.relpath(src_path, DATA)
|
||||
rel_dest = os.path.relpath(dest_path, DATA)
|
||||
if _under_data_exempt(rel_src) or _under_data_exempt(rel_dest):
|
||||
return
|
||||
vault_src = os.path.join(VAULT, rel_src)
|
||||
vault_dest = os.path.join(VAULT, rel_dest)
|
||||
if not os.path.isdir(vault_src):
|
||||
# No /vault counterpart to rename -- treat the arrival at the new
|
||||
# name like a normal reverse-sync create.
|
||||
handle_data_created(dest_path)
|
||||
return
|
||||
if os.path.exists(vault_dest):
|
||||
log(f"SKIP data-move mirror (vault dest exists): {rel_src} -> {rel_dest}")
|
||||
return
|
||||
os.makedirs(os.path.dirname(vault_dest), exist_ok=True)
|
||||
os.rename(vault_src, vault_dest)
|
||||
# keep the folder-note's own filename in sync with its containing
|
||||
# folder's new name (matches the "<ID name>/<ID name>.md" convention
|
||||
# used throughout the vault)
|
||||
old_base = os.path.basename(vault_src)
|
||||
new_base = os.path.basename(vault_dest)
|
||||
old_note = os.path.join(vault_dest, old_base + ".md")
|
||||
new_note = os.path.join(vault_dest, new_base + ".md")
|
||||
if old_base != new_base and os.path.isfile(old_note) and not os.path.exists(new_note):
|
||||
os.rename(old_note, new_note)
|
||||
log(f"MIRRORED rename /vault: {rel_src} -> {rel_dest}")
|
||||
commit_paths = [vault_dest] if os.path.isdir(vault_dest) else []
|
||||
_git_commit(f"vault_sync_daemon: renamed {old_base} -> {new_base}", commit_paths)
|
||||
ntfy_post(
|
||||
"Vault sync: renamed",
|
||||
f"/vault: {rel_src} -> {rel_dest}",
|
||||
tags="package",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# watchdog wiring
|
||||
#
|
||||
@@ -546,6 +599,11 @@ class DataHandler(FileSystemEventHandler):
|
||||
return
|
||||
_event_queue.put(("data_created", event.src_path))
|
||||
|
||||
def on_moved(self, event):
|
||||
if not event.is_directory:
|
||||
return
|
||||
_event_queue.put(("data_moved", event.src_path, event.dest_path))
|
||||
|
||||
|
||||
def _worker_loop():
|
||||
"""Single consumer thread: drains _event_queue and does the actual
|
||||
@@ -576,6 +634,9 @@ def _worker_loop():
|
||||
_, path = item
|
||||
time.sleep(0.5) # small settle delay so a batch mkdir/copy isn't caught half-written
|
||||
handle_data_created(path)
|
||||
elif kind == "data_moved":
|
||||
_, src_path, dest_path = item
|
||||
handle_data_moved(src_path, dest_path)
|
||||
except Exception as e:
|
||||
log(f"ERROR handling {item}: {e}")
|
||||
ntfy_post(
|
||||
@@ -588,6 +649,78 @@ def _worker_loop():
|
||||
_event_queue.task_done()
|
||||
|
||||
|
||||
RECONCILE_INTERVAL_SECONDS = 180
|
||||
|
||||
|
||||
def reconcile():
|
||||
"""
|
||||
Full-tree self-healing sweep, independent of inotify.
|
||||
|
||||
inotify is best-effort: watch descriptors can be dropped, events can be
|
||||
coalesced or lost under load, and — per Ryan 2026-09-24 — a plain
|
||||
`mkdir` under R.12.22 produced zero log line and zero ntfy with no
|
||||
reproducible cause found (kernel-level inotifywait on the same exact
|
||||
directory caught the same mkdir instantly, so it wasn't a mount/fs
|
||||
issue — watchdog's own bookkeeping simply missed it). Rather than chase
|
||||
every possible way an event can go missing, the daemon must be
|
||||
resilient no matter what: periodically diff the whole tree in both
|
||||
directions and self-heal anything inotify failed to deliver.
|
||||
|
||||
This walks every directory in /data and /vault and calls the exact
|
||||
same idempotent handlers used for live events (handle_data_created /
|
||||
handle_vault_created). Both are "check target state first, act only if
|
||||
something's actually missing" by construction (see module docstring),
|
||||
so calling them on every already-mirrored directory is a cheap no-op —
|
||||
only genuinely missed folders get created, stubbed, and ntfy'd here.
|
||||
|
||||
Deliberately does NOT attempt rename detection: a rename that happened
|
||||
while the daemon was blind looks identical to a delete+create pair,
|
||||
which is genuinely ambiguous to reconstruct after the fact. Live
|
||||
on_moved handling remains the only path for renames; reconcile's job
|
||||
is just to guarantee creates are never permanently lost.
|
||||
"""
|
||||
t0 = time.time()
|
||||
data_count = 0
|
||||
for dirpath, dirnames, _ in os.walk(DATA):
|
||||
dirnames.sort()
|
||||
if dirpath == DATA:
|
||||
continue
|
||||
if _in_skip(dirpath, DATA):
|
||||
dirnames[:] = []
|
||||
continue
|
||||
with _lock:
|
||||
handle_data_created(dirpath)
|
||||
data_count += 1
|
||||
vault_count = 0
|
||||
for dirpath, dirnames, _ in os.walk(VAULT):
|
||||
dirnames.sort()
|
||||
if dirpath == VAULT:
|
||||
continue
|
||||
if _in_skip(dirpath, VAULT):
|
||||
dirnames[:] = []
|
||||
continue
|
||||
with _lock:
|
||||
handle_vault_created(dirpath)
|
||||
vault_count += 1
|
||||
log(f"reconcile: swept {data_count} /data dirs, {vault_count} /vault dirs "
|
||||
f"in {time.time() - t0:.1f}s")
|
||||
|
||||
|
||||
def _reconcile_loop():
|
||||
while True:
|
||||
time.sleep(RECONCILE_INTERVAL_SECONDS)
|
||||
try:
|
||||
reconcile()
|
||||
except Exception as e:
|
||||
log(f"ERROR in reconcile sweep: {e}")
|
||||
ntfy_post(
|
||||
"Vault sync ERROR",
|
||||
f"Periodic reconcile sweep failed:\n{e}",
|
||||
tags="rotating_light",
|
||||
priority="urgent",
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
log("vault_sync_daemon starting")
|
||||
worker = threading.Thread(target=_worker_loop, daemon=True)
|
||||
@@ -598,6 +731,14 @@ def main():
|
||||
observer.start()
|
||||
log(f"watching {VAULT} and {DATA}")
|
||||
ntfy_post("Vault sync daemon started", f"Watching {VAULT} and {DATA}", tags="rocket", priority="low")
|
||||
# Run one reconcile sweep at startup (catches anything that happened
|
||||
# while the daemon was down, e.g. a reboot) then keep sweeping on a
|
||||
# timer as a standing backstop under live inotify events.
|
||||
try:
|
||||
reconcile()
|
||||
except Exception as e:
|
||||
log(f"ERROR in startup reconcile sweep: {e}")
|
||||
threading.Thread(target=_reconcile_loop, daemon=True).start()
|
||||
try:
|
||||
while True:
|
||||
time.sleep(5)
|
||||
|
||||
Reference in New Issue
Block a user