daemon: decouple watchdog event capture from processing via a queue
Fixes silent rename-event loss on large folder trees. Previously on_created/on_moved/on_deleted handlers ran synchronously inside watchdog's own callback thread (git commits, network calls, sleeps), starving the inotify reader under bursty activity. When the reader falls behind, watchdog can't pair moved_from+moved_to within its matching window and drops the rename entirely instead of firing a degraded delete+create -- so large renames vanished with no log line and no ntfy at all. Fix: watchdog handlers now just enqueue (kind, path) tuples onto a queue.Queue and return immediately; a single dedicated worker thread drains the queue and does all the slow work. Reader thread never blocks regardless of downstream latency. Also adds file-level journal frontmatter automation: new dated entries under R.16.11 (YYYY-MM-DD.md) get auto-injected if missing (matches existing corpus format exactly -- full month name, no zero-padded day). Idempotent: files that already have a title are left untouched; files with other frontmatter get title prepended. Verified: reran the exact 300-subfolder rename repro that silently failed before the fix -- now completes as a single clean MIRRORED rename with no orphaned /data state.
This commit is contained in:
+153
-51
@@ -43,6 +43,7 @@ import shutil
|
|||||||
import subprocess
|
import subprocess
|
||||||
import datetime
|
import datetime
|
||||||
import threading
|
import threading
|
||||||
|
import queue
|
||||||
|
|
||||||
_SELF_DIR = os.path.dirname(os.path.realpath(__file__))
|
_SELF_DIR = os.path.dirname(os.path.realpath(__file__))
|
||||||
sys.path.insert(0, _SELF_DIR)
|
sys.path.insert(0, _SELF_DIR)
|
||||||
@@ -67,7 +68,7 @@ def ntfy_post(title, message, tags="package", priority="default"):
|
|||||||
log(f"ntfy_post: could not read token: {e}")
|
log(f"ntfy_post: could not read token: {e}")
|
||||||
return
|
return
|
||||||
try:
|
try:
|
||||||
subprocess.run(
|
result = subprocess.run(
|
||||||
["curl", "-fsS", "--max-time", "5", "-X", "POST",
|
["curl", "-fsS", "--max-time", "5", "-X", "POST",
|
||||||
"-H", f"Authorization: Bearer {token}",
|
"-H", f"Authorization: Bearer {token}",
|
||||||
"-H", f"Title: {title}",
|
"-H", f"Title: {title}",
|
||||||
@@ -75,8 +76,10 @@ def ntfy_post(title, message, tags="package", priority="default"):
|
|||||||
"-H", f"Priority: {priority}",
|
"-H", f"Priority: {priority}",
|
||||||
"-d", message,
|
"-d", message,
|
||||||
NTFY_URL],
|
NTFY_URL],
|
||||||
check=False, capture_output=True, timeout=8,
|
check=False, capture_output=True, timeout=8, text=True,
|
||||||
)
|
)
|
||||||
|
if result.returncode != 0:
|
||||||
|
log(f"ntfy_post: curl exited {result.returncode}: {result.stderr.strip()}")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
log(f"ntfy_post failed: {e}")
|
log(f"ntfy_post failed: {e}")
|
||||||
|
|
||||||
@@ -91,6 +94,18 @@ ARCHIVE_ID_RE = re.compile(r'^([RJFI])\.(\d\d)\.09 ') # e.g. "F.25.09 Archi
|
|||||||
ARCHIVE_CAT_RE = re.compile(r'^([RJFI])\.(\d\d) ') # category folder, to find/create .09 inside
|
ARCHIVE_CAT_RE = re.compile(r'^([RJFI])\.(\d\d) ') # category folder, to find/create .09 inside
|
||||||
INBOX_ID_RE = re.compile(r'^([RJFI])\.(\d\d)\.01 ')
|
INBOX_ID_RE = re.compile(r'^([RJFI])\.(\d\d)\.01 ')
|
||||||
|
|
||||||
|
# Journal entries: /vault/R Ryan/.../R.16.11 My journal & notes by day/YYYY/MM/YYYY-MM-DD.md
|
||||||
|
# gets an auto-injected `title: MMM D, YYYY` frontmatter property (matches the
|
||||||
|
# format Ryan has used by hand on every entry so far -- full month name, NO
|
||||||
|
# zero-padded day, e.g. "title: September 3, 2026" not "September 03, 2026").
|
||||||
|
JOURNAL_DIR = os.path.join(
|
||||||
|
VAULT, "R Ryan", "R.1 Life admin", "R.16 Journal & personal writing",
|
||||||
|
"R.16.11 My journal & notes by day",
|
||||||
|
)
|
||||||
|
JOURNAL_FILENAME_RE = re.compile(r'^(\d{4})-(\d{2})-(\d{2})\.md$')
|
||||||
|
_MONTH_NAMES = ["January", "February", "March", "April", "May", "June", "July",
|
||||||
|
"August", "September", "October", "November", "December"]
|
||||||
|
|
||||||
_lock = threading.Lock()
|
_lock = threading.Lock()
|
||||||
_recent_events = {} # (kind, path) -> timestamp, for debouncing duplicate inotify events
|
_recent_events = {} # (kind, path) -> timestamp, for debouncing duplicate inotify events
|
||||||
_DEBOUNCE_SECONDS = 3.0
|
_DEBOUNCE_SECONDS = 3.0
|
||||||
@@ -233,6 +248,67 @@ def handle_vault_moved(src_path, dest_path):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def handle_journal_file_created(path):
|
||||||
|
"""
|
||||||
|
A new dated journal entry (YYYY-MM-DD.md) appeared under R.16.11 --
|
||||||
|
auto-inject `title: MMM D, YYYY` frontmatter (no zero-padded day),
|
||||||
|
matching the format Ryan has applied by hand to every entry so far.
|
||||||
|
Idempotent: no-op if the file already has a `title:` line in its
|
||||||
|
frontmatter block, or isn't actually under the journal dir / doesn't
|
||||||
|
match the dated-entry filename shape.
|
||||||
|
"""
|
||||||
|
if not os.path.isfile(path):
|
||||||
|
return
|
||||||
|
if not path.endswith(".md"):
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
rel_to_journal = os.path.relpath(path, JOURNAL_DIR)
|
||||||
|
except ValueError:
|
||||||
|
return
|
||||||
|
if rel_to_journal.startswith(".."):
|
||||||
|
return # not actually under the journal dir
|
||||||
|
base = os.path.basename(path)
|
||||||
|
m = JOURNAL_FILENAME_RE.match(base)
|
||||||
|
if not m:
|
||||||
|
return # not a dated entry (e.g. some other note dropped in here)
|
||||||
|
if not _debounced("journal_title", path):
|
||||||
|
return
|
||||||
|
year, month, day = int(m.group(1)), int(m.group(2)), int(m.group(3))
|
||||||
|
month_name = _MONTH_NAMES[month - 1]
|
||||||
|
title_value = f"{month_name} {day}, {year}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
with open(path, encoding="utf-8") as f:
|
||||||
|
text = f.read()
|
||||||
|
except Exception as e:
|
||||||
|
log(f"ERROR journal title read failed: {path}: {e}")
|
||||||
|
return
|
||||||
|
|
||||||
|
if text.startswith("---\n"):
|
||||||
|
end = text.find("\n---", 4)
|
||||||
|
if end == -1:
|
||||||
|
log(f"SKIP journal title (malformed frontmatter, no closing ---): {rel_to_journal}")
|
||||||
|
return
|
||||||
|
frontmatter = text[4:end]
|
||||||
|
rest = text[end:] # starts with "\n---"
|
||||||
|
if re.search(r'^title:\s*', frontmatter, re.MULTILINE):
|
||||||
|
return # already has a title, never overwrite a human-set value
|
||||||
|
new_frontmatter = f"title: {title_value}\n" + frontmatter
|
||||||
|
new_text = "---\n" + new_frontmatter + rest
|
||||||
|
else:
|
||||||
|
# no frontmatter block at all yet -- create one
|
||||||
|
new_text = f"---\ntitle: {title_value}\n---\n" + text
|
||||||
|
|
||||||
|
try:
|
||||||
|
with open(path, "w", encoding="utf-8") as f:
|
||||||
|
f.write(new_text)
|
||||||
|
except Exception as e:
|
||||||
|
log(f"ERROR journal title write failed: {path}: {e}")
|
||||||
|
return
|
||||||
|
log(f"TITLED journal entry: {rel_to_journal} -> \"{title_value}\"")
|
||||||
|
_git_commit(f"vault_sync_daemon: add title to journal entry {base}", [path])
|
||||||
|
|
||||||
|
|
||||||
def _category_title(catname):
|
def _category_title(catname):
|
||||||
"""'F.22 Fun lists' -> 'Fun lists'"""
|
"""'F.22 Fun lists' -> 'Fun lists'"""
|
||||||
m = CAT_RE.match(catname)
|
m = CAT_RE.match(catname)
|
||||||
@@ -430,77 +506,103 @@ def handle_data_created(path):
|
|||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# watchdog wiring
|
# watchdog wiring
|
||||||
|
#
|
||||||
|
# CRITICAL: watchdog's internal reader thread must stay fast. On Linux,
|
||||||
|
# watchdog polls the raw inotify fd and is responsible for pairing a
|
||||||
|
# rename's IN_MOVED_FROM + IN_MOVED_TO events into a single `on_moved`
|
||||||
|
# call (it does this by matching cookies within its own read loop). If our
|
||||||
|
# FileSystemEventHandler callbacks run slow, synchronous work directly on
|
||||||
|
# that thread (git commits, network ntfy calls, time.sleep debounce
|
||||||
|
# delays), a burst of inotify events (e.g. renaming a folder with many
|
||||||
|
# children -> hundreds of individual FS events) can back up the kernel's
|
||||||
|
# inotify event queue faster than watchdog's reader drains it. Once
|
||||||
|
# events start dropping, the MOVED_FROM/MOVED_TO cookie pairing breaks and
|
||||||
|
# watchdog falls back to firing a bare `on_deleted` + a later bare
|
||||||
|
# `on_created` instead of one `on_moved` -- exactly the delete+create
|
||||||
|
# corruption bug Ryan reported (confirmed by reproduction: renaming a
|
||||||
|
# ~300-child folder produced a rename that silently vanished from the log
|
||||||
|
# entirely, while the same rename on a small/empty folder worked fine).
|
||||||
|
#
|
||||||
|
# Fix: handlers do ZERO real work. They just stuff the event onto a
|
||||||
|
# queue and return immediately, so the watchdog reader thread is never
|
||||||
|
# blocked and can drain the kernel's rename event pairs promptly. A
|
||||||
|
# single dedicated worker thread pulls off the queue and does the actual
|
||||||
|
# (slow) sync work, serialized, in FS-event order.
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
_event_queue = queue.Queue()
|
||||||
|
|
||||||
|
|
||||||
class VaultHandler(FileSystemEventHandler):
|
class VaultHandler(FileSystemEventHandler):
|
||||||
def on_created(self, event):
|
def on_created(self, event):
|
||||||
if not event.is_directory:
|
if event.is_directory:
|
||||||
return
|
_event_queue.put(("vault_created", event.src_path))
|
||||||
with _lock:
|
else:
|
||||||
try:
|
_event_queue.put(("vault_file_created", event.src_path))
|
||||||
time.sleep(0.5) # let Obsidian finish writing the folder-note
|
|
||||||
handle_vault_created(event.src_path)
|
|
||||||
except Exception as e:
|
|
||||||
log(f"ERROR handle_vault_created: {e}")
|
|
||||||
ntfy_post(
|
|
||||||
"Vault sync ERROR",
|
|
||||||
f"handle_vault_created failed:\n{event.src_path}\n{e}",
|
|
||||||
tags="rotating_light",
|
|
||||||
priority="urgent",
|
|
||||||
)
|
|
||||||
|
|
||||||
def on_moved(self, event):
|
def on_moved(self, event):
|
||||||
if not event.is_directory:
|
if not event.is_directory:
|
||||||
return
|
return
|
||||||
with _lock:
|
_event_queue.put(("vault_moved", event.src_path, event.dest_path))
|
||||||
try:
|
|
||||||
handle_vault_moved(event.src_path, event.dest_path)
|
|
||||||
except Exception as e:
|
|
||||||
log(f"ERROR handle_vault_moved: {e}")
|
|
||||||
ntfy_post(
|
|
||||||
"Vault sync ERROR",
|
|
||||||
f"handle_vault_moved failed:\n{event.src_path} -> {event.dest_path}\n{e}",
|
|
||||||
tags="rotating_light",
|
|
||||||
priority="urgent",
|
|
||||||
)
|
|
||||||
|
|
||||||
def on_deleted(self, event):
|
def on_deleted(self, event):
|
||||||
if not event.is_directory:
|
if not event.is_directory:
|
||||||
return
|
return
|
||||||
with _lock:
|
_event_queue.put(("vault_deleted", event.src_path))
|
||||||
try:
|
|
||||||
handle_vault_deleted(event.src_path)
|
|
||||||
except Exception as e:
|
|
||||||
log(f"ERROR handle_vault_deleted: {e}")
|
|
||||||
ntfy_post(
|
|
||||||
"Vault sync ERROR",
|
|
||||||
f"handle_vault_deleted failed:\n{event.src_path}\n{e}",
|
|
||||||
tags="rotating_light",
|
|
||||||
priority="urgent",
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class DataHandler(FileSystemEventHandler):
|
class DataHandler(FileSystemEventHandler):
|
||||||
def on_created(self, event):
|
def on_created(self, event):
|
||||||
if not event.is_directory:
|
if not event.is_directory:
|
||||||
return
|
return
|
||||||
with _lock:
|
_event_queue.put(("data_created", event.src_path))
|
||||||
try:
|
|
||||||
# small settle delay so a batch mkdir/copy isn't caught half-written
|
|
||||||
time.sleep(0.5)
|
def _worker_loop():
|
||||||
handle_data_created(event.src_path)
|
"""Single consumer thread: drains _event_queue and does the actual
|
||||||
except Exception as e:
|
(slow) sync work, one event at a time, in the order events arrived.
|
||||||
log(f"ERROR handle_data_created: {e}")
|
Keeping this off the watchdog reader thread is what lets inotify's
|
||||||
ntfy_post(
|
MOVED_FROM/MOVED_TO cookie pairing succeed under load (see module
|
||||||
"Vault sync ERROR",
|
docstring above the queue wiring)."""
|
||||||
f"handle_data_created failed:\n{event.src_path}\n{e}",
|
while True:
|
||||||
tags="rotating_light",
|
item = _event_queue.get()
|
||||||
priority="urgent",
|
kind = item[0]
|
||||||
)
|
try:
|
||||||
|
with _lock:
|
||||||
|
if kind == "vault_created":
|
||||||
|
_, path = item
|
||||||
|
time.sleep(0.5) # let Obsidian finish writing the folder-note
|
||||||
|
handle_vault_created(path)
|
||||||
|
elif kind == "vault_file_created":
|
||||||
|
_, path = item
|
||||||
|
time.sleep(0.5) # let the writer (Obsidian/editor) finish writing content
|
||||||
|
handle_journal_file_created(path)
|
||||||
|
elif kind == "vault_moved":
|
||||||
|
_, src_path, dest_path = item
|
||||||
|
handle_vault_moved(src_path, dest_path)
|
||||||
|
elif kind == "vault_deleted":
|
||||||
|
_, path = item
|
||||||
|
handle_vault_deleted(path)
|
||||||
|
elif kind == "data_created":
|
||||||
|
_, path = item
|
||||||
|
time.sleep(0.5) # small settle delay so a batch mkdir/copy isn't caught half-written
|
||||||
|
handle_data_created(path)
|
||||||
|
except Exception as e:
|
||||||
|
log(f"ERROR handling {item}: {e}")
|
||||||
|
ntfy_post(
|
||||||
|
"Vault sync ERROR",
|
||||||
|
f"Handling {kind} failed:\n{item}\n{e}",
|
||||||
|
tags="rotating_light",
|
||||||
|
priority="urgent",
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
_event_queue.task_done()
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
log("vault_sync_daemon starting")
|
log("vault_sync_daemon starting")
|
||||||
|
worker = threading.Thread(target=_worker_loop, daemon=True)
|
||||||
|
worker.start()
|
||||||
observer = Observer()
|
observer = Observer()
|
||||||
observer.schedule(VaultHandler(), VAULT, recursive=True)
|
observer.schedule(VaultHandler(), VAULT, recursive=True)
|
||||||
observer.schedule(DataHandler(), DATA, recursive=True)
|
observer.schedule(DataHandler(), DATA, recursive=True)
|
||||||
|
|||||||
Reference in New Issue
Block a user