Vasanko/cista-storage
Vasanko/cista-storage
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

More robust updates

Browse Source
This commit is contained in:
Leo Vasanko 2025-08-13 11:21:29 -07:00
parent af4e90357f
commit 9cc210140e
1 changed files with 75 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

118
cista/watching.py
View File

@ -91,22 +91,7 @@ def treeinspos(rootmod: list[FileEntry], relpath: PurePosixPath, relfile: int):
isfile = 0 isfile = 0
level = 0 level = 0
i = 0 i = 0
iteration_count = 0
for i, rel, entry in treeiter(rootmod): for i, rel, entry in treeiter(rootmod):
iteration_count += 1
# Detect potential infinite loops in treeinspos
if iteration_count % 1000 == 0:
logger.debug(
f"DEBUG: treeinspos iteration {iteration_count}, i={i}, rel={rel}, entry.name={entry.name}, level={level}, entry.level={entry.level}"
)
if iteration_count > 10000: # Emergency brake for infinite loops
logger.error(
f"ERROR: treeinspos potential infinite loop! iteration={iteration_count}, relpath={relpath}, i={i}, level={level}"
)
break
if entry.level > level: if entry.level > level:
# We haven't found item at level, skip subdirectories # We haven't found item at level, skip subdirectories
@ -152,7 +137,7 @@ def treeinspos(rootmod: list[FileEntry], relpath: PurePosixPath, relfile: int):
logger.debug(f"DEBUG: treeinspos RETURN: cmp > 0, returning i={i}") logger.debug(f"DEBUG: treeinspos RETURN: cmp > 0, returning i={i}")
return i return i
if cmp < 0: if cmp < 0:
logger.debug(f"DEBUG: treeinspos CONTINUE: cmp < 0") logger.debug("DEBUG: treeinspos CONTINUE: cmp < 0")
continue continue
logger.debug(f"DEBUG: treeinspos INCREMENT_LEVEL: level {level} -> {level + 1}") logger.debug(f"DEBUG: treeinspos INCREMENT_LEVEL: level {level} -> {level + 1}")
@ -166,9 +151,7 @@ def treeinspos(rootmod: list[FileEntry], relpath: PurePosixPath, relfile: int):
logger.debug(f"DEBUG: treeinspos FOR_ELSE: incrementing i from {i} to {i + 1}") logger.debug(f"DEBUG: treeinspos FOR_ELSE: incrementing i from {i} to {i + 1}")
i += 1 i += 1
logger.debug( logger.debug(f"DEBUG: treeinspos EXIT: returning i={i}")
f"DEBUG: treeinspos EXIT: returning i={i}, iterations={iteration_count}"
)
return i return i
@ -336,6 +319,10 @@ def format_update(old, new):
update = [] update = []
keep_count = 0 keep_count = 0
iteration_count = 0 iteration_count = 0
# Precompute index maps to allow deterministic tie-breaking when both
# candidates exist in both sequences but are not equal (rename/move cases)
old_pos = {e: i for i, e in enumerate(old)}
new_pos = {e: i for i, e in enumerate(new)}
while oidx < len(old) and nidx < len(new): while oidx < len(old) and nidx < len(new):
iteration_count += 1 iteration_count += 1
@ -411,18 +398,38 @@ def format_update(old, new):
update.append(UpdIns(insert_items)) update.append(UpdIns(insert_items))
if not modified: if not modified:
logger.error( # Tie-break: both items exist in both lists but don't match here.
f"ERROR: format_update INFINITE_LOOP: nidx={nidx}, oidx={oidx}, old_len={len(old)}, new_len={len(new)}" # Decide whether to delete old[oidx] first or insert new[nidx] first
) # based on which alignment is closer.
logger.error( if oidx >= len(old) or nidx >= len(new):
f"ERROR: old[oidx]={old[oidx].name if oidx < len(old) else 'OUT_OF_BOUNDS'}" break
) cur_old = old[oidx]
logger.error( cur_new = new[nidx]
f"ERROR: new[nidx]={new[nidx].name if nidx < len(new) else 'OUT_OF_BOUNDS'}"
) pos_old_in_new = new_pos.get(cur_old)
raise Exception( pos_new_in_old = old_pos.get(cur_new)
f"Infinite loop in diff {nidx=} {oidx=} {len(old)=} {len(new)=}"
) # Default distances if not present (shouldn't happen if in remain sets)
dist_del = (pos_old_in_new - nidx) if pos_old_in_new is not None else 1
dist_ins = (pos_new_in_old - oidx) if pos_new_in_old is not None else 1
# Prefer the operation with smaller forward distance; tie => delete
if dist_del <= dist_ins:
# Delete current old item
oremain.discard(cur_old)
update.append(UpdDel(1))
oidx += 1
logger.debug(
f"DEBUG: format_update TIEBREAK_DEL: oidx->{oidx}, cur_old={cur_old.name}"
)
else:
# Insert current new item
nremain.discard(cur_new)
update.append(UpdIns([cur_new]))
nidx += 1
logger.debug(
f"DEBUG: format_update TIEBREAK_INS: nidx->{nidx}, cur_new={cur_new.name}"
)
# Diff any remaining # Diff any remaining
if keep_count > 0: if keep_count > 0:
@ -547,18 +554,43 @@ def watcher_inotify(loop):
) )
t0 = time.perf_counter() t0 = time.perf_counter()
logger.debug("DEBUG: inotify CALLING format_update") logger.debug("DEBUG: inotify CALLING format_update")
update = format_update(state.root, rootmod) try:
logger.debug("DEBUG: inotify format_update COMPLETED") update = format_update(state.root, rootmod)
t1 = time.perf_counter() logger.debug("DEBUG: inotify format_update COMPLETED")
with state.lock: t1 = time.perf_counter()
logger.debug("DEBUG: inotify BROADCASTING update") with state.lock:
broadcast(update, loop) logger.debug("DEBUG: inotify BROADCASTING update")
state.root = rootmod broadcast(update, loop)
logger.debug("DEBUG: inotify BROADCAST completed, state updated") state.root = rootmod
t2 = time.perf_counter() logger.debug("DEBUG: inotify BROADCAST completed, state updated")
logger.debug( t2 = time.perf_counter()
f"Format update took {t1 - t0:.1f}s, broadcast {t2 - t1:.1f}s" logger.debug(
) f"Format update took {t1 - t0:.1f}s, broadcast {t2 - t1:.1f}s"
)
except Exception:
logger.exception(
"format_update failed; falling back to full rescan"
)
# Fallback: full rescan and try diff again; last resort send full root
try:
fresh = walk(PurePosixPath())
try:
update = format_update(state.root, fresh)
with state.lock:
broadcast(update, loop)
state.root = fresh
logger.debug("Fallback diff succeeded after full rescan")
except Exception:
logger.exception(
"Fallback diff failed; sending full root snapshot"
)
with state.lock:
broadcast(format_root(fresh), loop)
state.root = fresh
except Exception:
logger.exception(
"Full rescan failed; dropping this batch of updates"
)
del i # Free the inotify object del i # Free the inotify object