- 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).
- 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
At startup, previously produced logs are relocated into an Archive/
subfolder next to the active log: the project sync.log gets a
-YYYY-MM-DD suffix when archived, and NSSM's nssm_*.log captures are
moved as-is. The log root then only shows the current day's sync.log.
Idempotent handler setup is preserved.
Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
setup_logging unconditionally addHandler'd on every call. Under the new
main.py entry the function was invoked twice for --loop mode (main.py and
service.run both called it), stacking two file + two console handlers so
every log line was written twice.
Clear any pre-existing root handlers before re-adding so repeated calls
always yield exactly one file handler + one console handler, regardless of
the caller. Preserves the standalone python -m sync.service entry point.
ACE ODBC reports lock contention ("无法更新；当前被锁定。(-1102)")
as a generic pyodbc.Error (HY000), NOT pyodbc.OperationalError, so the
old except clause never caught it and gave up after the first collision
— making cleanup_lock_retries:3 a no-op for the 氩弧焊/高写入库 -1102 case.
Now catch pyodbc.Error and retry only on lock messages ("被锁定"/"-1102"),
re-raising other errors immediately instead of retrying them 3×.
- access_reader.delete_log_ids returns the actual rows deleted (was None).
- sql_writer.mark_cleaned flips applied queue rows to 'cleaned' (sets CleanedAt) after their Access log rows are physically removed, so the same IDs are never deleted twice.
- sql_writer.purge_cleaned removes 'cleaned' rows older than a retention window (default 24h) so SyncQueue stops growing without bound.
- cleanup.cleanup_file marks rows cleaned after a successful delete and returns the real delete count, so the service log reports honest 'cleaned N' instead of a constant.
- service.cycle calls purge_cleaned once per pass; config adds cleaned_retention_hours (default 24).
- sql/01_sync_queue.sql adds CleanedAt column + IX_SyncQueue_Cleaned idempotently.
- tests: unit coverage for mark_cleaned/purge_cleaned/delete_log_ids return count; assert cycle purges each pass.
Route both upsert and delete branches off a single ranked CTE (rn=1 per RecordID over all pending ops ordered by SourceLogID DESC). The previous design used two independent ranked CTEs, which let a stale Delete outrank a newer Insert for the same RecordID and silently drop the row. Also gitignore .claude/ and .workbuddy/ runtime dirs.
Wires capture -> apply -> cleanup into cycle(cfg): per-file capture and
cleanup each wrapped in try/except + log.exception so one file's failure
does not abort the cycle; apply failure does not block cleanup; writer is
always closed in finally. run(cfg) loops cycle with sleep; main() loads
config from argv. logging_setup uses RotatingFileHandler 10MBx5 + console.
Unit tests cover all three error-isolation branches via mocks (no real
end-to-end smoke; integration deferred to Task 9 pilot).
Co-Authored-By: Claude <noreply@anthropic.com>
Adds SqlWriter: a pyodbc-backed writer that dedup-inserts into
dbo.SyncQueue (IF NOT EXISTS guarded by UX_SyncQueue_Dedup), invokes
dbo.usp_SyncApply, and reports applied SourceLogIDs.
Connection is opened with autocommit=True per the controller revision:
usp_SyncApply manages its own transaction internally (BEGIN/ROLLBACK),
and an outer pyodbc transaction would conflict on ROLLBACK (SQL error
266). The dedup IF NOT EXISTS...INSERT is a single atomic statement.
Integration test self-cleans via SourceFile='sqlw_test.accdb' marker;
conn_str comes from the gitignored config.yaml (no hardcoded creds).
Co-Authored-By: Claude <noreply@anthropic.com>
Add to_jsonable() (Access/pyodbc value -> JSON-safe) with strict TDD
(5 tests: datetime iso, bool preserved before int, decimal->str,
none/numbers, dict round-trip). bool branch precedes int because bool
is a subclass of int in Python.
Add AccessReader: reads TableChangeLog (read_log), reads a full source
row by RecordID via to_jsonable (read_row, warns on >4000-char values
that JSON_VALUE would truncate), and deletes processed log IDs in
chunked, retried batches (delete_log_ids, no-op on empty list).
Connects via the ACE ODBC driver from config.access.driver, shared
autocommit mode to coexist with the live .laccdb client.
Integration test against the real 氩弧焊.accdb (read-only; delete_log_ids
only exercised with []) passes.
Co-Authored-By: Claude <noreply@anthropic.com>
- sql/01_sync_queue.sql: idempotent DDL for dbo.SyncQueue (PK + unique
dedup index + pending lookup index), safe to re-run.
- sql/02_sync_apply.sql: dbo.usp_SyncApply (@MaxRetries INT=5). Per
distinct (TargetSchema,TargetTable) it builds column projections from
sys.columns (excludes ID key/computed/identity/rowversion) and runs a
dynamic-SQL MERGE (last-write-wins via ROW_NUMBER over SourceLogID DESC)
for Insert/Update plus a DELETE for the last op = Delete.
SET IDENTITY_INSERT ON preserves Access PKs.
- tests/conftest.py: sql_conn fixture reads conn_str from gitignored
config.yaml via load_config; skipped without RUN_INTEGRATION=1.
- tests/test_apply_proc.py: integration test covering IDENTITY-preserving
INSERT, last-write-wins UPDATE, BIT conversion, and DELETE; cleans up.
Deviation from the brief's procedure (root-cause fix, design preserved):
every JSON path key is quoted ('$."col"') so non-ASCII column names
(e.g. Chinese 名字/数量) parse correctly. Without quoting, JSON_VALUE
raises "JSON path format is not correct" on Chinese columns, which is the
real target schema for this Access->SQL Server sync.
Co-Authored-By: Claude <noreply@anthropic.com>
Add Pydantic config models (SqlServerConfig, AccessConfig, RuntimeConfig,
FileMapping, SyncConfig) and a YAML loader (load_config). FileMapping provides
source_path() and target_table() helpers; number-typed YAML keys/values (e.g.
root: 2026) are coerced to str via coerce_numbers_to_str. Includes
config.example.yaml template (config.yaml with real credentials stays
gitignored) and pyproject.toml pytest config (pythonpath=src).
Co-Authored-By: Claude <noreply@anthropic.com>
