Skip to content

fix(sync): include needUpload/needSyncMtime in initial pull-ack decision #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
lyston11 wants to merge 1 commit into from
Closed

fix(sync): include needUpload/needSyncMtime in initial pull-ack decision #22

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,3 +13,8 @@ build/
# Local integration test artifacts — not committed
test_sync_local.py
test_sync_report.log

# local test artifacts (Fast Note Sync investigation)
.venv/
config-test.yaml
vault/
20 changes: 18 additions & 2 deletions fns_cli/file_sync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,8 @@ def __init__(self, engine: SyncEngine) -> None:
self._pending_download_paths: set[str] = set()
self._expected_modify = 0
self._expected_delete = 0
self._expected_upload = 0
self._expected_mtime = 0
self._received_modify = 0
self._received_delete = 0
self._got_end = False
Expand Down Expand Up @@ -117,6 +119,8 @@ def _reset_counters(self) -> None:
self._pending_download_paths.clear()
self._expected_modify = 0
self._expected_delete = 0
self._expected_upload = 0
self._expected_mtime = 0
self._received_modify = 0
self._received_delete = 0
self._pending_last_time = 0
Expand Down Expand Up @@ -390,6 +394,10 @@ async def _on_sync_mtime(self, msg: WSMessage) -> None:
os.utime(full, (ts, ts))
except OSError:
pass
# FileSyncMtime is counted in needSyncMtimeCount; credit it so
# _check_complete can finish when only mtime updates were requested.
self._received_modify += 1
self._check_complete()

async def _on_chunk_download_start(self, msg: WSMessage) -> None:
data = _extract_inner(msg.data)
Expand Down Expand Up @@ -505,6 +513,14 @@ async def _on_sync_end(self, msg: WSMessage) -> None:
self._pending_last_time = last_time
self._expected_modify = int(data.get("needModifyCount") or 0)
self._expected_delete = int(data.get("needDeleteCount") or 0)
# FileUpload detail frames (server requests client to upload) and
# FileSyncMtime detail frames are delivered through the same paged
# download channel as FileSyncUpdate/FileSyncDelete. The initial pull
# ack (pageIndex=-1) must be sent whenever ANY of these is non-zero,
# otherwise the server never sends the page and uploads/mtime updates
# are silently dropped (same bug class as NoteSync).
self._expected_upload = int(data.get("needUploadCount") or 0)
self._expected_mtime = int(data.get("needSyncMtimeCount") or 0)
need_upload = data.get("needUploadCount", 0)

self._got_end = True
Expand All @@ -513,7 +529,7 @@ async def _on_sync_end(self, msg: WSMessage) -> None:
last_time, self._expected_modify, self._expected_delete, need_upload,
)

total_expected = self._expected_modify + self._expected_delete
total_expected = self._expected_modify + self._expected_delete + self._expected_upload + self._expected_mtime
if total_expected > 0:
await self._send_page_ack(self._sync_context, -1, self._sync_vault)

Expand All @@ -522,7 +538,7 @@ async def _on_sync_end(self, msg: WSMessage) -> None:
def _check_complete(self) -> None:
if not self._got_end:
return
total_expected = self._expected_modify + self._expected_delete
total_expected = self._expected_modify + self._expected_delete + self._expected_upload + self._expected_mtime
total_received = self._received_modify + self._received_delete
if total_received >= total_expected and not self._pending_download_paths and not self._download_sessions:
log.info(
Expand Down
29 changes: 27 additions & 2 deletions fns_cli/note_sync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,8 @@ def __init__(self, engine: SyncEngine) -> None:
self._sync_complete = False
self._expected_modify = 0
self._expected_delete = 0
self._expected_upload = 0
self._expected_mtime = 0
self._received_modify = 0
self._received_delete = 0
self._got_end = False
Expand Down Expand Up @@ -257,6 +259,10 @@ async def _on_sync_rename(self, msg: WSMessage) -> None:
self._try_remove_empty_parent(old_full)
except Exception:
log.exception("Failed to rename %s → %s", old_path, new_path)
# NoteSyncRename is delivered through the paged download channel; credit
# it so _check_all_received can finish when only renames were requested.
self._received_modify += 1
self._check_all_received()

async def _on_sync_mtime(self, msg: WSMessage) -> None:
data = _extract_inner(msg.data)
Expand All @@ -271,6 +277,10 @@ async def _on_sync_mtime(self, msg: WSMessage) -> None:
os.utime(full, (ts, ts))
except OSError:
pass
# NoteSyncMtime is counted in needSyncMtimeCount; credit it so
# _check_all_received can finish when only mtime updates were requested.
self._received_modify += 1
self._check_all_received()

async def _on_sync_need_push(self, msg: WSMessage) -> None:
data = _extract_inner(msg.data)
Expand All @@ -281,6 +291,10 @@ async def _on_sync_need_push(self, msg: WSMessage) -> None:
# NeedPush is an explicit server request to re-send the local content;
# it must bypass the normal echo suppression check.
await self.push_modify(rel_path, force=True)
# NoteSyncNeedPush is counted in needUploadCount; credit it so
# _check_all_received can finish when only uploads were requested.
self._received_modify += 1
self._check_all_received()

async def _on_sync_end(self, msg: WSMessage) -> None:
data = _extract_inner(msg.data)
Expand All @@ -305,7 +319,16 @@ async def _on_sync_end(self, msg: WSMessage) -> None:
data.get("needUploadCount", 0),
)

total_expected = self._expected_modify + self._expected_delete
# The server batches download items (NoteSyncNeedPush / NoteSyncModify /
# NoteSyncMtime / NoteSyncDelete) into paged frames. It only sends a page
# after the client sends the initial pull ack (pageIndex=-1). needUpload,
# needModify, needSyncMtime and needDelete all describe items the server
# will deliver through this paged channel, so the initial ack must be sent
# whenever ANY of them is non-zero — otherwise the page is never sent and
# uploads/remote modifications are silently dropped.
self._expected_upload = int(data.get("needUploadCount") or 0)
self._expected_mtime = int(data.get("needSyncMtimeCount") or 0)
total_expected = self._expected_modify + self._expected_delete + self._expected_upload + self._expected_mtime
if total_expected == 0:
self._sync_complete = True
self._commit_last_time()
Expand Down Expand Up @@ -344,6 +367,8 @@ def _reset_counters(self) -> None:
self._got_end = False
self._expected_modify = 0
self._expected_delete = 0
self._expected_upload = 0
self._expected_mtime = 0
self._received_modify = 0
self._received_delete = 0
self._pending_last_time = 0
Expand All @@ -353,7 +378,7 @@ def _reset_counters(self) -> None:
def _check_all_received(self) -> None:
if not self._got_end:
return
total_expected = self._expected_modify + self._expected_delete
total_expected = self._expected_modify + self._expected_delete + self._expected_upload + self._expected_mtime
total_received = self._received_modify + self._received_delete
if total_received >= total_expected:
log.info(
Expand Down
26 changes: 24 additions & 2 deletions fns_cli/setting_sync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,8 @@ def __init__(self, engine: SyncEngine) -> None:
self._sync_complete = False
self._expected_modify = 0
self._expected_delete = 0
self._expected_upload = 0
self._expected_mtime = 0
self._received_modify = 0
self._received_delete = 0
self._got_end = False
Expand Down Expand Up @@ -252,6 +254,10 @@ async def _on_sync_mtime(self, msg: WSMessage) -> None:
os.utime(full, (ts, ts))
except OSError:
pass
# SettingSyncMtime is counted in needSyncMtimeCount; credit it so
# _check_all_received can finish when only mtime updates were requested.
self._received_modify += 1
self._check_all_received()

async def _on_sync_need_upload(self, msg: WSMessage) -> None:
data = _extract_inner(msg.data)
Expand All @@ -263,6 +269,12 @@ async def _on_sync_need_upload(self, msg: WSMessage) -> None:
rel_path = item.get("path", "") if isinstance(item, dict) else str(item)
if rel_path:
await self.push_modify(rel_path)
# The server reports these items via needUploadCount and delivers them
# through the paged download channel. Credit them to the received count
# so _check_all_received can complete when only uploads were requested;
# without this the initial pull ack is never triggered either.
self._received_modify += len(need_upload)
self._check_all_received()

async def _on_sync_end(self, msg: WSMessage) -> None:
data = _extract_inner(msg.data)
Expand All @@ -286,7 +298,15 @@ async def _on_sync_end(self, msg: WSMessage) -> None:
data.get("needUploadCount", 0),
)

total_expected = self._expected_modify + self._expected_delete
# Settings detail frames (SettingSyncModify / SettingSyncDelete /
# SettingSyncMtime / SettingSyncNeedUpload) are delivered through the
# same paged download channel. The initial pull ack (pageIndex=-1)
# must be sent whenever ANY of these is non-zero, otherwise the server
# never sends the page and uploads/mtime updates are silently dropped
# (same bug class as NoteSync).
self._expected_upload = int(data.get("needUploadCount") or 0)
self._expected_mtime = int(data.get("needSyncMtimeCount") or 0)
total_expected = self._expected_modify + self._expected_delete + self._expected_upload + self._expected_mtime
if total_expected == 0:
self._sync_complete = True
self._commit_last_time()
Expand Down Expand Up @@ -325,6 +345,8 @@ def _reset_counters(self) -> None:
self._got_end = False
self._expected_modify = 0
self._expected_delete = 0
self._expected_upload = 0
self._expected_mtime = 0
self._received_modify = 0
self._received_delete = 0
self._pending_last_time = 0
Expand All @@ -334,7 +356,7 @@ def _reset_counters(self) -> None:
def _check_all_received(self) -> None:
if not self._got_end:
return
total_expected = self._expected_modify + self._expected_delete
total_expected = self._expected_modify + self._expected_delete + self._expected_upload + self._expected_mtime
total_received = self._received_modify + self._received_delete
if total_received >= total_expected:
log.info(
Expand Down