From ea48de290f637a23cd4a67bbf2a5b068ef95d20b Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Tue, 14 Jul 2026 16:32:47 +0800 Subject: [PATCH] fix(sync): track cleanup state to stop re-deleting log rows and bound SyncQueue - access_reader.delete_log_ids returns the actual rows deleted (was None). - sql_writer.mark_cleaned flips applied queue rows to 'cleaned' (sets CleanedAt) after their Access log rows are physically removed, so the same IDs are never deleted twice. - sql_writer.purge_cleaned removes 'cleaned' rows older than a retention window (default 24h) so SyncQueue stops growing without bound. - cleanup.cleanup_file marks rows cleaned after a successful delete and returns the real delete count, so the service log reports honest 'cleaned N' instead of a constant. - service.cycle calls purge_cleaned once per pass; config adds cleaned_retention_hours (default 24). - sql/01_sync_queue.sql adds CleanedAt column + IX_SyncQueue_Cleaned idempotently. - tests: unit coverage for mark_cleaned/purge_cleaned/delete_log_ids return count; assert cycle purges each pass. --- sql/01_sync_queue.sql | 20 ++++++++++ src/sync/access_reader.py | 15 ++++--- src/sync/cleanup.py | 17 ++++++-- src/sync/config.py | 1 + src/sync/service.py | 10 +++++ src/sync/sql_writer.py | 42 ++++++++++++++++++++ tests/test_cleanup_accounting.py | 67 ++++++++++++++++++++++++++++++++ tests/test_service.py | 1 + 8 files changed, 164 insertions(+), 9 deletions(-) create mode 100644 tests/test_cleanup_accounting.py diff --git a/sql/01_sync_queue.sql b/sql/01_sync_queue.sql index c7f85eb..f923636 100644 --- a/sql/01_sync_queue.sql +++ b/sql/01_sync_queue.sql @@ -40,3 +40,23 @@ BEGIN ON dbo.SyncQueue(Status, TargetSchema, TargetTable); END GO + +-- Cleanup bookkeeping: track when a queue row's Access log counterpart has +-- been physically deleted, so cleanup never re-deletes the same IDs and the +-- table can be purged to bound its growth. +IF NOT EXISTS (SELECT 1 FROM sys.columns + WHERE object_id = OBJECT_ID('dbo.SyncQueue') + AND name = 'CleanedAt') +BEGIN + ALTER TABLE dbo.SyncQueue ADD CleanedAt datetime2 NULL; +END +GO + +IF NOT EXISTS (SELECT 1 FROM sys.indexes + WHERE name = 'IX_SyncQueue_Cleaned' + AND object_id = OBJECT_ID('dbo.SyncQueue')) +BEGIN + CREATE INDEX IX_SyncQueue_Cleaned + ON dbo.SyncQueue(Status, CleanedAt); +END +GO diff --git a/src/sync/access_reader.py b/src/sync/access_reader.py index 8dbe782..d880002 100644 --- a/src/sync/access_reader.py +++ b/src/sync/access_reader.py @@ -87,15 +87,18 @@ class AccessReader: def delete_log_ids( self, ids: list[int], batch_size: int, retries: int - ) -> None: + ) -> int: """Delete the given log-row IDs in chunks, retrying on lock contention. - A no-op when ``ids`` is empty (never raises). Retries with linear - backoff because the live client may briefly hold a page lock on - ``TableChangeLog``. + Returns the total number of rows actually deleted (sum of per-chunk + ``cursor.rowcount``), so the caller can report honest counts. A no-op + (returns 0) when ``ids`` is empty. Retries with linear backoff because + the live client may briefly hold a page lock on ``TableChangeLog``. + Raises on final lock failure. """ if not ids: - return + return 0 + total = 0 cur = self._connect().cursor() for i in range(0, len(ids), batch_size): chunk = ids[i:i + batch_size] @@ -106,12 +109,14 @@ class AccessReader: f"DELETE FROM TableChangeLog WHERE ID IN ({placeholders})", *chunk, ) + total += cur.rowcount break except pyodbc.OperationalError: if attempt < retries - 1: time.sleep(0.2 * (attempt + 1)) else: raise + return total def close(self): if self._conn: diff --git a/src/sync/cleanup.py b/src/sync/cleanup.py index 547a789..3448134 100644 --- a/src/sync/cleanup.py +++ b/src/sync/cleanup.py @@ -17,11 +17,20 @@ log = logging.getLogger(__name__) def cleanup_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int: """Delete applied log rows for ``fm`` from the Access log. - Returns the number of log rows deleted. A no-op (returns 0) when no rows - are currently in the ``applied`` state for this file. + Returns the number of log rows actually deleted. A no-op (returns 0) when + no rows are currently in the ``applied`` state for this file. + + After the Access-side delete succeeds (``delete_log_ids`` raises on final + lock failure, so reaching the mark step means success), the corresponding + queue rows are flipped to ``cleaned`` so they are never re-deleted and can + be purged later. The returned count reflects genuine deletions, so the + service log is honest even when there is nothing new to clean. """ ids = writer.applied_log_ids(fm.file) if not ids: return 0 - reader.delete_log_ids(ids, cfg.runtime.cleanup_batch_size, cfg.runtime.cleanup_lock_retries) - return len(ids) + deleted = reader.delete_log_ids( + ids, cfg.runtime.cleanup_batch_size, cfg.runtime.cleanup_lock_retries + ) + writer.mark_cleaned(fm.file, ids) + return deleted diff --git a/src/sync/config.py b/src/sync/config.py index bae2580..71e8562 100644 --- a/src/sync/config.py +++ b/src/sync/config.py @@ -20,6 +20,7 @@ class RuntimeConfig(BaseModel): retry_backoff_seconds: int = 30 cleanup_batch_size: int = 200 cleanup_lock_retries: int = 3 + cleaned_retention_hours: int = 24 class FileMapping(BaseModel): model_config = ConfigDict(coerce_numbers_to_str=True) diff --git a/src/sync/service.py b/src/sync/service.py index b53c16d..270996a 100644 --- a/src/sync/service.py +++ b/src/sync/service.py @@ -75,6 +75,16 @@ def cycle(cfg): log.exception("cleanup failed for %s", fm.file) finally: reader.close() + + # Once Access log rows are removed, their queue rows are marked + # ``cleaned``; purge the old ones so SyncQueue stays bounded. + try: + purged = writer.purge_cleaned(cfg.runtime.cleaned_retention_hours) + if purged: + log.info("purged %d cleaned queue rows", purged) + except Exception: + log.exception("purge failed") + finally: writer.close() diff --git a/src/sync/sql_writer.py b/src/sync/sql_writer.py index 9212d6c..755864a 100644 --- a/src/sync/sql_writer.py +++ b/src/sync/sql_writer.py @@ -90,6 +90,48 @@ class SqlWriter: ) return [r[0] for r in cur.fetchall()] + def mark_cleaned(self, source_file: str, source_log_ids: list[int]) -> None: + """Mark applied queue rows for ``source_file`` as ``cleaned``. + + Called after the corresponding Access log rows have been physically + deleted. Once a row is ``cleaned``, ``applied_log_ids`` no longer + returns it, so the same IDs are never deleted from Access twice. Rows + are matched on ``(SourceFile, SourceLogID)`` and constrained to + ``Status='applied'``, so a row that errored out is never silently + marked clean. Chunked to stay under SQL Server's 2100-param limit. + """ + if not source_log_ids: + return + cur = self._conn.cursor() + for i in range(0, len(source_log_ids), 1000): + chunk = source_log_ids[i:i + 1000] + placeholders = ",".join("?" * len(chunk)) + cur.execute( + f"UPDATE dbo.SyncQueue SET Status='cleaned', " + f"CleanedAt=sysdatetime() " + f"WHERE SourceFile=? AND Status='applied' " + f"AND SourceLogID IN ({placeholders})", + source_file, + *chunk, + ) + + def purge_cleaned(self, retention_hours: int) -> int: + """Delete ``cleaned`` rows older than ``retention_hours``. + + Keeps the table bounded: cleanup marks rows ``cleaned`` every cycle, + and this removes the old ones after a short audit/debug window so + ``dbo.SyncQueue`` stops growing without bound. Returns the number of + rows removed. + """ + cur = self._conn.cursor() + cur.execute( + "DELETE FROM dbo.SyncQueue " + "WHERE Status='cleaned' " + "AND CleanedAt < DATEADD(hour, -?, GETDATE())", + retention_hours, + ) + return cur.rowcount + def close(self) -> None: """Close the underlying pyodbc connection.""" self._conn.close() diff --git a/tests/test_cleanup_accounting.py b/tests/test_cleanup_accounting.py new file mode 100644 index 0000000..652f409 --- /dev/null +++ b/tests/test_cleanup_accounting.py @@ -0,0 +1,67 @@ +"""Unit tests for the cleanup bookkeeping added to SqlWriter / AccessReader. + +These mock pyodbc so no database is required. They pin the contract that +``mark_cleaned`` / ``purge_cleaned`` issue the right statements, and that +``delete_log_ids`` returns the actual number of rows deleted (not ``None``). +""" +from unittest.mock import MagicMock, patch + +from sync.sql_writer import SqlWriter +from sync.access_reader import AccessReader + + +def _writer(): + """Build a SqlWriter whose pyodbc connection is a MagicMock.""" + with patch("sync.sql_writer.pyodbc.connect") as mconnect: + conn = MagicMock() + mconnect.return_value = conn + w = SqlWriter("conn_str", "dbo.SyncQueue") + return w, conn + + +def test_mark_cleaned_updates_only_applied_rows_for_file(): + w, conn = _writer() + cur = conn.cursor.return_value + w.mark_cleaned("a.accdb", [1, 2, 3]) + assert cur.execute.call_count == 1 + sql = cur.execute.call_args.args[0] + assert "UPDATE dbo.SyncQueue SET Status='cleaned'" in sql + assert "CleanedAt=sysdatetime()" in sql + assert "SourceFile=?" in sql + assert "Status='applied'" in sql + # first param is the file, the rest are the SourceLogIDs + assert cur.execute.call_args.args[1] == "a.accdb" + assert list(cur.execute.call_args.args[2:]) == [1, 2, 3] + + +def test_mark_cleaned_no_ids_is_noop(): + w, conn = _writer() + cur = conn.cursor.return_value + w.mark_cleaned("a.accdb", []) + cur.execute.assert_not_called() + + +def test_purge_cleaned_deletes_old_cleaned_rows(): + w, conn = _writer() + cur = conn.cursor.return_value + cur.rowcount = 42 + n = w.purge_cleaned(24) + assert n == 42 + sql = cur.execute.call_args.args[0] + assert "DELETE FROM dbo.SyncQueue" in sql + assert "Status='cleaned'" in sql + assert "CleanedAt < DATEADD(hour, -?, GETDATE())" in sql + assert cur.execute.call_args.args[1] == 24 + + +def test_delete_log_ids_returns_rowcount_and_chunks(): + r = AccessReader("x.accdb", "drv") + fake_conn = MagicMock() + fake_cur = MagicMock() + fake_cur.rowcount = 7 # per-chunk affected rows + fake_conn.cursor.return_value = fake_cur + with patch.object(r, "_connect", return_value=fake_conn): + # 5 ids, batch size 2 -> 3 chunks, 3 DELETEs, total 21 rows + n = r.delete_log_ids([1, 2, 3, 4, 5], 2, 3) + assert n == 21 + assert fake_cur.execute.call_count == 3 diff --git a/tests/test_service.py b/tests/test_service.py index bb5fdc2..1168b52 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -47,6 +47,7 @@ def test_cycle_continues_on_capture_error_and_still_applies_and_cleans(): assert mcapture.call_count == 2 # both files attempted (bad swallowed) writer_inst.call_apply.assert_called_once() # apply still called assert mcleanup.call_count == 2 # cleanup attempted for both files + writer_inst.purge_cleaned.assert_called_once() # cleanup bookkeeping writer_inst.close.assert_called() # writer closed in finally