Event-Driven Orchestration: Reacting to StatCan Releases

Time	Block
0:00 – 0:18	59.1 Schedules vs events
0:18 – 0:40	59.2 Fixing the lock: readers and writers
0:40 – 0:48	Break
0:48 – 1:20	59.3 Write–Audit–Publish as a flow
1:20 – 1:43	59.4 The release watcher: events, triggers, idempotency
1:43 – 1:50	Break
1:50 – 2:10	59.5 Release briefings: telling economists what changed
2:10 – 2:30	59.6 Mini-project: simulate a release, end to end

---

59.1 Schedules vs Events (18 min)

Schedule-driven	Event-driven
"run at 09:15 and check for news"	"run when news arrives"
simple, predictable	timely, efficient
lag between release and dashboard can be hours	lag of minutes
runs even when nothing happened	runs only when something happened

StatCan publishes at 08:30 ET on release days, but which days isn't fixed in our schedules (CPI, LFS and GDP each have their own calendars). A watcher that polls StatCan's metadata every few minutes during the release window, and starts WAP as soon as something changes, combines the best of both: cheap checks, prompt reactions.

Prefect supports this pattern in two complementary ways:

1. Events. Your code emits named events (e.g. statcan.table.republished). They appear in the UI's event feed and can trigger deployments via automations.
2. Direct triggering. A flow calls run_deployment(...) for another deployment (Day 56).

We'll do both: direct triggering for guaranteed behaviour, and events for observability (with an optional automation).

---

59.2 Fixing the Lock: Readers and Writers (23 min)

On Day 57 we serialised writers with duckdb-writer. But DuckDB's rule across processes is stricter:

A database file can be opened either by one read-write process, or by any number of read-only processes, but not both at once.

So a GX audit (a read-only process) starting while dlt is writing will fail to open the file. We need a readers–writer lock: many readers or one writer.

Prefect's global concurrency limits make this neat with occupy: give the limit N slots; readers occupy 1 slot, writers occupy all N.

prefect gcl create duckdb --limit 10

Replace orchestration/tasks/locks.py:

"""Readers–writer access to macro.duckdb, enforced across all Prefect runs."""
from contextlib import contextmanager

from prefect.concurrency.sync import concurrency

DUCKDB_LIMIT = "duckdb"
DUCKDB_SLOTS = 10          # must equal the limit created with `prefect gcl create duckdb --limit 10`


@contextmanager
def duckdb_access(write: bool):
    """Writers take all slots (exclusive); readers take one (shared, up to 10 at once)."""
    with concurrency(DUCKDB_LIMIT, occupy=DUCKDB_SLOTS if write else 1):
        yield

Holder	Slots	Can run alongside
writer (dlt load, WAP publish, dbt build)	10 of 10	nobody
reader (GX audit, fingerprint, briefing, backup)	1 of 10	other readers

Apply it consistently:

• writers: ingest_source, WAP write and publish, dbt_invoke(writes=True);
• readers: GX audits and warehouse checks (wrap the subprocess call), fingerprints, briefings, backups.

Delete the old limit (prefect gcl delete duckdb-writer) once nothing uses it.

Starvation note: with this scheme a long stream of readers could delay a waiting writer. Our readers are short and infrequent, so it isn't a problem in practice. But it's the classic trade-off of readers–writer locks, worth knowing.

---

59.3 Write–Audit–Publish as a Flow (33 min)

Day 50's ingest_statcan_tables.py performs WAP in one script. As a flow, each stage and table becomes visible, retryable where appropriate, and lockable correctly.

Library first

Move the functions from ingest_statcan_tables.py into macro_ingest/wap.py: published_release_times(), write_incoming(changed), audit(table), publish(con, table), quarantine(con, table, stamp), record_published(current, product_ids). Keep the script as a thin CLI over them (the 58.2 pattern).

Tasks

Create orchestration/tasks/wap.py:

"""Write–Audit–Publish steps as Prefect tasks with correct DuckDB access."""
import subprocess
from datetime import datetime, timezone

import duckdb
from prefect import get_run_logger, task

from macro_ingest import wap
from macro_ingest.statcan_tables import current_release_times
from orchestration.settings import ENV, GX_PY, REPO
from orchestration.tasks.ingest import retry_if_transient_exception
from orchestration.tasks.locks import duckdb_access


@task(retries=2, retry_delay_seconds=60)
def detect_changes(tables: dict, force: bool = False) -> tuple[dict, dict]:
    current = current_release_times(tables.keys())
    with duckdb_access(write=False):
        published = wap.published_release_times()
    changed = {pid: name for pid, name in tables.items()
               if force or current.get(pid) != published.get(pid)}
    get_run_logger().info(f"changed tables: {changed or 'none'}")
    return changed, current


@task(retries=1, retry_delay_seconds=900, retry_condition_fn=retry_if_transient_exception,
      timeout_seconds=3600)
def write_table(pid: str, table: str):
    with duckdb_access(write=True):
        wap.write_incoming({pid: table})


@task(timeout_seconds=1800)
def audit_table(table: str) -> int:
    """GX in its own environment and process; read-only access."""
    with duckdb_access(write=False):
        proc = subprocess.run([GX_PY, "quality/validate_incoming.py", table],
                              cwd=REPO, env=ENV, capture_output=True, text=True)
    logger = get_run_logger()
    for line in proc.stdout.splitlines():
        (logger.error if "✗" in line else logger.info)(f"[audit:{table}] {line}")
    return proc.returncode                          # 0 pass · 1 critical · 2 warnings


@task
def publish_or_quarantine(pid: str, table: str, verdict: int) -> str:
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
    with duckdb_access(write=True):
        with duckdb.connect(ENV["MACRO_DUCKDB_PATH"]) as con:
            if verdict in (0, 2):
                wap.publish(con, table)
                return "published_with_warnings" if verdict == 2 else "published"
            wap.quarantine(con, table, stamp)
            return "quarantined"


@task
def record_markers(current: dict, published_pids: list[str]):
    if published_pids:
        with duckdb_access(write=True):
            wap.record_published(current, published_pids)

The flow

Create orchestration/flows/statcan_release.py:

"""Event-driven WAP: audit and publish StatCan tables, then refresh what depends on them."""
import dlt
from prefect import flow, get_run_logger
from prefect.events import emit_event
from prefect.states import Completed, Failed

from orchestration.flows.daily_refresh import quality_flow, publish_flow   # Day 60 adds publish_flow
from orchestration.tasks.wap import (audit_table, detect_changes, publish_or_quarantine,
                                     record_markers, write_table)


@flow(name="statcan-release")
def statcan_release(product_ids: list[str] | None = None, force: bool = False,
                    target: str = "prod"):
    logger = get_run_logger()
    tables = dlt.config["sources.statcan_tables.tables"]
    if product_ids:
        tables = {pid: t for pid, t in tables.items() if pid in product_ids}

    changed, current = detect_changes(tables, force)
    if not changed:
        return Completed(name="NothingToDo", message="no republished tables")

    outcomes = {}
    for pid, table in changed.items():                     # sequential: writes need the lock
        write_table.with_options(name=f"write:{table}")(pid, table)
        verdict = audit_table.with_options(name=f"audit:{table}")(table)
        outcomes[pid] = publish_or_quarantine.with_options(name=f"publish:{table}")(
            pid, table, verdict)
        emit_event(event=f"statcan.table.{outcomes[pid]}",
                   resource={"prefect.resource.id": f"statcan.table.{pid}",
                             "statcan.table": table},
                   payload={"release_time": current.get(pid)})

    published = [pid for pid, o in outcomes.items() if o.startswith("published")]
    record_markers(current, published)

    if published:                                          # refresh only what changed (Day 58)
        from orchestration.flows.daily_refresh import transform_flow
        transform_flow(target=target)
        quality_flow()

    quarantined = [changed[p] for p, o in outcomes.items() if o == "quarantined"]
    if quarantined and not published:
        return Failed(message=f"all changed tables quarantined: {quarantined}")
    if quarantined or any(o.endswith("warnings") for o in outcomes.values()):
        return Completed(name="CompletedWithWarnings", message=f"outcomes: {outcomes}")
    return Completed(message=f"published: {[changed[p] for p in published]}")

Because transform_flow uses per-source fingerprints (58.5), a new LFS release rebuilds only the LFS-dependent models.

Add the deployment to prefect.yaml (no schedule: it's triggered):

  - name: wap
    description: "Write–Audit–Publish for republished StatCan tables, then targeted refresh."
    entrypoint: orchestration/flows/statcan_release.py:statcan_release
    work_pool:
      name: local-process
      job_variables:
        env:
          PYTHONPATH: /workspaces/macro-data-course
    concurrency_limit: 1
    tags: [macro, prod, statcan]

---

59.4 The Release Watcher: Events, Triggers, Idempotency (23 min)

Create orchestration/flows/release_watcher.py:

"""Poll StatCan release metadata; start WAP as soon as a configured table is republished."""
import dlt
from prefect import flow, get_run_logger
from prefect.deployments import run_deployment
from prefect.events import emit_event

from orchestration.tasks.wap import detect_changes


@flow(name="statcan-release-watcher")
def release_watcher(simulate: list[str] | None = None):
    logger = get_run_logger()
    tables = dlt.config["sources.statcan_tables.tables"]
    changed, current = detect_changes(tables)
    pids = sorted(set(changed) | set(simulate or []))

    if not pids:
        logger.info("No StatCan table republished.")
        return

    for pid in pids:
        emit_event(event="statcan.table.republished",
                   resource={"prefect.resource.id": f"statcan.table.{pid}",
                             "statcan.table": tables[pid]},
                   payload={"release_time": current.get(pid), "simulated": pid in (simulate or [])})

    key = "wap-" + "-".join(f"{p}@{current.get(p)}" for p in pids)
    run = run_deployment("statcan-release/wap",
                         parameters={"product_ids": pids, "force": bool(simulate)},
                         idempotency_key=key, timeout=0)
    logger.info(f"Triggered statcan-release/wap ({run.name}) for {pids}")

• idempotency_key ensures that if the watcher sees the same release twice (e.g. it runs again before WAP has published), Prefect reuses the existing triggered run instead of starting a duplicate. The key includes each table's release time, so a genuinely new release gets a new key.
• emit_event records each detected release in the UI's Event Feed, with the table as the event's resource, so you can filter events by table.

Schedule it during the release window: every 10 minutes, 08:30–10:50 ET on weekdays.

  - name: statcan-watcher
    entrypoint: orchestration/flows/release_watcher.py:release_watcher
    work_pool:
      name: local-process
      job_variables:
        env:
          PYTHONPATH: /workspaces/macro-data-course
    concurrency_limit: 1
    schedules:
      - cron: "*/10 8-10 * * 1-5"
        timezone: "America/Toronto"

(It also runs at 08:00–08:20. Those polls find nothing and cost one small metadata request each.)

Optional: an automation instead of run_deployment

Prefect deployments can declare triggers that start a run when a matching event occurs. In prefect.yaml, under the wap deployment:

    triggers:
      - enabled: false                 # enable to replace the watcher's run_deployment call
        match:
          prefect.resource.id: "statcan.table.*"
        expect:
          - "statcan.table.republished"
        parameters:
          force: false

With such a trigger, the watcher only needs to emit events. The trade-off: event-triggered runs are elegant and decoupled, while run_deployment is explicit and easy to test. Check the trigger syntax against your Prefect version's documentation before enabling it. We keep it disabled and rely on the watcher's direct call.

What happens to the morning schedule?

prod-morning (09:15 ET, Day 56) now overlaps with the watcher. Keep it as a safety net: if the watcher missed a release (e.g. the worker was down at 08:30), the morning refresh catches it, because WAP is idempotent. Once a release is published, its marker makes later runs find nothing to do (Day 50). Overlapping runs are serialised by the duckdb limit and deployment concurrency limits.

---

59.5 Release Briefings: Telling Economists What Changed (20 min)

When a release is published, the people who care want the economics, not the plumbing: "September LFS: unemployment 6.9% (+0.1pp); Alberta highest at …". Create a release-specific briefing in orchestration/tasks/reporting.py:

RELEASE_BRIEFS = {
    "14100287": ("Labour Force Survey", """
        with latest as (select max(month) as m from marts.lfs_provincial_monthly)
        select geo, unemployment_rate, ur_change_12m_pp, ur_gap_vs_canada_pp
        from marts.lfs_provincial_monthly, latest
        where month = latest.m
        order by unemployment_rate desc
    """),
    "18100004": ("Consumer Price Index", """
        select month, headline_yoy, core_ex_fe_yoy, median_major_yoy, diffusion_above_3
        from marts.macro_monthly
        where headline_yoy is not null
        order by month desc limit 3
    """),
}


@task
def publish_release_briefing(product_ids: list[str]) -> str:
    sections = []
    with duckdb_access(write=False):
        with duckdb.connect(ENV["MACRO_DUCKDB_PATH"], read_only=True) as con:
            for pid in product_ids:
                if pid not in RELEASE_BRIEFS:
                    continue
                title, sql = RELEASE_BRIEFS[pid]
                df = con.sql(sql).df().round(2)
                sections.append(f"## {title}\n\n{df.to_markdown(index=False)}")
    md = "# New StatCan release published\n\n" + "\n\n".join(sections)
    create_markdown_artifact(key="statcan-release-briefing", markdown=md,
                             description="Economic summary of newly published StatCan data")
    return md

(DataFrame.to_markdown needs the small tabulate package: pip install tabulate.)

Call it at the end of statcan_release when anything was published, and include a one-line headline in the notification. For the LFS, compute the national rate and its monthly change from the first rows. Economists receive the news; engineers have the logs.

---

59.6 Mini-Project: Simulate a Release, End to End (20 min)

You can't wait for StatCan's calendar, so use the watcher's simulate parameter to force the path:

prefect deploy --all
prefect deployment run 'statcan-release-watcher/statcan-watcher' --param 'simulate=["14100287"]'

Verify, using the UI:

1. Event Feed: a statcan.table.republished event for resource statcan.table.14100287.
2. The watcher run triggered statcan-release/wap.
3. In the WAP run: write:statcan_lfs_full → audit:statcan_lfs_full → publish:statcan_lfs_full, then a transform subflow building only source:statcan.lfs_full+ (check the dbt-node-results artifact), then quality.
4. A statcan.table.published event, and the release briefing artifact.
5. Trigger the watcher again with the same simulation while WAP is running: the idempotency key prevents a duplicate WAP run.

Then combine with Day 50's chaos: inject rate_out_of_range into raw_incoming (skip the write step via a skip_write=True parameter you add for drills), run WAP, and confirm the quarantine path: statcan.table.quarantined event, Failed or warning state per policy, alert sent, production table untouched.

Record the evidence in orchestration/ACCEPTANCE.md.

Day 59 checklist

• I can contrast schedule-driven and event-driven orchestration and justify a hybrid
• A readers–writer lock (occupy) prevents any reader–writer collision on DuckDB
• WAP runs as a flow with per-table write, audit, publish/quarantine tasks and events
• A release watcher polls during the 08:30 ET window and triggers WAP idempotently
• A new release rebuilds only its downstream dbt models, re-validates, and produces an economic briefing
• Simulated releases and simulated bad releases follow the documented paths

git add . && git commit -m "Day 59: readers-writer lock, WAP flow, release watcher, briefings" && git push