Publish or Quarantine, and Closing the Loop

Time	Block
0:00 – 0:28	50.1 Completing WAP: audit in a separate process, publish or quarantine
0:28 – 0:53	50.2 Fault injection: prove the gate works
0:53 – 1:00	Break
1:00 – 1:28	50.3 dbt: labour-market marts from published data
1:28 – 1:45	50.4 Data quality as a warehouse KPI
1:45 – 1:53	Break
1:53 – 2:15	50.5 The final pipeline and runbook
2:15 – 2:30	50.6 Module 5 recap and bridge to orchestration

---

50.1 Completing WAP (28 min)

Two design decisions shape the final orchestrator.

1. Audit in a separate process. GX runs in .venv-gx (Day 42). The orchestrator runs in .venv (dlt). So the orchestrator calls GX as a subprocess. This also solves a subtle DuckDB issue: two different database engines in one process can't both open the same file, but separate processes can take turns. When the subprocess ends, its connection and lock are gone.

2. Publish atomically per table; record the marker afterwards. Copy raw_incoming.<table> → raw_dlt.<table> inside one DuckDB transaction. Then record the release as published (Day 30's rule: markers after success).

Replace main() in ingest_statcan_tables.py and add:

import subprocess
from datetime import datetime, timezone

from macro_ingest.resilience import notify

GX_PYTHON = ".venv-gx/bin/python"


def audit(table):
    """AUDIT: run the GX checkpoint group for one incoming table in the GX environment."""
    proc = subprocess.run([GX_PYTHON, "quality/validate_incoming.py", table])
    return proc.returncode                          # 0 pass · 1 critical · 2 warnings


def publish(con, table):
    """PUBLISH: atomically replace the production raw table with the audited copy."""
    con.execute("begin transaction")
    con.execute(f"create or replace table raw_dlt.{table} as select * from raw_incoming.{table}")
    con.execute("commit")


def quarantine(con, table, stamp):
    """QUARANTINE: keep the failed release for investigation; production is untouched."""
    con.execute("create schema if not exists raw_quarantine")
    con.execute(f"create or replace table raw_quarantine.{table}__{stamp} "
                f"as select * from raw_incoming.{table}")


def record_published(current, product_ids):
    pipeline = dlt.pipeline(pipeline_name="macro_ingest", destination="duckdb",
                            dataset_name="raw_dlt")
    now = datetime.now(timezone.utc)
    pipeline.run([{"product_id": pid, "release_time": current.get(pid), "loaded_at": now}
                  for pid in product_ids],
                 table_name="statcan_releases", write_disposition="append")


def main():
    changed, current = changed_tables()
    if not changed:
        print("✓ No StatCan table republished since last publish — nothing to do.")
        return 0

    write_incoming(changed)                                           # WRITE
    verdicts = {pid: audit(table) for pid, table in changed.items()}  # AUDIT

    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
    published, quarantined, warned = [], [], []
    with duckdb.connect(DB) as con:                                   # PUBLISH / QUARANTINE
        for pid, table in changed.items():
            if verdicts[pid] in (0, 2):
                publish(con, table)
                published.append(pid)
                if verdicts[pid] == 2:
                    warned.append(table)
            else:
                quarantine(con, table, stamp)
                quarantined.append(table)

    if published:
        record_published(current, published)                          # markers after success

    print(f"published: {[changed[p] for p in published]} | quarantined: {quarantined}")
    if quarantined:
        notify(f"⛔ StatCan release QUARANTINED: {quarantined}. Production keeps the last good "
               f"release. Inspect raw_quarantine.*__{stamp} and Data Docs.")
    elif warned:
        notify(f"⚠ StatCan release published with warnings: {warned}. Review Data Docs.")

    if quarantined and not published:
        return 1
    if quarantined or warned:
        return 2
    return 0

Why is this safe?

Failure point	What production sees
download fails during WRITE	nothing changes (raw_incoming is incomplete, but unused)
audit fails (critical)	nothing changes; the release is quarantined; alert
publish transaction fails midway	the transaction rolls back; the old table remains
marker recording fails after publish	data is published; the next run re-publishes the same release (harmless, idempotent)

The table names in the SQL come from your own configuration, not from user input, so building SQL with f-strings is acceptable here. Never do this with values that come from outside (Day 13's SQL-injection warning).

First publish

source .venv/bin/activate
python ingest_statcan_tables.py --full-refresh; echo "exit: $?"
duckdb macro.duckdb -readonly -c "select * from raw_dlt.statcan_releases order by loaded_at desc limit 5"

Now that raw_dlt.statcan_lfs_full exists, register the revisions asset (49.4) and its spec, and rebuild the GX objects.

---

50.2 Fault Injection: Prove the Gate Works (25 min)

A safety gate you haven't seen close is a hope, not a gate. Following Day 29's habit, inject faults deliberately, only into raw_incoming, and confirm the audit catches each one.

Create quality/chaos_incoming.py (runs in .venv-gx; it only modifies the holding schema):

"""Inject deliberate faults into raw_incoming.statcan_lfs_full to test the audit.

NEVER run against raw_dlt. Usage: python quality/chaos_incoming.py <scenario>
Restore afterwards with:  python ingest_statcan_tables.py --full-refresh  (from .venv)
"""
import sys

import duckdb

from gx_helpers import PROJECT_ROOT

DB = str(PROJECT_ROOT.parent / "macro.duckdb")
T = "raw_incoming.statcan_lfs_full"

SCENARIOS = {
    # Unit error: one unemployment rate ×10 → L01 (0–100%) must fail
    "rate_out_of_range": f"""
        update {T} set value = '64.0'
        where rowid = (select min(rowid) from {T}
                       where labour_force_characteristics = 'Unemployment rate'
                         and statistics = 'Estimate' and try_cast(value as double) between 6 and 7)
    """,
    # Missing month inside one series (its 13th month) → L10 (continuity) must fail
    "gap_in_series": f"""
        delete from {T}
        where (vector, ref_date) = (
            select vector, ref_date from {T}
            where vector = (select min(vector) from {T})
            order by ref_date limit 1 offset 12)
    """,
    # Renamed dimension (Gender → Sex) → B01 (required columns) must fail
    "renamed_column": f"alter table {T} rename column gender to sex",
    # Identity broken: unemployment count doubled in one cell → L20/L21 must fail
    "identity_broken": f"""
        update {T} set value = cast(try_cast(value as double) * 2 as varchar)
        where rowid = (select min(rowid) from {T}
                       where labour_force_characteristics = 'Unemployment'
                         and statistics = 'Estimate' and try_cast(value as double) > 100)
    """,
}

if __name__ == "__main__":
    scenario = sys.argv[1]
    with duckdb.connect(DB) as con:
        for statement in SCENARIOS[scenario].split(";"):
            if statement.strip():
                con.execute(statement)
    print(f"⚡ injected scenario '{scenario}' into {T}")

(rowid is DuckDB's internal row number, a handy way to target exactly one row. The gap_in_series scenario deletes the 13th month of one series, inside its range, so it creates a genuine internal gap rather than just shortening the series.)

The test protocol

For each scenario:

source .venv-gx/bin/activate
python quality/chaos_incoming.py rate_out_of_range
python quality/validate_incoming.py statcan_lfs_full; echo "exit: $?"   # expect 1, naming L01
deactivate

source .venv && python ingest_statcan_tables.py --full-refresh           # restore raw_incoming

Record the results in quality/CHAOS_RESULTS.md:

Scenario	Expected failing check	Observed	Exit code	Pass?
rate_out_of_range	L01
gap_in_series	L10
renamed_column	B01 (and the queries depending on gender)
identity_broken	L20 / L21

Then run one scenario through the full orchestrator to see the quarantine path end to end:

# (inside ingest_statcan_tables.py, temporarily skip write_incoming() so the injected data is audited)
python ingest_statcan_tables.py --full-refresh; echo "exit: $?"
duckdb macro.duckdb -readonly -c "select table_name from information_schema.tables where table_schema = 'raw_quarantine'"

Confirm that raw_dlt.statcan_lfs_full is unchanged, a quarantine table exists, and an alert was printed or sent. Then restore the write_incoming() call.

Notice what renamed_column teaches. In the notebook, a Gender/Sex rename was absorbed silently by an if/elif. Here it stops publication and alerts. The right response is then deliberate: update stg_statcan__lfs to the new name, adjust the contract, and re-run. The contract turns a silent patch into a reviewed decision.

---

50.3 dbt: Labour-Market Marts from Published Data (28 min)

Now the warehouse can build on the audited LFS table.

Source and staging

Add to macro_dbt/models/staging/statcan/_statcan__sources.yml:

      - name: lfs_full
        identifier: statcan_lfs_full
        description: "StatCan 14-10-0287-01 (LFS), published only after passing the GX incoming audit (WAP)."

macro_dbt/models/staging/statcan/stg_statcan__lfs.sql:

-- Source: raw_dlt.statcan_lfs_full (audited; WAP)
-- Grain:  one row per month × geography × characteristic × gender × age group × statistic × data type
with source as (

    select * from {{ source('statcan', 'lfs_full') }}

),

renamed as (

    select
        cast(ref_date || '-01' as date)       as month,
        geo,
        labour_force_characteristics          as characteristic,
        gender,
        age_group,
        statistics                            as statistic,
        data_type,
        uom                                   as unit,
        vector                                as vector_id,
        try_cast(value as double)             as value,
        nullif(trim(status), '')              as status_flag,
        coalesce(terminated = 't', false)     as is_terminated
    from source

)

select r.*, g.geo_code, coalesce(g.geo_level, 'other') as geo_level
from renamed as r
left join {{ ref('seed_geo') }} as g using (geo)

Intermediate: the headline labour-market panel

macro_dbt/models/intermediate/labour/int_lfs__headline.sql:

-- Grain: one row per month × geography (Canada + provinces).
-- Seasonally adjusted estimates, all genders, 15 years and over.
select
    month,
    geo,
    geo_code,
    max(value) filter (where characteristic = 'Unemployment rate')   as unemployment_rate,
    max(value) filter (where characteristic = 'Participation rate')  as participation_rate,
    max(value) filter (where characteristic = 'Employment rate')     as employment_rate,
    max(value) filter (where characteristic = 'Employment')          as employment_thousands,
    max(value) filter (where characteristic = 'Labour force')        as labour_force_thousands
from {{ ref('stg_statcan__lfs') }}
where statistic = 'Estimate'
  and data_type = 'Seasonally adjusted'
  and age_group = '15 years and over'
  and gender    = 'Total - Gender'
  and geo_level in ('national', 'province')
group by all

(Use the exact member names from your profile notes.)

Mart: provincial labour markets

macro_dbt/models/marts/labour/lfs_provincial_monthly.sql:

-- Grain: one row per month × geography. Dashboard mart for provincial labour markets.
with h as (

    select * from {{ ref('int_lfs__headline') }}

),

national as (

    select month, unemployment_rate as national_unemployment_rate
    from h where geo_code = 'CA'

)

select
    h.*,
    h.unemployment_rate - n.national_unemployment_rate                       as ur_gap_vs_canada_pp,
    h.unemployment_rate
      - lag(h.unemployment_rate, 12) over (partition by h.geo order by h.month) as ur_change_12m_pp,
    {{ pct_change('h.employment_thousands',
                  'lag(h.employment_thousands, 12) over (partition by h.geo order by h.month)') }}
                                                                             as employment_growth_yoy
from h
left join national as n using (month)

Tests in _labour__models.yml, reusing Module 4's tools:

version: 2

models:
  - name: lfs_provincial_monthly
    description: "Provincial and national labour-market indicators (LFS, SA, 15+, all genders)."
    data_tests:
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns: [month, geo]
    columns:
      - name: month
        data_tests:
          - is_first_of_period: {period: month}
          - dbt_utils.sequential_values:
              interval: 1
              datepart: month
              group_by_columns: ['geo']
      - name: unemployment_rate
        data_tests:
          - within_range: {min_value: 0, max_value: 100}
      - name: participation_rate
        data_tests:
          - within_range: {min_value: 0, max_value: 100}

Yes, 0–100% and continuity appear again, now at the mart. Defence in depth: GX guards the boundary (what StatCan sent), dbt guards the transformation (what we built). If a future staging change broke continuity, dbt would catch it even though the raw data was fine.

Add the mart to the rates_fx_dashboard or a new labour_dashboard exposure (Day 40), then:

cd macro_dbt && source ../.venv/bin/activate
dbt build --target prod --select +lfs_provincial_monthly

---

50.4 Data Quality as a Warehouse KPI (18 min)

On Day 47 you previewed reading the GX history into dbt. Let's finish it, so that data quality appears in the macro cockpit next to the economic KPIs.

macro_dbt/models/staging/quality/_quality__sources.yml (from Day 47), then stg_quality__gx_results.sql:

-- Grain: one row per expectation per GX run
select
    cast(run_time as timestamp)   as run_time,
    checkpoint,
    suite,
    asset,
    plan_id,
    dimension,
    expectation_type,
    column_name,
    success,
    unexpected_count
from {{ source('quality', 'gx_results') }}

macro_dbt/models/marts/quality/quality_latest.sql:

-- Grain: one row per checkpoint, for its most recent run.
with latest as (

    select checkpoint, max(run_time) as run_time
    from {{ ref('stg_quality__gx_results') }}
    group by checkpoint

)

select
    r.checkpoint,
    r.run_time,
    count(*)                                         as checks,
    count(*) filter (where not r.success)            as failures,
    round(100.0 * avg(r.success::int), 1)            as pct_passing,
    list(distinct r.plan_id) filter (where not r.success) as failing_plan_ids
from {{ ref('stg_quality__gx_results') }} as r
join latest using (checkpoint, run_time)
group by all

Add a KPI to macro_kpis (Day 39): a union branch with pillar P0 Data quality, KPI "% warehouse checks passing", taken from quality_latest where checkpoint = 'critical', with signal negative below 100. Now the one-screen status says not only how the economy is doing but how much to trust the numbers.

Ordering matters. quality_latest reads the history written by GX runs. In the final pipeline (next block), the warehouse GX checks run after dbt. So the KPI shows the previous run's warehouse checks, plus the current run's incoming audits. Say so in the model's description. It's a trade-off, not a bug.

---

50.5 The Final Pipeline and Runbook (23 min)

pipeline.sh: the complete flow after Module 5

#!/usr/bin/env bash
# pipeline.sh — APIs → WAP StatCan tables → dbt → warehouse quality → docs
set -e
cd "$(dirname "$0")"
source .venv/bin/activate
export MACRO_DUCKDB_PATH="${MACRO_DUCKDB_PATH:-macro.duckdb}"
export MACRO_DUCKDB_URL="${MACRO_DUCKDB_URL:-duckdb:///$(pwd)/macro.duckdb}"

# 1. API series (BoC, FRED, StatCan vectors): direct, incremental, contracted
python ingest.py --only boc,fred,statcan || [ $? -eq 2 ]

# 2. Full StatCan tables: WRITE → AUDIT (GX) → PUBLISH or QUARANTINE
python ingest_statcan_tables.py || [ $? -eq 2 ]          # 2 = some warnings/quarantine; warehouse stays consistent

# 3. Transform & test (dbt, prod) — builds only from published data
mkdir -p data/marts
( cd macro_dbt && dbt deps --quiet && dbt build --target prod && dbt docs generate --target prod )

# 4. Warehouse quality (GX critical/warning checkpoints, Data Docs, history)
set +e
./quality/run_quality.sh
quality_status=$?
set -e
if [ $quality_status -eq 1 ]; then
    echo "⛔ Critical warehouse quality failure — marts must not be used. See Data Docs."
    exit 1
fi
[ $quality_status -eq 2 ] && echo "⚠ Quality warnings — review Data Docs."

echo "✓ Pipeline complete"

The flow at a glance:

BoC · FRED · StatCan vectors ──► ingest.py ─────────────────────────────► raw_dlt.*_observations
StatCan full tables ────────────► ingest_statcan_tables.py
                                    WRITE  raw_incoming.*
                                    AUDIT  GX incoming_* checkpoints (0–100%, continuity, identities…)
                                    PUBLISH raw_dlt.statcan_*_full   │  QUARANTINE raw_quarantine.*
                                                                       ▼
                                         dbt build (prod): staging → intermediate → marts (+ dbt tests)
                                                                       ▼
                                         GX critical/warning on staging & marts → Data Docs, history
                                                                       ▼
                                         data/marts/*.parquet → dashboards · macro_kpis (incl. quality)

Runbook additions

Append to RUNBOOK.md (Day 30):

## StatCan release quarantined
1. Read the alert: which table, which timestamp.
2. Open Data Docs → latest `incoming_<table>_critical` run → failing expectations and their plan IDs.
3. Run the diagnostic query named in the failing expectation's notes, against
   `raw_quarantine.<table>__<timestamp>` (edit the table name in the query).
4. Decide, and write the decision in quality/specs/REVIEW_LOG.md:
   a) **StatCan error** → keep quarantined; production stays on the previous release.
      Re-check when StatCan republishes (the next release is audited automatically).
   b) **Legitimate change** (e.g. dimension renamed, new member) → update staging and the
      contract spec in one reviewed change; re-run `python ingest_statcan_tables.py --full-refresh`.
   c) **Miscalibrated check** → adjust the threshold with evidence; re-run.
5. Never publish a quarantined table by hand without a written, reviewed decision.

## Chaos testing (quarterly)
Run every scenario in quality/chaos_incoming.py against raw_incoming and update CHAOS_RESULTS.md.

Final check

./pipeline.sh; echo "exit: $?"

---

50.6 Module 5 Recap and Bridge to Orchestration (15 min)

What you built in Module 5 (Days 41–50)

Days	Capability
41	quality dimensions, contracts, severities; a data quality plan
42–43	GX environment, data context, DuckDB data source, assets at every layer, smoke checks
44	suites as code from reviewed YAML specs, linked to the plan
45	profiling that drafts expectations; economists calibrate; checks on changes and distributions
46	validation definitions, critical/warning checkpoints, exit codes, one alert channel
47	Data Docs for non-engineers; a queryable history; trends
48	Write–Audit–Publish; the LFS table; a boundary contract; checkpoint groups
49	0–100% rates and continuous dates, precisely; identities; cross-source and revision checks; diagnostics
50	publish or quarantine; fault injection; labour-market marts; quality as a KPI; the complete pipeline

Defence in depth: where each safeguard lives

Safeguard	Tool	Question
resilient extraction	dlt + resilience layer (Module 3)	Did we receive the data?
schema contracts	dlt (Day 25)	Does it have the shape we expect?
boundary audit (WAP)	GX (Days 48–50)	Is this release fit to publish?
transformation tests	dbt (Day 36)	Did our models preserve meaning?
mart contracts	dbt (Day 40)	Is the product's shape stable?
warehouse validation	GX (Days 44–47)	Is what consumers see correct and fresh?
quality KPI	dbt + GX history	How much should decision-makers trust today's numbers?

What still hurts: the case for orchestration

pipeline.sh works, but look at what it can't do:

Limitation	Example
no schedule	someone must run it; LFS release days aren't known in advance
all-or-nothing retries	a flaky FRED call reruns the whole pipeline
no visibility	"Did last night's run succeed? How long did the audit take?" means reading logs
no concurrency control	two overlapping runs would fight over the DuckDB lock
exit-code juggling	`

Module 6 (Days 51–60): Prefect. Each step becomes a task inside a flow, with per-task retries, caching, scheduling, logging and a UI. On Days 58–60 you'll orchestrate the daily refresh, linking dlt extraction, the WAP audit, dbt transformations and GX validation, triggered by StatCan's release signals from Day 27.

Day 50 checklist

• The orchestrator writes, audits (GX in a separate process), and publishes or quarantines each table atomically
• Release markers are recorded only after successful publication
• Every fault-injection scenario is caught by the intended check, and production data stays untouched
• dbt builds provincial labour-market marts from audited data, with the 0–100% and continuity rules re-tested at the mart
• Data quality appears as a KPI in the macro cockpit, with its timing documented
• pipeline.sh runs the complete flow, and the runbook explains how to handle a quarantined release

git add .
git commit -m "Day 50: WAP publish/quarantine, chaos tests, LFS marts, quality KPI — Module 5 complete"
git push