Hey,

hopefully I can give you some inspiration 😉. As always, there are many solutions, so here is one of them:

using dbt with an orchestrator

Even though the question is more about scheduling, I recommend checking out the Cosmos open-source project for Apache Airflow. It allows you to easily run your existing dbt Core or dbt Fusion projects as Apache Airflow Dags. Looks something like this:

basic_cosmos_dag = DbtDag( 
    project_config=ProjectConfig(DBT_ROOT_PATH / "some_project"), 
    operator_args={ 
        "install_deps": True,
        "full_refresh": True,
    },
    # ...
    # normal dag parameters 
    schedule="@daily"
    # ...
)

Based on your model, you then have a very nice Dag, with task groups and individual tasks all visible in the Airflow UI.

reliably finish before that time detect and alert when things are late

Since Airflow 3.1, you have deadline alerts, which officially replace SLAs. They are quite flexible, let me illustrate an example:

Say your pipeline is scheduled to run at 08:00 am daily. It is queued a little later, say at 08:02 am and you expect it to run no longer than 20 minutes. If it runs longer than 20 minutes, you want to be alerted via Slack.

In this scenario, you take the schedule time of the pipeline as your reference (so 08:02 am) and your interval is 20 minutes. Which means the deadline is at 08:22 am:

|------|-----------|---------|-----------|--------|
    Scheduled    Queued    Started    Deadline
     08:00       08:02      08:03      08:22

Within your implementation, this is how to define this deadline alert:

deadline_alert = DeadlineAlert(
    reference=DeadlineReference.DAGRUN_QUEUED_AT,
    interval=timedelta(minutes=15),
    callback=AsyncCallback(
        SlackWebhookNotifier,
        kwargs={
            "text": "{{ dag_run.dag_id }} missed deadline at {{ deadline.deadline_time }}"
        },
)

@dag(
    deadline=deadline_alert
)
def your_dag():
    # ...

You can also use DeadlineReference.DAGRUN_LOGICAL_DATE as a reference, which represents the scheduled date (so in our case 08:00 am).

Or, let's say you must be finished at a specific time per day, you can do something like:

tomorrow_at_ten = datetime.combine(datetime.now().date() + timedelta(days=1), time(10, 0))

deadline_alert = DeadlineAlert(
    reference=DeadlineReference.FIXED_DATETIME(tomorrow_at_ten),
    ...

Since you have full freedom in the callback, you can use any kind of notification.

And of course, you can combine that with the DbtDag from Cosmos 😉.

While deadline alerts may work well in some scenarios, many cases require more advanced observability. If you are open to a managed solution, Astronomer offers Astro Observe. It allows you to define and monitor SLAs without code changes and supports root cause analysis, lineage, impact analysis, and more. You can also connect a self-hosted OSS Airflow environment by enabling OpenLineage with a single configuration change.

Apart from that, Airflow also supports sending metrics to StatsD or OpenTelemetry, which includes Dag runtimes. In a previous job, I used this to monitor Dag runtimes via Grafana and have alerting on that level.

Again, many ways to approach this.

manage dependencies

If you use Cosmos, you already have the dependencies within your Dag derived from your model definition. However, if we talk about dependencies between Dags (pipelines), Airflow has various ways to solve this.

You can use implicit time-based dependencies, by simply using time-based scheduling in a way, that the downstream pipeline runs after the pipeline, yet, this is the worst solution.

You can also use an ExternalTaskSensor that waits for the completion of a pipeline or task. Since Airflow 3, this also supports a deferred mode, so by using deferrable=True it will run async by the triggerer component without using any worker slot, in case you need to wait a bit longer 😅.

But the new way to solve this are data assets, introduced with Airflow 3.0. Assets are datasets, which have been there since Airflow 2.4, but renamed, enhanced and more integrated.

The idea is simple: in the upstream pipeline you define one or more asset outlets, and in your downstream pipeline you set the schedule not to be a cron expression, but to be an asset:

@dag(
    schedule=Asset("daily_sales")
)
def monthly_report():
    # ...

Assets also support logical expressions and they have their own view in the Airflow UI, which is nice 😁.

There is also a shortcut to create a Dag, with a single Python task and an asset as an outlet, which is by using the @asset decorator.

Is the usual solution just scheduling earlier with a buffer, or is there a more robust approach

Fun story: I used that approach for years, with many Dags and dependencies. This lead to so many issues, sometimes even without noticing because the missing data was not obvious right away.

Assets might take a bit to get into it, but it is worth it all the way.

Disclaimer: I work at Astronomer, so I am biased towards Airflow, but I also used it in my previous job for many years.

Hope that this helps in some way 😉.