Function workflow_test

test/canary-environment/mzcompose.py:636–707

(c: Composition, parser: WorkflowArgumentParser)

def workflow_test(c: Composition, parser: WorkflowArgumentParser) -> None:
    # These replace the former `dbt test` data tests: every source is running,
    # every replica is online, compute frontiers (and their imports) keep up,
    # and every monitored object's write frontier keeps advancing (i.e. it is
    # still sealing new output times — making progress). Unlike the dbt tests
    # (which only checked the result was empty), each check reports the specific
    # failing objects.
    #
    # The checks are independent (each opens its own connection), so they run
    # concurrently and report as each finishes via run_parallel — the same
    # spinner/✓/✗ output as bin/lint.
    #
    # Progress is measured by a write-frontier advance, NOT by a
    # SUBSCRIBE(SNAPSHOT=false) emitting a row. A subscribe is the wrong probe
    # for the big UPSERT source-tables: it emits nothing for a live source whose
    # content happens not to change (customer re-sends no-op upserts), and it can
    # take minutes just to hydrate a fresh dataflow over a ~1B-row persist shard
    # — both false timeouts. A frozen write frontier, by contrast, is exactly
    # "not making progress", and the sample is one cheap catalog query.
    parser.add_argument(
        "--progress-window",
        default=15.0,
        type=float,
        help="Seconds to watch each object's write frontier; it must advance within this window.",
    )
    args = parser.parse_args()

    # `stop` lets the frontier checks bail out promptly on Ctrl-C: signals go to
    # the main thread, so the workers can't see the KeyboardInterrupt — they poll
    # this event while waiting out the sample window.
    stop = threading.Event()

    def task(fn: Callable[[], list[str]]) -> Callable[[], tuple[bool, str]]:
        # Adapt a check (returns the list of failures) to run_parallel's
        # (success, output) contract; the output is shown only on failure.
        def run() -> tuple[bool, str]:
            fails = fn()
            return not fails, "\n".join(fails)

        return run

    checks: list[tuple[str, TaskSpec]] = [
        ("RDS Postgres replication-slot count", task(_check_rds_slots)),
        ("all canary sources running", task(_check_sources_running)),
        ("all canary cluster replicas online", task(_check_replicas_online)),
        ("compute frontiers within 1 minute of now", task(_check_frontiers)),
    ]
    for obj in MONITORED_MVS + MONITORED_SOURCE_TABLES:
        # Label with just the object name (the db/schema is always
        # qa_canary_environment.public_*); the check uses the FQN.
        checks.append(
            (
                f"{obj.split('.')[-1]} makes progress",
                task(
                    lambda obj=obj: _frontier_advances(obj, args.progress_window, stop)
                ),
            )
        )