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.
This commit is contained in:
@@ -40,3 +40,23 @@ BEGIN
|
|||||||
ON dbo.SyncQueue(Status, TargetSchema, TargetTable);
|
ON dbo.SyncQueue(Status, TargetSchema, TargetTable);
|
||||||
END
|
END
|
||||||
GO
|
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
|
||||||
|
|||||||
@@ -87,15 +87,18 @@ class AccessReader:
|
|||||||
|
|
||||||
def delete_log_ids(
|
def delete_log_ids(
|
||||||
self, ids: list[int], batch_size: int, retries: int
|
self, ids: list[int], batch_size: int, retries: int
|
||||||
) -> None:
|
) -> int:
|
||||||
"""Delete the given log-row IDs in chunks, retrying on lock contention.
|
"""Delete the given log-row IDs in chunks, retrying on lock contention.
|
||||||
|
|
||||||
A no-op when ``ids`` is empty (never raises). Retries with linear
|
Returns the total number of rows actually deleted (sum of per-chunk
|
||||||
backoff because the live client may briefly hold a page lock on
|
``cursor.rowcount``), so the caller can report honest counts. A no-op
|
||||||
``TableChangeLog``.
|
(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:
|
if not ids:
|
||||||
return
|
return 0
|
||||||
|
total = 0
|
||||||
cur = self._connect().cursor()
|
cur = self._connect().cursor()
|
||||||
for i in range(0, len(ids), batch_size):
|
for i in range(0, len(ids), batch_size):
|
||||||
chunk = ids[i:i + batch_size]
|
chunk = ids[i:i + batch_size]
|
||||||
@@ -106,12 +109,14 @@ class AccessReader:
|
|||||||
f"DELETE FROM TableChangeLog WHERE ID IN ({placeholders})",
|
f"DELETE FROM TableChangeLog WHERE ID IN ({placeholders})",
|
||||||
*chunk,
|
*chunk,
|
||||||
)
|
)
|
||||||
|
total += cur.rowcount
|
||||||
break
|
break
|
||||||
except pyodbc.OperationalError:
|
except pyodbc.OperationalError:
|
||||||
if attempt < retries - 1:
|
if attempt < retries - 1:
|
||||||
time.sleep(0.2 * (attempt + 1))
|
time.sleep(0.2 * (attempt + 1))
|
||||||
else:
|
else:
|
||||||
raise
|
raise
|
||||||
|
return total
|
||||||
|
|
||||||
def close(self):
|
def close(self):
|
||||||
if self._conn:
|
if self._conn:
|
||||||
|
|||||||
@@ -17,11 +17,20 @@ log = logging.getLogger(__name__)
|
|||||||
def cleanup_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int:
|
def cleanup_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int:
|
||||||
"""Delete applied log rows for ``fm`` from the Access log.
|
"""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
|
Returns the number of log rows actually deleted. A no-op (returns 0) when
|
||||||
are currently in the ``applied`` state for this file.
|
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)
|
ids = writer.applied_log_ids(fm.file)
|
||||||
if not ids:
|
if not ids:
|
||||||
return 0
|
return 0
|
||||||
reader.delete_log_ids(ids, cfg.runtime.cleanup_batch_size, cfg.runtime.cleanup_lock_retries)
|
deleted = reader.delete_log_ids(
|
||||||
return len(ids)
|
ids, cfg.runtime.cleanup_batch_size, cfg.runtime.cleanup_lock_retries
|
||||||
|
)
|
||||||
|
writer.mark_cleaned(fm.file, ids)
|
||||||
|
return deleted
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ class RuntimeConfig(BaseModel):
|
|||||||
retry_backoff_seconds: int = 30
|
retry_backoff_seconds: int = 30
|
||||||
cleanup_batch_size: int = 200
|
cleanup_batch_size: int = 200
|
||||||
cleanup_lock_retries: int = 3
|
cleanup_lock_retries: int = 3
|
||||||
|
cleaned_retention_hours: int = 24
|
||||||
|
|
||||||
class FileMapping(BaseModel):
|
class FileMapping(BaseModel):
|
||||||
model_config = ConfigDict(coerce_numbers_to_str=True)
|
model_config = ConfigDict(coerce_numbers_to_str=True)
|
||||||
|
|||||||
@@ -75,6 +75,16 @@ def cycle(cfg):
|
|||||||
log.exception("cleanup failed for %s", fm.file)
|
log.exception("cleanup failed for %s", fm.file)
|
||||||
finally:
|
finally:
|
||||||
reader.close()
|
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:
|
finally:
|
||||||
writer.close()
|
writer.close()
|
||||||
|
|
||||||
|
|||||||
@@ -90,6 +90,48 @@ class SqlWriter:
|
|||||||
)
|
)
|
||||||
return [r[0] for r in cur.fetchall()]
|
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:
|
def close(self) -> None:
|
||||||
"""Close the underlying pyodbc connection."""
|
"""Close the underlying pyodbc connection."""
|
||||||
self._conn.close()
|
self._conn.close()
|
||||||
|
|||||||
67
tests/test_cleanup_accounting.py
Normal file
67
tests/test_cleanup_accounting.py
Normal file
@@ -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
|
||||||
@@ -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)
|
assert mcapture.call_count == 2 # both files attempted (bad swallowed)
|
||||||
writer_inst.call_apply.assert_called_once() # apply still called
|
writer_inst.call_apply.assert_called_once() # apply still called
|
||||||
assert mcleanup.call_count == 2 # cleanup attempted for both files
|
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
|
writer_inst.close.assert_called() # writer closed in finally
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user