diff --git a/sql/02_sync_apply.sql b/sql/02_sync_apply.sql index a23cf2c..ee212ab 100644 --- a/sql/02_sync_apply.sql +++ b/sql/02_sync_apply.sql @@ -10,15 +10,33 @@ -- Delete), which let a stale Delete outrank a newer Insert for the same RecordID -- and silently drop a row whose true last op was an Insert. Routing off the one -- winning row's OperateType guarantees only the genuine last op wins. +-- +-- AUDIT NOTE: when ProductionDataBaseSync.SyncApplyRunLog exists (see +-- sql/04_sync_apply_runlog.sql), one audit row is written per target table and +-- per invocation: pending/distinct counts going in, the MERGE/DELETE rowcounts, +-- how many queue rows were flipped to applied (or to error/dead on failure), +-- the error message, the duration, and the caller-supplied @CycleID that ties +-- the row to the Python service log's [cyc:...] lines. The audit insert is +-- best-effort (wrapped in its own TRY/CATCH, outside the data transaction) and +-- can never fail the apply itself. @CycleID defaults to NULL so the legacy +-- single-parameter EXEC keeps working. CREATE OR ALTER PROCEDURE ProductionDataBaseSync.usp_SyncApply - @MaxRetries INT = 5 + @MaxRetries INT = 5, + @CycleID NVARCHAR(40) = NULL AS BEGIN SET NOCOUNT ON; DECLARE @sch NVARCHAR(128), @tbl NVARCHAR(255), @FullName NVARCHAR(514); DECLARE @sql NVARCHAR(MAX), @cols NVARCHAR(MAX), @upd NVARCHAR(MAX), @ins NVARCHAR(MAX); + DECLARE @hasRunLog bit = + CASE WHEN OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog', 'U') IS NOT NULL + THEN 1 ELSE 0 END; + DECLARE @t0 DATETIME2, @pending INT, @distinctRecs INT, + @merged INT, @deleted INT, @applied INT, + @errCnt INT, @deadCnt INT, + @errMsg NVARCHAR(MAX), @outcome VARCHAR(10); -- Re-queue rows still under the retry budget. UPDATE ProductionDataBaseSync.SyncQueue @@ -37,6 +55,19 @@ BEGIN BEGIN SET @FullName = QUOTENAME(@sch) + N'.' + QUOTENAME(@tbl); + -- Per-table audit state. Explicit reset every iteration: local + -- variables keep their previous value across cursor loops. + SELECT @t0 = SYSDATETIME(), + @merged = NULL, @deleted = NULL, @applied = NULL, + @errCnt = NULL, @deadCnt = NULL, + @errMsg = NULL, @outcome = 'ok', + @cols = NULL, @upd = NULL, @ins = NULL; + + SELECT @pending = COUNT(*), + @distinctRecs = COUNT(DISTINCT RecordID) + FROM ProductionDataBaseSync.SyncQueue + WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending'; + BEGIN TRY BEGIN TRAN; -- @cols: comma-quoted column names (for INSERT target list) @@ -66,6 +97,9 @@ BEGIN -- pending ops so the rn=1 row is the true last op for that -- RecordID; an earlier Delete can no longer outrank a newer -- Insert for the same RecordID. + -- AUDIT: @@ROWCOUNT is captured into @MergedOut IMMEDIATELY + -- after the MERGE -- SET IDENTITY_INSERT (like any SET option) + -- resets @@ROWCOUNT, so the order below is load-bearing. SET @sql = N'SET IDENTITY_INSERT ' + @FullName + N' ON;' + N';WITH ranked AS (' + N' SELECT RecordID, OperateType, RowData,' @@ -81,9 +115,11 @@ BEGIN + N' WHEN MATCHED THEN UPDATE SET ' + @upd + N' WHEN NOT MATCHED THEN INSERT (ID,' + @cols + N')' + N' VALUES (TRY_CAST(src.RecordID AS int),' + @ins + N');' + + N'SET @MergedOut = @@ROWCOUNT;' + N'SET IDENTITY_INSERT ' + @FullName + N' OFF;'; EXEC sp_executesql @sql, - N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl; + N'@sch NVARCHAR(128),@tbl NVARCHAR(255),@MergedOut INT OUTPUT', + @sch, @tbl, @MergedOut = @merged OUTPUT; -- Delete: winners (rn=1) whose winning OperateType is Delete. -- Same ranked CTE — only the genuine last op can be a delete. @@ -95,14 +131,17 @@ BEGIN + N')' + N'DELETE t FROM ' + @FullName + N' t' + N' JOIN (SELECT RecordID FROM ranked WHERE rn=1 AND OperateType=''Delete'') d' - + N' ON t.ID = TRY_CAST(d.RecordID AS int);'; + + N' ON t.ID = TRY_CAST(d.RecordID AS int);' + + N'SET @DeletedOut = @@ROWCOUNT;'; EXEC sp_executesql @sql, - N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl; + N'@sch NVARCHAR(128),@tbl NVARCHAR(255),@DeletedOut INT OUTPUT', + @sch, @tbl, @DeletedOut = @deleted OUTPUT; END UPDATE ProductionDataBaseSync.SyncQueue SET Status = 'applied', AppliedAt = SYSDATETIME() WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending'; + SET @applied = @@ROWCOUNT; COMMIT; END TRY @@ -113,13 +152,48 @@ BEGIN -- issues its own rollback. We roll back here so the partial per-table -- work is discarded before marking rows as 'error'/'dead'. IF @@TRANCOUNT > 0 ROLLBACK; + SELECT @errMsg = ERROR_MESSAGE(), @outcome = 'error'; + + -- Split of the previous single CASE update so the audit row can + -- report exactly how many rows died vs. how many will be retried. + -- Net effect on SyncQueue is identical. UPDATE ProductionDataBaseSync.SyncQueue - SET Status = CASE WHEN RetryCount + 1 >= @MaxRetries THEN 'dead' ELSE 'error' END, - RetryCount = RetryCount + 1, - ErrorMsg = ERROR_MESSAGE() + SET Status = 'dead', RetryCount = RetryCount + 1, ErrorMsg = @errMsg + WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending' + AND RetryCount + 1 >= @MaxRetries; + SET @deadCnt = @@ROWCOUNT; + + UPDATE ProductionDataBaseSync.SyncQueue + SET Status = 'error', RetryCount = RetryCount + 1, ErrorMsg = @errMsg WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending'; + SET @errCnt = @@ROWCOUNT; END CATCH + -- Best-effort audit row: sits outside the data transaction (after + -- COMMIT/ROLLBACK) so it survives either outcome, and its own failure + -- can never break the apply loop. + IF @hasRunLog = 1 + BEGIN + BEGIN TRY + INSERT ProductionDataBaseSync.SyncApplyRunLog + (CycleID, TargetSchema, TargetTable, + PendingCount, DistinctRecords, + MergedCount, DeletedCount, AppliedCount, + ErrorCount, DeadCount, + Outcome, ErrorMsg, StartedAt, DurationMs) + VALUES + (@CycleID, @sch, @tbl, + @pending, @distinctRecs, + @merged, @deleted, @applied, + @errCnt, @deadCnt, + @outcome, @errMsg, @t0, + DATEDIFF(millisecond, @t0, SYSDATETIME())); + END TRY + BEGIN CATCH + PRINT 'SyncApplyRunLog insert failed: ' + ERROR_MESSAGE(); + END CATCH + END + FETCH NEXT FROM cur INTO @sch, @tbl; END diff --git a/sql/04_sync_apply_runlog.sql b/sql/04_sync_apply_runlog.sql new file mode 100644 index 0000000..0a0ac97 --- /dev/null +++ b/sql/04_sync_apply_runlog.sql @@ -0,0 +1,75 @@ +-- SyncApplyRunLog: per-invocation, per-target-table audit of what usp_SyncApply +-- actually did. This closes the biggest observability gap in the pipeline: the +-- apply phase used to be a black box ("apply done") -- queue rows could flip +-- to 'error' or 'dead' with no trace in the service log, and there was no +-- record of how many rows a MERGE/DELETE touched at any point in time. With +-- this table, "what did apply do to table X around time T, and did it fail?" +-- is a single indexed query, and CycleID joins each row back to the exact +-- [cyc:xxxxxxxx] lines in the Python service log. +-- +-- Written by usp_SyncApply (sql/02_sync_apply.sql) as a best-effort insert per +-- (invocation, target table). Idle cycles write nothing (the proc's cursor +-- only visits tables that have pending rows), so growth tracks real change +-- traffic, not poll frequency. Retention: unmanaged by default; if it ever +-- grows large, purge by StartedAt, e.g. +-- DELETE FROM ProductionDataBaseSync.SyncApplyRunLog +-- WHERE StartedAt < DATEADD(day, -90, SYSDATETIME()); +-- +-- Lives under the ProductionDataBaseSync schema (run 00_schema.sql first). +-- Idempotent: safe to re-run. Deploy alongside the updated 02_sync_apply.sql; +-- ordering is forgiving either way (the proc checks for this table and skips +-- the audit insert when it is absent). + +IF OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog', 'U') IS NULL +BEGIN + CREATE TABLE ProductionDataBaseSync.SyncApplyRunLog ( + RunLogID bigint IDENTITY(1,1) NOT NULL, + CycleID nvarchar(40) NULL, -- correlation id from the Python service ([cyc:...]) + TargetSchema nvarchar(128) NOT NULL, + TargetTable nvarchar(255) NOT NULL, + PendingCount int NOT NULL, -- pending queue rows seen for this table + DistinctRecords int NULL, -- distinct RecordIDs among them + MergedCount int NULL, -- rows affected by the MERGE (insert + update) + DeletedCount int NULL, -- rows affected by the DELETE + AppliedCount int NULL, -- queue rows flipped to 'applied' + ErrorCount int NULL, -- queue rows flipped to 'error' (will retry) + DeadCount int NULL, -- queue rows flipped to 'dead' (retries exhausted) + Outcome varchar(10) NOT NULL, -- 'ok' | 'error' + ErrorMsg nvarchar(max) NULL, + StartedAt datetime2 NOT NULL + CONSTRAINT DF_SyncApplyRunLog_Started DEFAULT sysdatetime(), + DurationMs int NULL, + CONSTRAINT PK_SyncApplyRunLog PRIMARY KEY CLUSTERED (RunLogID) + ); +END +GO + +-- Join back to the Python service log of one cycle. +IF NOT EXISTS (SELECT 1 FROM sys.indexes + WHERE name = 'IX_SyncApplyRunLog_Cycle' + AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog')) +BEGIN + CREATE INDEX IX_SyncApplyRunLog_Cycle + ON ProductionDataBaseSync.SyncApplyRunLog(CycleID); +END +GO + +-- "What happened to this table around time T?" +IF NOT EXISTS (SELECT 1 FROM sys.indexes + WHERE name = 'IX_SyncApplyRunLog_Table' + AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog')) +BEGIN + CREATE INDEX IX_SyncApplyRunLog_Table + ON ProductionDataBaseSync.SyncApplyRunLog(TargetSchema, TargetTable, StartedAt); +END +GO + +-- Fast scan for failed applies. +IF NOT EXISTS (SELECT 1 FROM sys.indexes + WHERE name = 'IX_SyncApplyRunLog_Outcome' + AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog')) +BEGIN + CREATE INDEX IX_SyncApplyRunLog_Outcome + ON ProductionDataBaseSync.SyncApplyRunLog(Outcome, StartedAt); +END +GO diff --git a/src/sync/access_reader.py b/src/sync/access_reader.py index 06b7ea8..1dfc39c 100644 --- a/src/sync/access_reader.py +++ b/src/sync/access_reader.py @@ -94,7 +94,9 @@ class AccessReader: ``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. + Each retry is logged at WARNING (previously a silent sleep) and the + final lock failure at ERROR before raising, so lock churn on the + production files is visible in the audit trail. """ if not ids: return 0 @@ -121,8 +123,20 @@ class AccessReader: msg = str(e) is_lock = "被锁定" in msg or "-1102" in msg if is_lock and attempt < retries - 1: + log.warning( + "TableChangeLog delete lock contention " + "(attempt %d/%d) db=%s ids=%s..%s -- retrying: %s", + attempt + 1, retries, self.db_path, + chunk[0], chunk[-1], msg[:200], + ) time.sleep(0.2 * (attempt + 1)) else: + if is_lock: + log.error( + "TableChangeLog delete still locked after %d " + "attempts db=%s ids=%s..%s -- giving up", + retries, self.db_path, chunk[0], chunk[-1], + ) raise return total diff --git a/src/sync/capture.py b/src/sync/capture.py index 730f594..c331aaa 100644 --- a/src/sync/capture.py +++ b/src/sync/capture.py @@ -1,5 +1,28 @@ +"""Capture phase: read Access change-log rows and stage them into SyncQueue. + +Every consumed ``TableChangeLog`` row is appended to the permanent audit store +(``SyncLogArchive``) BEFORE it is enqueued, so the original evidence survives +cleanup. On top of that, this module now emits a detailed audit trail to the +service log: + +- WARNING for every operate-type downgrade (Insert/Update -> Delete because + the source row was unreadable at capture time), with record id, log id and + the Access log timestamp -- the exact event class behind the 14287 incident, + previously invisible in the text log; +- WARNING for unknown operate types, with enough identity (log id / record id + / time) to locate and repair the offending log row manually; +- a per-file INFO summary: rows read, newly enqueued, dedup-skipped + (re-capture after a failed apply -- a symptom worth noticing), downgraded, + out-of-scope, unknown ops, the processed log-ID range and a per-operation + breakdown; +- DEBUG detail for individual out-of-scope and dedup skips. +""" from __future__ import annotations -import json, logging + +import json +import logging +from dataclasses import dataclass, field + from .access_reader import AccessReader from .sql_writer import SqlWriter, QueueRow, ArchiveRow from .config import FileMapping, SyncConfig @@ -7,22 +30,92 @@ from .targets import is_synced_table log = logging.getLogger(__name__) -def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int: + +@dataclass +class CaptureStats: + """Counters for one capture pass (per file, or aggregated per cycle).""" + + read: int = 0 # change-log rows read from Access + enqueued: int = 0 # rows newly inserted into SyncQueue + dedup_skipped: int = 0 # already queued (re-capture after a failed apply) + downgraded: int = 0 # Insert/Update downgraded to Delete + out_of_scope: int = 0 # log rows for tables outside the sync scope + unknown_op: int = 0 # log rows with an unrecognised OperateType + min_log_id: int | None = None + max_log_id: int | None = None + ops: dict[str, int] = field(default_factory=dict) # processed op -> count + + def note_op(self, op: str) -> None: + self.ops[op] = self.ops.get(op, 0) + 1 + + def merge(self, other: "CaptureStats") -> None: + """Fold *other* into this instance (cycle-level aggregation).""" + self.read += other.read + self.enqueued += other.enqueued + self.dedup_skipped += other.dedup_skipped + self.downgraded += other.downgraded + self.out_of_scope += other.out_of_scope + self.unknown_op += other.unknown_op + for k, v in other.ops.items(): + self.ops[k] = self.ops.get(k, 0) + v + for attr, pick in (("min_log_id", min), ("max_log_id", max)): + a, b = getattr(self, attr), getattr(other, attr) + if a is None: + setattr(self, attr, b) + elif b is not None: + setattr(self, attr, pick(a, b)) + + def ops_str(self) -> str: + return ",".join(f"{k}={v}" for k, v in sorted(self.ops.items())) or "-" + + +def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, + cfg: SyncConfig) -> CaptureStats: + """Capture one file's pending change-log rows. Returns detailed stats. + + NOTE: previously returned a bare int (rows enqueued); that count is now + ``stats.enqueued``. ``sync.service.cycle`` is the only in-repo caller and + has been updated accordingly. + """ rows = reader.read_log(cfg.runtime.capture_batch_size) - n = 0 + st = CaptureStats(read=len(rows)) + if rows: # read_log orders by ID ascending + st.min_log_id, st.max_log_id = rows[0].id, rows[-1].id + for lr in rows: if not is_synced_table(fm, lr.table_name): + st.out_of_scope += 1 + log.debug("capture skip (out of scope) file=%s table=%s log_id=%s", + fm.file, lr.table_name, lr.id) continue op = lr.operate_type row_data = None if op in ("Insert", "Update"): d = reader.read_row(lr.table_name, lr.record_id) if d is None: - op = "Delete" # 行已删，降级 + # 行已删（或此刻不可读），降级为 Delete。这是数据差异排查的 + # 头号嫌疑事件（参见 14287 事件），必须在文本日志显式留痕， + # 而不只是写入 SyncLogArchive。 + op = "Delete" + st.downgraded += 1 + log.warning( + "capture DOWNGRADE %s->Delete file=%s table=%s " + "record_id=%s log_id=%s log_time=%s " + "(source row unreadable at capture time; original intent " + "preserved in SyncLogArchive.OriginalOperateType)", + lr.operate_type, fm.file, lr.table_name, + lr.record_id, lr.id, lr.time, + ) else: row_data = json.dumps(d, ensure_ascii=False) elif op != "Delete": - log.warning("unknown OperateType %r in %s log %s", op, fm.file, lr.id) + st.unknown_op += 1 + log.warning( + "capture UNKNOWN OperateType %r file=%s table=%s " + "record_id=%s log_id=%s log_time=%s -- row skipped; it will " + "be re-read every cycle until removed from TableChangeLog", + op, fm.file, lr.table_name, lr.record_id, lr.id, lr.time, + ) continue target_schema = fm.schema target_table = fm.target_table(lr.table_name) @@ -52,6 +145,31 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: operate_type=op, row_data=row_data, ) - writer.insert_queue_row(qr) - n += 1 - return n + if writer.insert_queue_row(qr): + st.enqueued += 1 + st.note_op(op) + else: + # Dedup hit: this log row was already staged by an earlier cycle + # whose apply failed (the Access log row is only deleted after a + # successful apply). A persistently non-zero dedup count therefore + # points straight at a stuck apply -- see the queue-health WARNINGs + # emitted by sync.service.cycle. + st.dedup_skipped += 1 + log.debug( + "capture dedup-skip (already queued) file=%s table=%s " + "log_id=%s record_id=%s op=%s", + fm.file, lr.table_name, lr.id, lr.record_id, op, + ) + + if st.read: + log.info( + "capture file=%s read=%d enqueued=%d dedup_skipped=%d " + "downgraded=%d out_of_scope=%d unknown_op=%d " + "log_ids=%s..%s ops={%s}", + fm.file, st.read, st.enqueued, st.dedup_skipped, st.downgraded, + st.out_of_scope, st.unknown_op, st.min_log_id, st.max_log_id, + st.ops_str(), + ) + else: + log.debug("capture file=%s: change log empty", fm.file) + return st diff --git a/src/sync/cleanup.py b/src/sync/cleanup.py index 7864106..4e8fc8a 100644 --- a/src/sync/cleanup.py +++ b/src/sync/cleanup.py @@ -5,6 +5,12 @@ After ``usp_SyncApply`` flips queue rows to ``applied``, those rows' 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. + +Audit trail: the INFO line records the exact log-ID range removed from each +file, and a WARNING is raised when fewer rows were deleted than expected -- +i.e. some applied log IDs were already absent from the Access log (removed +externally, or by a previously interrupted run), which is worth knowing when +reconstructing what happened around a divergence. """ from __future__ import annotations import logging @@ -28,9 +34,25 @@ def cleanup_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: """ ids = writer.applied_log_ids(fm.file) if not ids: + log.debug("cleanup file=%s: no applied rows to clean", fm.file) return 0 + log.debug("cleanup file=%s: deleting %d applied log rows ids=%s..%s", + fm.file, len(ids), ids[0], ids[-1]) deleted = reader.delete_log_ids( ids, cfg.runtime.cleanup_batch_size, cfg.runtime.cleanup_lock_retries ) + if deleted != len(ids): + log.warning( + "cleanup file=%s: deleted %d of %d applied log rows " + "(some log IDs were already absent from Access -- removed " + "externally or by an earlier interrupted run)", + fm.file, deleted, len(ids), + ) writer.mark_cleaned(fm.file, ids) + if deleted: + log.info( + "cleanup file=%s: removed %d access log rows ids=%s..%s; " + "queue rows marked cleaned", + fm.file, deleted, ids[0], ids[-1], + ) return deleted diff --git a/src/sync/config.py b/src/sync/config.py index 51025d7..a020513 100644 --- a/src/sync/config.py +++ b/src/sync/config.py @@ -8,6 +8,10 @@ class SqlServerConfig(BaseModel): sync_queue_table: str = "ProductionDataBaseSync.SyncQueue" archive_table: str = "ProductionDataBaseSync.SyncLogArchive" apply_proc: str = "ProductionDataBaseSync.usp_SyncApply" + # Per-invocation, per-table apply audit written by usp_SyncApply + # (see sql/04_sync_apply_runlog.sql). Default matches the shipped script; + # existing config.yaml files need no change. + apply_runlog_table: str = "ProductionDataBaseSync.SyncApplyRunLog" class AccessConfig(BaseModel): model_config = ConfigDict(coerce_numbers_to_str=True) @@ -23,6 +27,9 @@ class RuntimeConfig(BaseModel): cleanup_batch_size: int = 200 cleanup_lock_retries: int = 3 cleaned_retention_hours: int = 24 + # Idle cycles now log at DEBUG; the service emits an INFO heartbeat at this + # interval while idle so a quiet log still proves the service is alive. + idle_heartbeat_seconds: int = 600 class FileMapping(BaseModel): model_config = ConfigDict(coerce_numbers_to_str=True) diff --git a/src/sync/logging_setup.py b/src/sync/logging_setup.py index e7e1c93..d331838 100644 --- a/src/sync/logging_setup.py +++ b/src/sync/logging_setup.py @@ -3,6 +3,13 @@ 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``. +Every record written to the log file carries a cycle correlation id +(``[cyc:xxxxxxxx]``): ``sync.service.cycle`` allocates one per pass via +``set_cycle_id`` and ``_CycleIdFilter`` injects it into each record, so every +capture/apply/cleanup line of one pass -- and the matching +``SyncApplyRunLog.CycleID`` rows on SQL Server -- can be correlated with a +single grep. Outside a cycle (fullsync, compare, startup) the field is ``-``. + Log files are managed per-day: at startup any previously produced log (including the project's own ``sync.log`` and NSSM's ``nssm_*.log`` captures) is relocated into an ``Archive/`` subfolder next to the active log. Where a log @@ -10,6 +17,7 @@ lacks a timestamp, a ``-YYYY-MM-DD`` suffix is added (derived from its first log line, falling back to mtime) so historical files carry a date. The log root therefore only ever shows the current day's ``sync.log``. """ +import contextvars import datetime import logging import logging.handlers @@ -20,6 +28,33 @@ import shutil _LOG_DATE_RE = re.compile(r"^(\d{4}-\d{2}-\d{2})") +# Current cycle correlation id ("-" when not inside a service cycle). +_cycle_id: contextvars.ContextVar[str] = contextvars.ContextVar( + "sync_cycle_id", default="-" +) + + +def set_cycle_id(cycle_id: str | None) -> None: + """Set (or clear, with ``None``) the id stamped on every log record. + + Called by ``sync.service.cycle`` at the start/end of each pass. The same + id is passed to ``usp_SyncApply`` so SQL-side ``SyncApplyRunLog`` rows can + be joined back to the exact log lines of the cycle that produced them. + """ + _cycle_id.set(cycle_id or "-") + + +class _CycleIdFilter(logging.Filter): + """Inject the current cycle id into every record as ``record.cycle``. + + Attached to the handlers (not the logger) so records emitted through any + module logger -- capture, cleanup, access_reader, ... -- are covered. + """ + + def filter(self, record: logging.LogRecord) -> bool: # noqa: A003 + record.cycle = _cycle_id.get() + return True + def _first_line_date(path: str) -> str | None: """Best-effort extraction of the first log line's YYYY-MM-DD date.""" @@ -118,11 +153,19 @@ def setup_logging(cfg_dict: dict | None): pass root.setLevel(level) + cycle_filter = _CycleIdFilter() + 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")) + h.addFilter(cycle_filter) + h.setFormatter(logging.Formatter( + "%(asctime)s %(levelname)s [%(name)s] [cyc:%(cycle)s] %(message)s" + )) root.addHandler(h) + sh = logging.StreamHandler() + sh.addFilter(cycle_filter) + # Console keeps the short format (fullsync/compare are interactive there). sh.setFormatter(logging.Formatter("%(levelname)s %(message)s")) root.addHandler(sh) diff --git a/src/sync/service.py b/src/sync/service.py index 9e8746a..99b63dc 100644 --- a/src/sync/service.py +++ b/src/sync/service.py @@ -5,6 +5,21 @@ 2. apply — drain the queue via ``usp_SyncApply``; 3. cleanup — delete applied log rows from each file's Access log. +Observability (rebuilt so data divergence is traceable from the log alone): +- every cycle gets a short correlation id; ``logging_setup`` stamps it on each + log line as ``[cyc:xxxxxxxx]`` and the same id is passed to ``usp_SyncApply`` + so ``SyncApplyRunLog`` rows on SQL Server join back to the exact log lines of + the cycle that produced them; +- after apply, the per-table run-log rows (pending/merged/deleted/applied/ + error/dead counts, duration, error message) are read back and logged -- + the old single "apply done" line hid all of this; +- queue health is checked every cycle: ``error``/``dead`` rows, which + previously accumulated in complete silence, now emit WARNINGs with per-row + samples (table, record id, retry count, error message) -- these are exactly + the changes that exist in Access but never reached SQL Server; +- idle cycles log at DEBUG so INFO stays high-signal; ``run`` emits a periodic + idle heartbeat so a quiet log still proves the service is alive. + 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()`` @@ -13,14 +28,15 @@ loads the config from ``argv[1]`` (default ``config.yaml``). from __future__ import annotations import sys import time +import uuid import logging from .config import load_config from .access_reader import AccessReader from .sql_writer import SqlWriter -from .capture import capture_file +from .capture import capture_file, CaptureStats from .cleanup import cleanup_file -from .logging_setup import setup_logging +from .logging_setup import setup_logging, set_cycle_id log = logging.getLogger("sync.service") @@ -30,52 +46,156 @@ def run(cfg): 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. + ``cycle()`` is the testable unit. Emits an idle heartbeat every + ``runtime.idle_heartbeat_seconds`` so a quiet log still proves liveness + now that idle cycles log at DEBUG. """ setup_logging(cfg.logging) + log.info( + "service started: files=%d poll_interval=%ds idle_heartbeat=%ds", + len(cfg.files), cfg.runtime.poll_interval_seconds, + cfg.runtime.idle_heartbeat_seconds, + ) + idle_since = None + idle_cycles = 0 while True: - cycle(cfg) + active = cycle(cfg) + now = time.monotonic() + if active: + idle_since, idle_cycles = None, 0 + else: + idle_cycles += 1 + if idle_since is None: + idle_since = now + elif now - idle_since >= cfg.runtime.idle_heartbeat_seconds: + log.info("idle heartbeat: %d cycles with no changes in the " + "last %ds", idle_cycles, int(now - idle_since)) + idle_since, idle_cycles = now, 0 time.sleep(cfg.runtime.poll_interval_seconds) -def cycle(cfg): +def cycle(cfg) -> bool: """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 + Returns True when the cycle did any work (captured / applied / cleaned / + purged anything); ``run`` uses this for idle-heartbeat pacing. 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). """ + cycle_id = uuid.uuid4().hex[:8] + set_cycle_id(cycle_id) + t0 = time.monotonic() + activity = False writer = SqlWriter( cfg.sql_server.conn_str, cfg.sql_server.sync_queue_table, cfg.sql_server.archive_table, cfg.sql_server.apply_proc, + cfg.sql_server.apply_runlog_table, ) try: - total_captured = 0 + # ---- capture ------------------------------------------------------ + total = CaptureStats() 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 + total.merge(capture_file(fm, reader, writer, cfg)) except Exception: log.exception("capture failed for %s", fm.file) finally: reader.close() - log.info("captured %d rows", total_captured) + if total.read: + activity = True + log.info( + "capture summary: read=%d enqueued=%d dedup_skipped=%d " + "downgraded=%d out_of_scope=%d unknown_op=%d ops={%s}", + total.read, total.enqueued, total.dedup_skipped, + total.downgraded, total.out_of_scope, total.unknown_op, + total.ops_str(), + ) + else: + log.debug("capture summary: no new change-log rows in any file") + # ---- apply ---------------------------------------------------------- try: - writer.call_apply(cfg.runtime.max_retries) - log.info("apply done") + writer.call_apply(cfg.runtime.max_retries, cycle_id) + runs = writer.apply_run_results(cycle_id) + if runs is None: + # Audit infra (SyncApplyRunLog / updated proc) not deployed + # yet: keep the legacy coarse line. sql_writer already warned + # once about what to deploy. + log.info("apply done") + elif not runs: + if total.enqueued: + # Rows were staged but the proc wrote no audit rows: the + # table exists but the deployed proc is probably the old + # version that does not write into it. + log.info("apply done (proc wrote no run-log rows -- " + "re-run the updated sql/02_sync_apply.sql?)") + else: + log.debug("apply done: queue was empty") + else: + activity = True + for r in runs: + if r["Outcome"] == "ok": + log.info( + "apply ok table=%s.%s pending=%s records=%s " + "merged=%s deleted=%s applied=%s dur_ms=%s", + r["TargetSchema"], r["TargetTable"], + r["PendingCount"], r["DistinctRecords"], + r["MergedCount"], r["DeletedCount"], + r["AppliedCount"], r["DurationMs"], + ) + else: + log.warning( + "apply FAILED table=%s.%s pending=%s records=%s " + "-> error=%s dead=%s dur_ms=%s msg=%s", + r["TargetSchema"], r["TargetTable"], + r["PendingCount"], r["DistinctRecords"], + r["ErrorCount"], r["DeadCount"], + r["DurationMs"], r["ErrorMsg"], + ) except Exception: log.exception("apply failed") + # ---- queue health --------------------------------------------------- + # error/dead rows are changes that exist in Access but never reached + # the SQL mirror. Before this check they accumulated with zero trace + # in this log -- the classic "data diverged, no idea why" scenario. + try: + status = writer.queue_status_summary() + stuck = status.get("error", 0) + status.get("dead", 0) + if stuck: + log.warning( + "queue health: %d stuck row(s) [%s] -- these changes are " + "NOT in SQL Server and will show up as data divergence", + stuck, + ",".join(f"{k}={v}" for k, v in sorted(status.items())), + ) + for s in writer.queue_error_samples(10): + log.warning( + " stuck row: table=%s.%s record_id=%s " + "source_log_id=%s op=%s status=%s retries=%s " + "captured_at=%s err=%s", + s["TargetSchema"], s["TargetTable"], s["RecordID"], + s["SourceLogID"], s["OperateType"], s["Status"], + s["RetryCount"], s["CapturedAt"], s["ErrorMsg"], + ) + elif status.get("pending", 0): + log.warning( + "queue health: %d row(s) still pending after apply " + "(apply may have failed this cycle)", status["pending"], + ) + except Exception: + log.exception("queue health check failed") + + # ---- cleanup -------------------------------------------------------- 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) + if cleanup_file(fm, reader, writer, cfg): + activity = True except Exception: log.exception("cleanup failed for %s", fm.file) finally: @@ -86,12 +206,18 @@ def cycle(cfg): try: purged = writer.purge_cleaned(cfg.runtime.cleaned_retention_hours) if purged: - log.info("purged %d cleaned queue rows", purged) + activity = True + log.info("purged %d cleaned queue rows (older than %dh)", + purged, cfg.runtime.cleaned_retention_hours) except Exception: log.exception("purge failed") + (log.info if activity else log.debug)( + "cycle finished in %.2fs", time.monotonic() - t0) + return activity finally: writer.close() + set_cycle_id(None) def main(): diff --git a/src/sync/sql_writer.py b/src/sync/sql_writer.py index 58e5d36..38406b1 100644 --- a/src/sync/sql_writer.py +++ b/src/sync/sql_writer.py @@ -8,19 +8,39 @@ Access change-log row to the permanent audit store (``ProductionDataBaseSync.SyncLogArchive``). The queue/archive/proc names are injected from config so the whole sync can live under a dedicated schema. +Observability additions: +- the dedup inserts now report whether a row was actually inserted (True) or + already present (False), so capture can log dedup hits -- the tell-tale of a + re-capture after a failed apply; +- ``call_apply`` passes the service's cycle correlation id to the proc, which + records one audit row per target table into + ``ProductionDataBaseSync.SyncApplyRunLog`` (see sql/04_sync_apply_runlog.sql); +- ``apply_run_results`` reads those rows back so the service log shows + per-table merged/deleted/applied/error/dead counts instead of "apply done"; +- ``queue_status_summary`` / ``queue_error_samples`` surface stuck + (``error``/``dead``) queue rows, which previously accumulated silently. + The connection is opened with ``autocommit=True`` on purpose. ``usp_SyncApply`` manages its own transaction internally (BEGIN TRAN ... ROLLBACK on error); if the caller held an outer implicit transaction, the proc's ROLLBACK would -cascade and raise SQL error 266. The dedup ``IF NOT EXISTS ... INSERT`` is a -single statement that is atomic under autocommit, so no explicit transaction is -needed on the write path either. +cascade and raise SQL error 266. The dedup ``INSERT .. SELECT .. WHERE NOT +EXISTS`` is a single statement that is atomic under autocommit, so no explicit +transaction is needed on the write path either. """ from __future__ import annotations +import logging from dataclasses import dataclass import pyodbc +log = logging.getLogger(__name__) + +# Process-wide deployment-state flags (SqlWriter instances are recreated every +# cycle, so per-instance flags would re-warn each cycle). +_proc_lacks_cycle_id = False # deployed usp_SyncApply predates @CycleID +_runlog_missing_noted = False # SyncApplyRunLog table not deployed yet + @dataclass class QueueRow: @@ -68,34 +88,35 @@ class SqlWriter: queue_table: str = "ProductionDataBaseSync.SyncQueue", archive_table: str = "ProductionDataBaseSync.SyncLogArchive", apply_proc: str = "ProductionDataBaseSync.usp_SyncApply", + runlog_table: str = "ProductionDataBaseSync.SyncApplyRunLog", ): self.conn_str = conn_str self.queue_table = queue_table self.archive_table = archive_table self.apply_proc = apply_proc + self.runlog_table = runlog_table # autocommit=True: usp_SyncApply manages its own transaction internally. # An outer pyodbc transaction would conflict on ROLLBACK (SQL error 266). self._conn = pyodbc.connect(conn_str, autocommit=True) - def insert_queue_row(self, row: QueueRow) -> None: + def insert_queue_row(self, row: QueueRow) -> bool: """Idempotently enqueue ``row`` (dedup on SourceFile/Table/LogID). - ``IF NOT EXISTS ... INSERT`` is a single statement, atomic under - autocommit. The unique index UX_SyncQueue_Dedup is the DB backstop. + Returns True when a new queue row was inserted, False on a dedup hit + (the row was already staged -- typically a re-capture after a failed + apply left the Access log row in place). ``INSERT .. SELECT .. WHERE + NOT EXISTS`` is a single statement, atomic under autocommit, and its + deterministic rowcount (0/1) is what makes the dedup outcome + observable for the capture audit log. The unique index + UX_SyncQueue_Dedup is the DB backstop. """ - # IF NOT EXISTS and the VALUES list each carry their own ? markers; - # pyodbc binds them positionally, so the 3 dedup keys are supplied - # twice (once for the EXISTS check, once for the INSERT). cur = self._conn.cursor() cur.execute( - f"IF NOT EXISTS (SELECT 1 FROM {self.queue_table} " - "WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?) " - f"INSERT {self.queue_table}(SourceFile,SourceTable,SourceLogID," + f"INSERT INTO {self.queue_table}(SourceFile,SourceTable,SourceLogID," "TargetSchema,TargetTable,RecordID,OperateType,RowData,Status) " - "VALUES (?,?,?,?,?,?,?,?, 'pending')", - row.source_file, - row.source_table, - row.source_log_id, + "SELECT ?,?,?,?,?,?,?,?,'pending' " + f"WHERE NOT EXISTS (SELECT 1 FROM {self.queue_table} " + "WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?)", row.source_file, row.source_table, row.source_log_id, @@ -104,29 +125,32 @@ class SqlWriter: row.record_id, row.operate_type, row.row_data, + row.source_file, + row.source_table, + row.source_log_id, ) # autocommit: statement already committed. + return cur.rowcount > 0 - def insert_archive_row(self, row: ArchiveRow) -> None: + def insert_archive_row(self, row: ArchiveRow) -> bool: """Append ``row`` to the permanent audit store (dedup on source keys). Written during capture, BEFORE cleanup deletes the Access log, so the original evidence survives even after the queue row is purged. Stores both the original and processed operate types plus the Access log time. - ``IF NOT EXISTS`` keeps the first archive record if capture re-runs the - same log id (a prior cycle's apply failed and the Access log persisted). + The ``WHERE NOT EXISTS`` guard keeps the first archive record if + capture re-runs the same log id (a prior cycle's apply failed and the + Access log persisted). Returns True when a new archive row was + written, False on a dedup hit. """ cur = self._conn.cursor() cur.execute( - f"IF NOT EXISTS (SELECT 1 FROM {self.archive_table} " - "WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?) " - f"INSERT {self.archive_table}(SourceFile,SourceTable,SourceLogID," + f"INSERT INTO {self.archive_table}(SourceFile,SourceTable,SourceLogID," "RecordID,TargetSchema,TargetTable,OriginalOperateType," "ProcessedOperateType,RowData,OriginalTime) " - "VALUES (?,?,?,?,?,?,?,?,?,?)", - row.source_file, - row.source_table, - row.source_log_id, + "SELECT ?,?,?,?,?,?,?,?,?,? " + f"WHERE NOT EXISTS (SELECT 1 FROM {self.archive_table} " + "WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?)", row.source_file, row.source_table, row.source_log_id, @@ -137,16 +161,105 @@ class SqlWriter: row.processed_operate_type, row.row_data, row.original_time, + row.source_file, + row.source_table, + row.source_log_id, ) # autocommit: statement already committed. + return cur.rowcount > 0 - def call_apply(self, max_retries: int) -> None: + def call_apply(self, max_retries: int, cycle_id: str | None = None) -> None: """Drain the pending queue via the stored procedure. - ``usp_SyncApply`` flips rows to ``applied`` (or ``error`` after retries). + ``usp_SyncApply`` flips rows to ``applied`` (or ``error``/``dead`` + after retries) and -- once the updated proc plus the SyncApplyRunLog + table are deployed -- writes one audit row per target table tagged + with ``cycle_id``, so SQL-side apply stats join back to the service + log's ``[cyc:...]`` lines. Falls back to the legacy single-parameter + signature when the deployed proc predates ``@CycleID`` (SQL errors + 8144/8145), so rolling out the Python side first keeps working. + """ + global _proc_lacks_cycle_id + cur = self._conn.cursor() + if cycle_id is not None and not _proc_lacks_cycle_id: + try: + cur.execute( + f"EXEC {self.apply_proc} @MaxRetries=?, @CycleID=?", + max_retries, cycle_id, + ) + return + except pyodbc.Error as e: + msg = str(e) + if "8144" in msg or "8145" in msg: + _proc_lacks_cycle_id = True + log.warning( + "apply proc %s does not accept @CycleID yet -- run the " + "updated sql/02_sync_apply.sql to enable per-table " + "apply auditing; falling back to the legacy signature", + self.apply_proc, + ) + else: + raise + cur.execute(f"EXEC {self.apply_proc} ?", max_retries) + + def apply_run_results(self, cycle_id: str) -> list[dict] | None: + """Per-table apply outcomes recorded by the proc for ``cycle_id``. + + Reads ``SyncApplyRunLog`` (pending/distinct counts, MERGE/DELETE + rowcounts, applied/error/dead queue-row counts, duration and error + message per target table). Returns ``None`` when the audit + infrastructure is not deployed yet (legacy proc, or table missing) so + the caller can fall back to coarse logging; returns ``[]`` when the + queue was simply empty. + """ + global _runlog_missing_noted + if _proc_lacks_cycle_id: + return None # legacy proc never writes run-log rows + cur = self._conn.cursor() + try: + cur.execute( + "SELECT TargetSchema,TargetTable,PendingCount,DistinctRecords," + "MergedCount,DeletedCount,AppliedCount,ErrorCount,DeadCount," + "Outcome,ErrorMsg,StartedAt,DurationMs " + f"FROM {self.runlog_table} WHERE CycleID=? ORDER BY RunLogID", + cycle_id, + ) + except pyodbc.Error: + if not _runlog_missing_noted: + _runlog_missing_noted = True + log.warning( + "run-log table %s not available -- run " + "sql/04_sync_apply_runlog.sql to enable per-table apply " + "stats in this log", self.runlog_table, + ) + return None + cols = [c[0] for c in cur.description] + return [dict(zip(cols, r)) for r in cur.fetchall()] + + def queue_status_summary(self) -> dict[str, int]: + """Row counts per Status (pending/applied/error/dead/cleaned).""" + cur = self._conn.cursor() + cur.execute( + f"SELECT Status, COUNT(*) FROM {self.queue_table} GROUP BY Status" + ) + return {r[0]: r[1] for r in cur.fetchall()} + + def queue_error_samples(self, limit: int = 10) -> list[dict]: + """Most recent ``error``/``dead`` queue rows, for WARNING-level triage. + + These are exactly the changes that exist in Access but never reached + the SQL mirror -- the prime suspects for any data divergence. """ cur = self._conn.cursor() - cur.execute(f"EXEC {self.apply_proc} ?", max_retries) + cur.execute( + "SELECT TOP (?) TargetSchema,TargetTable,RecordID,SourceLogID," + "OperateType,Status,RetryCount,ErrorMsg,CapturedAt " + f"FROM {self.queue_table} WHERE Status IN ('error','dead') " + "ORDER BY QueueID DESC", + limit, + ) + cols = [c[0] for c in cur.description] + return [dict(zip(cols, r)) for r in cur.fetchall()] def applied_log_ids(self, source_file: str) -> list[int]: """Return applied SourceLogIDs for ``source_file`` in ascending order."""