Prove an Incremental Pipeline Handles Restart, Duplicate Delivery, Late Data, Deletes, and Reprocessing

import sys, sqlite3print(sys.version)print(sqlite3.sqlite_version)# Generation-time validation: Python 3.13.5 / SQLite 3.46.1.# Record your own versions; do not assume they are identical.

from __future__ import annotationsimport hashlib, json, os, shutil, sqlite3from pathlib import PathBASELINE_SEQ = 200BASELINE_SALES = [    ("O1000",1,"2026-09-18T15:00:00Z","C001","P200",1,75,45,"paid",120),    ("O1001",1,"2026-09-18T09:15:00Z","C001","P100",2,100,60,"paid",121),    ("O1001",2,"2026-09-18T09:15:00Z","C001","P200",1,25,15,"paid",122),    ("O1002",1,"2026-09-18T11:30:00Z","C002","P300",1,190,125,"paid",180),    ("O1003",1,"2026-09-19T15:00:00Z","C001","P400",1,100,60,"paid",123),    ("O1003",2,"2026-09-19T15:00:00Z","C001","P200",2,50,30,"paid",124),    ("O1005",1,"2026-09-20T07:10:00Z","C004","P100",1,50,30,"paid",125),    ("O1005",2,"2026-09-20T07:10:00Z","C004","P400",1,100,60,"paid",126),]BASELINE_CUSTOMERS = [    ("C001","Ada Retail","SMB",110),    ("C002","Ben Home","Consumer",111),    ("C003","Cyra Labs","Enterprise",112),    ("C004","Dara Studio","Consumer",113),]# source_updated_at is deliberately not monotonic by source_seq; seq 205 is clock-skewed backward.EVENTS = [    {"event_id":"e201","source_seq":201,"schema_version":1,"entity":"sale","op":"I","key":"O1006|1","event_ts":"2026-09-21T08:00:00Z","source_updated_at":"2026-09-21T08:01:00Z","after":{"order_id":"O1006","line_no":1,"customer_id":"C002","product_id":"P100","quantity":1,"extended_amount":60,"extended_cost":35,"status":"paid"}},    {"event_id":"e202","source_seq":202,"schema_version":1,"entity":"sale","op":"U","key":"O1003|2","event_ts":"2026-09-19T15:00:00Z","source_updated_at":"2026-09-21T08:02:00Z","before":{"extended_amount":50,"extended_cost":30},"after":{"order_id":"O1003","line_no":2,"customer_id":"C001","product_id":"P200","quantity":2,"extended_amount":55,"extended_cost":30,"status":"paid"}},    {"event_id":"e203","source_seq":203,"schema_version":1,"entity":"customer","op":"U","key":"C002","event_ts":"2026-09-21T08:03:00Z","source_updated_at":"2026-09-21T08:03:00Z","before":{"segment":"Consumer"},"after":{"customer_id":"C002","name":"Ben Home","segment":"Growth"}},    {"event_id":"e204","source_seq":204,"schema_version":1,"entity":"sale","op":"D","key":"O1005|2","event_ts":"2026-09-20T07:10:00Z","source_updated_at":"2026-09-21T08:04:00Z","before":{"order_id":"O1005","line_no":2,"extended_amount":100,"extended_cost":60}},    {"event_id":"e205","source_seq":205,"schema_version":1,"entity":"sale","op":"I","key":"O0999|1","event_ts":"2026-09-17T13:00:00Z","source_updated_at":"2026-09-21T07:59:58Z","after":{"order_id":"O0999","line_no":1,"customer_id":"C003","product_id":"P300","quantity":1,"extended_amount":80,"extended_cost":50,"status":"paid"}},    {"event_id":"e206","source_seq":206,"schema_version":1,"entity":"sale","op":"U","key":"O1002|1","event_ts":"2026-09-18T11:30:00Z","source_updated_at":"2026-09-21T08:05:00Z","before":{"extended_amount":190,"extended_cost":125},"after":{"order_id":"O1002","line_no":1,"customer_id":"C002","product_id":"P300","quantity":1,"extended_amount":195,"extended_cost":125,"status":"paid"}},]BREAKING_EVENT = {"event_id":"e207","source_seq":207,"schema_version":2,"entity":"sale","op":"U","key":"O1001|1","event_ts":"2026-09-18T09:15:00Z","source_updated_at":"2026-09-21T08:06:00Z","after":{"order_id":"O1001","line_no":1,"extended_amount_cents":10000}}def canonical_hash(rows) -> str:    payload=json.dumps(rows, sort_keys=True, separators=(",",":"), ensure_ascii=False)    return hashlib.sha256(payload.encode("utf-8")).hexdigest()def create_db(path: Path) -> sqlite3.Connection:    if path.exists(): path.unlink()    conn=sqlite3.connect(path)    conn.execute("PRAGMA foreign_keys=ON")    conn.executescript("""    CREATE TABLE fact_sales_current(      order_id TEXT NOT NULL, line_no INTEGER NOT NULL, event_ts TEXT NOT NULL,      customer_id TEXT NOT NULL, product_id TEXT NOT NULL, quantity INTEGER NOT NULL,      extended_amount INTEGER NOT NULL, extended_cost INTEGER NOT NULL, status TEXT NOT NULL,      source_seq INTEGER NOT NULL, PRIMARY KEY(order_id,line_no)    );    CREATE TABLE dim_customer_current(      customer_id TEXT PRIMARY KEY, name TEXT NOT NULL, segment TEXT NOT NULL, source_seq INTEGER NOT NULL    );    CREATE TABLE sales_tombstone(      order_id TEXT NOT NULL, line_no INTEGER NOT NULL, deleted_source_seq INTEGER NOT NULL,      event_id TEXT NOT NULL UNIQUE, PRIMARY KEY(order_id,line_no,deleted_source_seq)    );    CREATE TABLE processed_event(      event_id TEXT PRIMARY KEY, source_seq INTEGER NOT NULL UNIQUE, entity TEXT NOT NULL, op TEXT NOT NULL,      payload_sha256 TEXT NOT NULL    );    CREATE TABLE pipeline_state(      stream_name TEXT PRIMARY KEY, committed_watermark INTEGER NOT NULL, last_batch_id TEXT,      updated_at TEXT NOT NULL    );    CREATE TABLE batch_audit(      batch_id TEXT PRIMARY KEY, low_watermark INTEGER NOT NULL, high_watermark INTEGER NOT NULL,      delivery_count INTEGER NOT NULL, distinct_event_count INTEGER NOT NULL,      status TEXT NOT NULL, target_sha256 TEXT, committed_at TEXT    );    """)    conn.executemany("INSERT INTO fact_sales_current VALUES (?,?,?,?,?,?,?,?,?,?)", BASELINE_SALES)    conn.executemany("INSERT INTO dim_customer_current VALUES (?,?,?,?)", BASELINE_CUSTOMERS)    conn.execute("INSERT INTO pipeline_state VALUES ('erp_cdc',?,?,?)",(BASELINE_SEQ,'baseline','2026-09-21T00:00:00Z'))    conn.commit()    return conndef sales_controls(conn):    return conn.execute("""      SELECT count(*), count(DISTINCT order_id), sum(quantity), sum(extended_amount),             sum(extended_cost), sum(extended_amount-extended_cost)      FROM fact_sales_current WHERE status='paid'    """).fetchone()def target_hash(conn):    sales=conn.execute("SELECT * FROM fact_sales_current ORDER BY order_id,line_no").fetchall()    customers=conn.execute("SELECT * FROM dim_customer_current ORDER BY customer_id").fetchall()    tomb=conn.execute("SELECT * FROM sales_tombstone ORDER BY deleted_source_seq").fetchall()    return canonical_hash({"sales":sales,"customers":customers,"tombstones":tomb})def validate_event(e):    if e.get("schema_version") != 1:        raise ValueError(f"unsupported schema_version={e.get('schema_version')} at source_seq={e.get('source_seq')}")    for k in ("event_id","source_seq","entity","op","key","event_ts","source_updated_at"):        if k not in e: raise ValueError(f"missing {k}")    if e["entity"] not in {"sale","customer"}: raise ValueError("unsupported entity")    if e["op"] not in {"I","U","D"}: raise ValueError("unsupported op")def payload_hash(e):    return canonical_hash(e)def apply_event(conn, e):    validate_event(e)    h=payload_hash(e)    # Dedupe is event-identity based. A repeated event_id with a different payload is a hard conflict.    row=conn.execute("SELECT payload_sha256 FROM processed_event WHERE event_id=?",(e["event_id"],)).fetchone()    if row:        if row[0] != h: raise ValueError("conflicting duplicate event_id")        return "duplicate"    # Source sequence is unique in this simulated ordered log.    seqrow=conn.execute("SELECT event_id,payload_sha256 FROM processed_event WHERE source_seq=?",(e["source_seq"],)).fetchone()    if seqrow:        if seqrow != (e["event_id"], h): raise ValueError("conflicting source_seq")        return "duplicate"    if e["entity"]=="sale":        oid,line=e["key"].split("|"); line=int(line)        if e["op"] in {"I","U"}:            a=e["after"]            current=conn.execute("SELECT source_seq FROM fact_sales_current WHERE order_id=? AND line_no=?",(oid,line)).fetchone()            if current is None or e["source_seq"] > current[0]:                conn.execute("""                  INSERT INTO fact_sales_current(order_id,line_no,event_ts,customer_id,product_id,quantity,extended_amount,extended_cost,status,source_seq)                  VALUES(?,?,?,?,?,?,?,?,?,?)                  ON CONFLICT(order_id,line_no) DO UPDATE SET                    event_ts=excluded.event_ts, customer_id=excluded.customer_id, product_id=excluded.product_id,                    quantity=excluded.quantity, extended_amount=excluded.extended_amount,                    extended_cost=excluded.extended_cost, status=excluded.status, source_seq=excluded.source_seq                  WHERE excluded.source_seq > fact_sales_current.source_seq                """,(oid,line,e["event_ts"],a["customer_id"],a["product_id"],a["quantity"],a["extended_amount"],a["extended_cost"],a["status"],e["source_seq"]))        else:            current=conn.execute("SELECT source_seq FROM fact_sales_current WHERE order_id=? AND line_no=?",(oid,line)).fetchone()            if current is not None and e["source_seq"] > current[0]:                conn.execute("DELETE FROM fact_sales_current WHERE order_id=? AND line_no=?",(oid,line))            conn.execute("INSERT OR IGNORE INTO sales_tombstone VALUES (?,?,?,?)",(oid,line,e["source_seq"],e["event_id"]))    else:        a=e["after"]        conn.execute("""          INSERT INTO dim_customer_current(customer_id,name,segment,source_seq) VALUES(?,?,?,?)          ON CONFLICT(customer_id) DO UPDATE SET name=excluded.name,segment=excluded.segment,source_seq=excluded.source_seq          WHERE excluded.source_seq > dim_customer_current.source_seq        """,(a["customer_id"],a["name"],a["segment"],e["source_seq"]))    conn.execute("INSERT INTO processed_event VALUES (?,?,?,?,?)",(e["event_id"],e["source_seq"],e["entity"],e["op"],h))    return "applied"def process_batch(conn, deliveries, batch_id, crash_after=None):    low=conn.execute("SELECT committed_watermark FROM pipeline_state WHERE stream_name='erp_cdc'").fetchone()[0]    candidate=[e for e in deliveries if e["source_seq"]>low]    high=max([e["source_seq"] for e in candidate], default=low)    before_hash=target_hash(conn)    try:        conn.execute("BEGIN")        # Audit row is part of same transaction here; production systems may keep separate durable run telemetry.        conn.execute("INSERT INTO batch_audit VALUES (?,?,?,?,?,?,?,?)",                     (batch_id,low,high,len(deliveries),len({e['event_id'] for e in candidate}),'running',None,None))        applied=dupes=0        for i,e in enumerate(sorted(candidate, key=lambda x:(x["source_seq"],x["event_id"])),1):            outcome=apply_event(conn,e)            applied += outcome=="applied"; dupes += outcome=="duplicate"            if crash_after is not None and i==crash_after:                raise RuntimeError(f"simulated crash after delivery {i}")        # Critical rule: watermark moves in the same transaction as target changes.        conn.execute("UPDATE pipeline_state SET committed_watermark=?,last_batch_id=?,updated_at=? WHERE stream_name='erp_cdc'",                     (high,batch_id,'2026-09-21T08:10:00Z'))        th=target_hash(conn)        conn.execute("UPDATE batch_audit SET status='committed',target_sha256=?,committed_at=? WHERE batch_id=?",                     (th,'2026-09-21T08:10:00Z',batch_id))        conn.commit()        return {"status":"committed","low":low,"high":high,"applied":applied,"duplicates":dupes,"target_sha256":th}    except Exception:        conn.rollback()        assert target_hash(conn)==before_hash        raisedef timestamp_watermark_demo():    # If the cursor had advanced to 08:04 based only on source_updated_at, the seq-205 event would be missed.    cursor="2026-09-21T08:04:00Z"    missed=[e["event_id"] for e in EVENTS if e["source_seq"]>204 and e["source_updated_at"]<=cursor]    overlap_cursor="2026-09-21T07:59:00Z"    reread=[e["event_id"] for e in EVENTS if e["source_updated_at"]>overlap_cursor]    return missed,rereaddef main():    root=Path(os.environ.get("ATLASMART_LAB_HOME","atlasmart_ch15_lab"))    if root.exists(): shutil.rmtree(root)    root.mkdir(parents=True)    conn=create_db(root/"warehouse.db")    baseline=sales_controls(conn); baseline_hash=target_hash(conn)    assert baseline==(8,5,10,690,425,265), baseline    assert conn.execute("SELECT committed_watermark FROM pipeline_state").fetchone()[0]==200    # At-least-once delivery fixture includes duplicates; order of delivery is not trusted.    deliveries=[EVENTS[0],EVENTS[1],EVENTS[1],EVENTS[2],EVENTS[3],EVENTS[4],EVENTS[0],EVENTS[5]]    crashed=False    try:        process_batch(conn,deliveries,"B20260921-CDC-A",crash_after=3)    except RuntimeError as e:        crashed=True        crash_message=str(e)    assert crashed    after_crash=sales_controls(conn)    after_crash_hash=target_hash(conn)    wm_after_crash=conn.execute("SELECT committed_watermark FROM pipeline_state").fetchone()[0]    assert after_crash==baseline and after_crash_hash==baseline_hash and wm_after_crash==200    result=process_batch(conn,deliveries,"B20260921-CDC-B")    final=sales_controls(conn)    final_hash=target_hash(conn)    wm=conn.execute("SELECT committed_watermark FROM pipeline_state").fetchone()[0]    c002=conn.execute("SELECT segment,source_seq FROM dim_customer_current WHERE customer_id='C002'").fetchone()    tomb=conn.execute("SELECT order_id,line_no,deleted_source_seq FROM sales_tombstone").fetchall()    assert final==(9,7,11,740,450,290), final    assert wm==206 and c002==("Growth",203)    assert tomb==[("O1005",2,204)]    # Rerun/redelivery is safe: everything is <= committed watermark, so no target change.    rerun=process_batch(conn,deliveries,"B20260921-CDC-C")    assert sales_controls(conn)==final and target_hash(conn)==final_hash and rerun["high"]==206    missed,reread=timestamp_watermark_demo()    assert missed==["e205"] and "e205" in reread    # Breaking schema event must not move target or watermark.    pre_break_hash=target_hash(conn); pre_break_wm=wm    schema_blocked=False    try:        process_batch(conn,[BREAKING_EVENT],"B20260921-CDC-D")    except ValueError as e:        schema_blocked=True; schema_message=str(e)    assert schema_blocked    assert target_hash(conn)==pre_break_hash    assert conn.execute("SELECT committed_watermark FROM pipeline_state").fetchone()[0]==pre_break_wm    summary={      "baseline_controls":baseline,"baseline_watermark":200,"baseline_sha256":baseline_hash,      "crash":crash_message,"watermark_after_crash":wm_after_crash,"controls_after_crash":after_crash,      "commit_result":result,"final_controls":final,"final_watermark":wm,"final_sha256":final_hash,      "c002":c002,"tombstones":tomb,"rerun":rerun,      "timestamp_cursor_missed":missed,"overlap_reread":reread,      "schema_blocked":schema_message,      "python_sqlite":sqlite3.sqlite_version,    }    (root/"acceptance_summary.json").write_text(json.dumps(summary,indent=2),encoding="utf-8")    print("baseline controls:",baseline,"watermark:",200)    print("crash:",crash_message)    print("after crash controls:",after_crash,"watermark:",wm_after_crash,"hash_unchanged:",after_crash_hash==baseline_hash)    print("commit:",result)    print("final controls:",final,"watermark:",wm)    print("customer C002:",c002)    print("tombstones:",tomb)    print("timestamp-only cursor would miss:",missed)    print("overlap window rereads:",reread)    print("rerun target unchanged:",target_hash(conn)==final_hash)    print("schema gate:",schema_message,"watermark_unchanged:",conn.execute("SELECT committed_watermark FROM pipeline_state").fetchone()[0]==206)    print("final_sha256:",final_hash)    print("cleanup: remove",root)    conn.close()if __name__=="__main__":    main()

baseline controls: (8, 5, 10, 690, 425, 265) watermark: 200crash: simulated crash after delivery 3after crash controls: (8, 5, 10, 690, 425, 265) watermark: 200 hash_unchanged: Truecommit: {'status': 'committed', 'low': 200, 'high': 206, 'applied': 6, 'duplicates': 2, ...}final controls: (9, 7, 11, 740, 450, 290) watermark: 206customer C002: ('Growth', 203)tombstones: [('O1005', 2, 204)]timestamp-only cursor would miss: ['e205']rerun target unchanged: Trueschema gate: unsupported schema_version=2 at source_seq=207 watermark_unchanged: Truefinal_sha256: 3e0885d67598cc472a4678334b7f4a23e87eeb314efa4a6c405b2db4c43fc3ff

# Linux/macOS/Git Bashrm -rf atlasmart_ch15_labrm -f atlasmart_ch15.py# PowerShell# Remove-Item -Recurse -Force .\atlasmart_ch15_lab# Remove-Item -Force .\atlasmart_ch15.py