Refine incremental sync audit observability
- Add per-cycle correlation ID ([cyc:xxxxxxxx]) threaded through Python logs and SQL audit tables for end-to-end traceability of any divergence. - New ProductionDataBaseSync.SyncApplyRunLog table + @CycleID on usp_SyncApply for per-table apply auditing (pending/merged/deleted/applied, dead/error counts, duration). Audit write isolated in its own TRY/CATCH outside txn. - Capture/cleanup phases now emit full detail: Insert->Delete downgrade warning with record_id/log_id, dedup_skipped as apply-stall signal, per-file summary, and access log-id ranges on cleanup. - Queue health check surfaces error/dead rows with recent samples instead of silent accumulation (previously the top cause of data divergence). - sql_writer uses INSERT...SELECT...WHERE NOT EXISTS for observable dedup; idle cycles lowered to DEBUG with periodic heartbeat. - Backward compatible: old proc callers still work (CycleID nullable; legacy coarse-grained logging with one-time notice). Excluded from this commit: CODE.md, build_code_doc.py (doc generation).
This commit is contained in:
Misaka_Company
@@ -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
|
||||
|
||||
|
||||
75
sql/04_sync_apply_runlog.sql
Normal file
75
sql/04_sync_apply_runlog.sql
Normal file
@@ -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
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
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():
|
||||
|
||||
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user