d71b7eab62
- Add sql/00_schema.sql: create dedicated ProductionDataBaseSync schema (idempotent) - Move SyncQueue and usp_SyncApply from dbo into ProductionDataBaseSync - Add sql/03_sync_log_archive.sql: permanent, append-only SyncLogArchive that records both OriginalOperateType and ProcessedOperateType plus the Access log OriginalTime, so pipeline divergences (e.g. Insert applied as Delete) stay reconstructible forever (SyncQueue is transient and only keeps processed type) - config.py: inject sync_queue_table / archive_table / apply_proc (default to the new schema); SqlWriter takes these names instead of hardcoding dbo - sql_writer.py: add ArchiveRow + insert_archive_row (dedup on source keys), parametrize queue/archive/proc names throughout - capture.py: archive every consumed log row before enqueue (preserves evidence before cleanup deletes the Access log) - service.py: pass the three names into SqlWriter - tests: read queue/proc names from config instead of hardcoding dbo.SyncQueue
72 lines
3.7 KiB
Transact-SQL
72 lines
3.7 KiB
Transact-SQL
-- SyncLogArchive: permanent, append-only audit store for every Access change-log
|
|
-- row consumed by the incremental sync. This is the durable evidence layer that
|
|
-- SyncQueue is NOT: SyncQueue is a transient work queue (purged 24h after a row
|
|
-- is cleaned) and it only stores the *processed* OperateType, so a downgrade
|
|
-- (e.g. capture turning an Insert into a Delete when the row is momentarily
|
|
-- unreadable) erases the original intent. This table preserves both the ORIGINAL
|
|
-- operate type recorded by the Access data macro and the PROCESSED type actually
|
|
-- sent to SQL Server, plus the original Access log timestamp and the row payload.
|
|
--
|
|
-- With this in place, cases like the 14287 incident (a real Insert applied to SQL
|
|
-- as a Delete) stay fully reconstructible: OriginalOperateType != ProcessedOperateType
|
|
-- flags exactly where the pipeline diverged from the source.
|
|
--
|
|
-- Retention: PERMANENT. No purge job touches this table. SyncQueue keeps its
|
|
-- short-lived queue role; this table keeps history. Lives under the
|
|
-- ProductionDataBaseSync schema (run 00_schema.sql first).
|
|
-- Idempotent: safe to re-run.
|
|
|
|
IF OBJECT_ID('ProductionDataBaseSync.SyncLogArchive', 'U') IS NULL
|
|
BEGIN
|
|
CREATE TABLE ProductionDataBaseSync.SyncLogArchive (
|
|
ArchiveID bigint IDENTITY(1,1) NOT NULL,
|
|
SourceFile nvarchar(255) NOT NULL,
|
|
SourceTable nvarchar(255) NOT NULL,
|
|
SourceLogID bigint NOT NULL,
|
|
RecordID nvarchar(50) NOT NULL,
|
|
TargetSchema nvarchar(128) NOT NULL,
|
|
TargetTable nvarchar(255) NOT NULL,
|
|
OriginalOperateType varchar(10) NOT NULL, -- as recorded by the Access data macro
|
|
ProcessedOperateType varchar(10) NOT NULL, -- as actually sent to SyncQueue / SQL
|
|
RowData nvarchar(max) NULL, -- captured row payload (NULL for Delete)
|
|
OriginalTime datetime2 NULL, -- Access TableChangeLog.Time (previously discarded)
|
|
CapturedAt datetime2 NOT NULL
|
|
CONSTRAINT DF_SyncLogArchive_Captured DEFAULT sysdatetime(),
|
|
CONSTRAINT PK_SyncLogArchive PRIMARY KEY CLUSTERED (ArchiveID)
|
|
);
|
|
END
|
|
GO
|
|
|
|
-- Dedup: one archive row per source log entry. Capture may re-run the same log
|
|
-- row if a prior cycle's apply failed (the Access log is only deleted after a
|
|
-- successful apply), so the write path uses IF NOT EXISTS on these keys.
|
|
IF NOT EXISTS (SELECT 1 FROM sys.indexes
|
|
WHERE name = 'UX_SyncLogArchive_Dedup'
|
|
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncLogArchive'))
|
|
BEGIN
|
|
CREATE UNIQUE INDEX UX_SyncLogArchive_Dedup
|
|
ON ProductionDataBaseSync.SyncLogArchive(SourceFile, SourceTable, SourceLogID);
|
|
END
|
|
GO
|
|
|
|
-- Evidence lookup by table + record (e.g. "show every log ever seen for ID 14287").
|
|
IF NOT EXISTS (SELECT 1 FROM sys.indexes
|
|
WHERE name = 'IX_SyncLogArchive_Record'
|
|
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncLogArchive'))
|
|
BEGIN
|
|
CREATE INDEX IX_SyncLogArchive_Record
|
|
ON ProductionDataBaseSync.SyncLogArchive(SourceTable, RecordID);
|
|
END
|
|
GO
|
|
|
|
-- Fast filter for the anomaly the archive exists to catch: original != processed.
|
|
IF NOT EXISTS (SELECT 1 FROM sys.indexes
|
|
WHERE name = 'IX_SyncLogArchive_Divergence'
|
|
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncLogArchive'))
|
|
BEGIN
|
|
CREATE INDEX IX_SyncLogArchive_Divergence
|
|
ON ProductionDataBaseSync.SyncLogArchive(OriginalOperateType, ProcessedOperateType)
|
|
INCLUDE (SourceFile, SourceTable, RecordID, CapturedAt);
|
|
END
|
|
GO
|