From 3a3d3d029206c817a112ed68808126fc0497aa0e Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Tue, 14 Jul 2026 12:39:22 +0800 Subject: [PATCH] feat: cleanup phase and main service loop with error isolation Wires capture -> apply -> cleanup into cycle(cfg): per-file capture and cleanup each wrapped in try/except + log.exception so one file's failure does not abort the cycle; apply failure does not block cleanup; writer is always closed in finally. run(cfg) loops cycle with sleep; main() loads config from argv. logging_setup uses RotatingFileHandler 10MBx5 + console. Unit tests cover all three error-isolation branches via mocks (no real end-to-end smoke; integration deferred to Task 9 pilot). Co-Authored-By: Claude --- src/sync/cleanup.py | 27 ++++++++++++ src/sync/logging_setup.py | 30 +++++++++++++ src/sync/service.py | 93 +++++++++++++++++++++++++++++++++++++++ tests/test_service.py | 83 ++++++++++++++++++++++++++++++++++ 4 files changed, 233 insertions(+) create mode 100644 src/sync/cleanup.py create mode 100644 src/sync/logging_setup.py create mode 100644 src/sync/service.py create mode 100644 tests/test_service.py diff --git a/src/sync/cleanup.py b/src/sync/cleanup.py new file mode 100644 index 0000000..547a789 --- /dev/null +++ b/src/sync/cleanup.py @@ -0,0 +1,27 @@ +"""Cleanup phase: delete applied Access log rows. + +After ``dbo.usp_SyncApply`` flips queue rows to ``applied``, those rows' +``SourceLogID`` values are no longer needed on the Access side. This module +asks the writer which log IDs have been applied for a given source file and +deletes them from ``TableChangeLog`` via the reader, in batches with lock +retry. The delete is the only mutation that touches the Access side. +""" +from __future__ import annotations +import logging +from .access_reader import AccessReader +from .sql_writer import SqlWriter +from .config import FileMapping, SyncConfig + +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. + """ + 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) diff --git a/src/sync/logging_setup.py b/src/sync/logging_setup.py new file mode 100644 index 0000000..7ca32f9 --- /dev/null +++ b/src/sync/logging_setup.py @@ -0,0 +1,30 @@ +"""Logging configuration for the sync service. + +Configures the root logger with a RotatingFileHandler (10 MB x 5, UTF-8) plus +a console StreamHandler. The level and log path come from ``cfg.logging``. +""" +import logging +import logging.handlers +import os + + +def setup_logging(cfg_dict: dict | None): + """Configure root logging from the ``cfg.logging`` dict. + + ``cfg_dict`` is ``SyncConfig.logging`` (a dict or None). ``level`` is a + logging-level name string (default ``"INFO"``); ``path`` is the log file + path (default ``"sync.log"``). The parent directory is created if missing. + """ + level = getattr(logging, (cfg_dict or {}).get("level", "INFO")) if cfg_dict else logging.INFO + path = (cfg_dict or {}).get("path", "sync.log") + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + h = logging.handlers.RotatingFileHandler( + path, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8" + ) + h.setFormatter(logging.Formatter("%(asctime)s %(levelname)s [%(name)s] %(message)s")) + root = logging.getLogger() + root.setLevel(level) + root.addHandler(h) + sh = logging.StreamHandler() + sh.setFormatter(logging.Formatter("%(levelname)s %(message)s")) + root.addHandler(sh) diff --git a/src/sync/service.py b/src/sync/service.py new file mode 100644 index 0000000..b53c16d --- /dev/null +++ b/src/sync/service.py @@ -0,0 +1,93 @@ +"""Main service loop: capture -> apply -> cleanup, with per-file error isolation. + +``cycle(cfg)`` runs one full pass over every configured file: + 1. capture — read each file's change log and stage rows into SyncQueue; + 2. apply — drain the queue via ``dbo.usp_SyncApply``; + 3. cleanup — delete applied log rows from each file's Access log. + +Each file's capture and cleanup is wrapped in its own try/except so one +file's failure is logged and the cycle continues; the writer is always closed +in a ``finally``. ``run(cfg)`` loops ``cycle`` with a sleep; ``main()`` +loads the config from ``argv[1]`` (default ``config.yaml``). +""" +from __future__ import annotations +import sys +import time +import logging + +from .config import load_config +from .access_reader import AccessReader +from .sql_writer import SqlWriter +from .capture import capture_file +from .cleanup import cleanup_file +from .logging_setup import setup_logging + +log = logging.getLogger("sync.service") + + +def run(cfg): + """Run ``cycle`` forever, sleeping ``poll_interval_seconds`` between passes. + + Configures logging once on entry. Intended to be started by ``main()`` + under the service host (e.g. NSSM). Not unit-tested (infinite loop); + ``cycle()`` is the testable unit. + """ + setup_logging(cfg.logging) + while True: + cycle(cfg) + time.sleep(cfg.runtime.poll_interval_seconds) + + +def cycle(cfg): + """One capture -> apply -> cleanup pass over all files. + + Per-file capture/cleanup failures are logged and do not abort the cycle. + Apply failure does not block cleanup. The writer is always closed in a + ``finally``. Safe to call directly from tests (does not sleep or loop). + """ + writer = SqlWriter(cfg.sql_server.conn_str, cfg.sql_server.sync_queue_table) + try: + total_captured = 0 + for fm in cfg.files: + reader = AccessReader(fm.source_path(cfg), cfg.access.driver) + try: + n = capture_file(fm, reader, writer, cfg) + total_captured += n + except Exception: + log.exception("capture failed for %s", fm.file) + finally: + reader.close() + log.info("captured %d rows", total_captured) + + try: + writer.call_apply(cfg.runtime.max_retries) + log.info("apply done") + except Exception: + log.exception("apply failed") + + for fm in cfg.files: + reader = AccessReader(fm.source_path(cfg), cfg.access.driver) + try: + c = cleanup_file(fm, reader, writer, cfg) + if c: + log.info("cleaned %d log rows from %s", c, fm.file) + except Exception: + log.exception("cleanup failed for %s", fm.file) + finally: + reader.close() + finally: + writer.close() + + +def main(): + """Entry point: load config from argv[1] (default config.yaml) and run.""" + cfg_path = sys.argv[1] if len(sys.argv) > 1 else "config.yaml" + cfg = load_config(cfg_path) + try: + run(cfg) + except KeyboardInterrupt: + log.info("stopped") + + +if __name__ == "__main__": + main() diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..bb5fdc2 --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,83 @@ +"""Unit tests for the service cycle: error isolation + apply-still-called. + +These verify the orchestration contract of ``cycle()`` only. ``run()`` is an +infinite loop and is deliberately not tested here; real end-to-end smoke is +deferred to the Task 9 pilot. +""" +from unittest.mock import MagicMock, patch + +from sync.config import ( + SyncConfig, + AccessConfig, + RuntimeConfig, + SqlServerConfig, + FileMapping, +) +from sync import service + + +def _cfg(): + return SyncConfig( + sql_server=SqlServerConfig(conn_str="x"), + access=AccessConfig(driver="d", roots={"2026": "r"}), + runtime=RuntimeConfig(), + files=[ + FileMapping(file="ok.accdb", root="2026", schema="s", year_suffix="_Y"), + FileMapping(file="bad.accdb", root="2026", schema="s", year_suffix="_Y"), + ], + ) + + +def test_cycle_continues_on_capture_error_and_still_applies_and_cleans(): + """A capture failure on one file must not abort the cycle. + + Verify: (a) capture is attempted for both files, (b) apply is still + called, (c) cleanup is attempted for both files, (d) the writer is + closed in the finally block. + """ + cfg = _cfg() + with patch.object(service, "SqlWriter") as MW, \ + patch.object(service, "AccessReader") as MR, \ + patch.object(service, "capture_file", side_effect=[5, RuntimeError("boom")]) as mcapture, \ + patch.object(service, "cleanup_file", return_value=0) as mcleanup: + writer_inst = MagicMock() + MW.return_value = writer_inst + MR.return_value = MagicMock() # reader instance mock + service.cycle(cfg) # must not raise + 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.close.assert_called() # writer closed in finally + + +def test_cycle_apply_failure_does_not_block_cleanup(): + """Apply failure must not block the cleanup phase.""" + cfg = _cfg() + with patch.object(service, "SqlWriter") as MW, \ + patch.object(service, "AccessReader") as MR, \ + patch.object(service, "capture_file", return_value=3) as mcapture, \ + patch.object(service, "cleanup_file", return_value=7) as mcleanup: + writer_inst = MagicMock() + writer_inst.call_apply.side_effect = RuntimeError("apply boom") + MW.return_value = writer_inst + MR.return_value = MagicMock() + service.cycle(cfg) # must not raise + assert mcapture.call_count == 2 # both files captured + writer_inst.call_apply.assert_called_once() # apply attempted + assert mcleanup.call_count == 2 # cleanup still runs despite apply failure + writer_inst.close.assert_called() + + +def test_cycle_cleanup_failure_does_not_abort_other_files(): + """A cleanup failure on one file must not skip cleanup of the other.""" + cfg = _cfg() + with patch.object(service, "SqlWriter") as MW, \ + patch.object(service, "AccessReader") as MR, \ + patch.object(service, "capture_file", return_value=1), \ + patch.object(service, "cleanup_file", side_effect=[0, RuntimeError("cleanup boom")]) as mcleanup: + writer_inst = MagicMock() + MW.return_value = writer_inst + MR.return_value = MagicMock() + service.cycle(cfg) # must not raise + assert mcleanup.call_count == 2 # second file cleanup still attempted + writer_inst.close.assert_called()