Dependency and Triggering

The previous lesson built the pipeline as a graph of forty-eight tasks and counted how many of them cannot run at all when a task fails. There was only one thing the graph did not say: who starts the chain. Thirty-six tasks are triggered by the completion of another task — the dependency trigger is the graph itself. The remaining twelve ingest tasks have no ancestor, and it is not the graph but a decision from outside that runs them.

This lesson divides that decision into three classes and measures them on the same pipeline. The measured number is the answer to a simple question: how many times and when did the same job run. A task running extra is not free, while a task not running at all is silent; the two flaws sit side by side in the same table.

• OO9. The source, the pipeline and the arrival plan are fictional; no real scheduler, workflow product or messaging system is called. The seed is 20260218, and the measurement does not depend on the run.
• OO10. A step is a period. No raw time is written; delay is counted in periods.
• OO11. The load of the source starts either in its own period or one step late, and arrives in one or two parts. Both pieces of information are derived from the seed, not chosen by hand.
• OO12. An empty run is a task that finishes without seeing any rows. A partial run is a task that sees only a part of the rows of the period.
• OO13. In this lesson the write is keyed: the second run of the same period does not duplicate rows. Duplication is the subject of the next lesson and is kept out so that it does not contaminate the numbers here.
• OO14. The baseline is the run without orchestration: a fixed calendar set by hand, no retry, no readiness criterion.
• OO15. The queue, topic and delivery guarantee procedure was measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum and is not repeated. The only question here is when the trigger starts the task.
• OO16. Resolution: periods 12, rows per period 200, healthy consumption 2200. No ratio is printed on serving; the value itself is written.

Three Trigger Classes

The time trigger is the cheapest: the task runs at a moment written in the calendar and never looks at the source. This is the written-down form of running the pipeline by hand, and it is the baseline of this lesson. Its contract is one-sided — the scheduler makes a promise, the source does not.

The event trigger hands the decision to the source: when the source drops a part of the load into the pipeline, it sends a notification, and the task runs on that notification. Because the source is the party that knows when it is ready, this looks more correct. The problem is what the notification announces: if a load arrives in two parts, two notifications are sent and the task runs twice.

The readiness trigger leaves the decision to the pipeline itself: the task waits until it verifies that the load of the period is complete. It produces no empty or partial run, but because it waits, it is delayed, and the criterion of readiness has to be written. When that criterion is written wrongly, the task never runs.

# FICTIONAL. The meter readings, the five-stage pipeline and the arrival plan are
# defined inside the lesson; no real scheduler, orchestrator or messaging system is called.
SEED, M32 = 20260218, 0xFFFFFFFF


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


REGION = ["east", "west", "north", "south", "central"]
TYPE = ["residential", "commercial"]
PERIODS = [f"2026-{a:02d}" for a in range(1, 13)]
SUBSCRIBERS = {}
for i in range(200):
    r = rng(SEED + 37 * i)
    SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)],
                                    "start": 10000 + int(r() * 60000)}


def generate_source():
    records = []
    for i, (sub, a) in enumerate(SUBSCRIBERS.items()):
        e = a["start"]
        for p in PERIODS:
            r = rng(SEED + 101 * i + 7 * PERIODS.index(p))
            e += 4 + int(r() * 46)
            records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"],
                            "index": e, "unit": "m3"})
    return records


def s1_screen(records):
    REQUIRED = ("subscriber", "period", "region", "index", "unit")
    keep = [k for k in records if all(a in k for a in REQUIRED)]
    return keep, len(records) - len(keep)


def s2_dedupe(records):
    seen, keep = set(), []
    for k in records:
        key = (k["subscriber"], k["period"])
        if key in seen:
            continue
        seen.add(key)
        keep.append(k)
    return keep, len(records) - len(keep)


def s3_consumption(records):
    table = {(k["subscriber"], k["period"]): k for k in records}
    out, dropped = [], 0
    for k in records:
        i = PERIODS.index(k["period"])
        if i == 0:
            continue
        prev = table.get((k["subscriber"], PERIODS[i - 1]))
        if prev is None:
            dropped += 1
            continue
        try:
            f = int(k["index"]) - int(prev["index"])
        except (TypeError, ValueError):
            dropped += 1
            continue
        out.append(dict(k, m3=f))
    return out, dropped


def s4_summary(records):
    totals = {}
    for k in records:
        key = (k["region"], k["period"])
        totals[key] = totals.get(key, 0) + k["m3"]
    return totals, 0


def s5_serve(summary):
    region = {}
    for (b, d), v in summary.items():
        region.setdefault(b, []).append(v)
    return {b: round(sum(v) / len(v), 2) for b, v in region.items()}, 0


def read_result(records):
    v, _ = s1_screen(records)
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    summ, _ = s4_summary(v)
    serving, _ = s5_serve(summ)
    return serving, len(v)

The Fictional Arrival Plan

What separates the triggers is the behavior of the source. If the source kept to the calendar, all three classes would give the same result; the distinction comes from the source arriving sometimes late and sometimes split. The second block derives this behavior from the seed: the load of each period starts at its own step or one step later, and arrives in one or two parts.

SOURCE = generate_source()
BASELINE_SERVING, BASELINE_N = read_result(SOURCE)
HORIZON = 13


def arrival_plan(late_share=0.25, split_share=0.35):
    """FICTIONAL: the load of each period arrives in one or two parts , some periods start one step late."""
    r = rng(SEED + 11)
    plan = {}
    for p in PERIODS:
        start = PERIODS.index(p) + (1 if r() < late_share else 0)
        parts = 2 if r() < split_share else 1
        plan[p] = {"start": start, "parts": parts, "done": start + parts - 1}
    return plan


PLAN = arrival_plan()


def visible(period, step):
    """Rows that have reached the pipeline by this step."""
    p, rows = PLAN[period], [k for k in SOURCE if k["period"] == period]
    if step < p["start"]:
        return []
    arrived = min(p["parts"], step - p["start"] + 1)
    return rows[:len(rows) * arrived // p["parts"]]


def schedule(mode):
    """(step , period) pairs: when and for which period the task runs."""
    if mode == "time":                                   # fixed calendar
        return [(t, PERIODS[t]) for t in range(12)]
    if mode == "time_window":                            # calendar , last two periods
        return sorted((t, PERIODS[g]) for t in range(HORIZON) for g in (t, t - 1) if 0 <= g < 12)
    if mode == "event":                                  # each part sends a notification
        return sorted((PLAN[p]["start"] + j, p) for p in PERIODS for j in range(PLAN[p]["parts"]))
    return [(PLAN[p]["done"], p) for p in PERIODS]       # readiness


def measure(mode):
    """The write is keyed: the second run of the same period does not duplicate rows."""
    store, empty, partial, counts, complete = {}, 0, 0, {}, {}
    for step, p in schedule(mode):
        v = visible(p, step)
        if not v:
            empty += 1
        elif len(v) < 200:
            partial += 1
        counts[p] = counts.get(p, 0) + 1
        for k in v:
            store[(k["subscriber"], k["period"])] = k
        if p not in complete and sum(1 for x in store if x[1] == p) == 200:
            complete[p] = step - PERIODS.index(p)
    serving, n = read_result(list(store.values()))
    return {"runs": len(schedule(mode)), "empty": empty, "partial": partial,
            "twice": sum(1 for p in counts if counts[p] > 1), "missing": len(PERIODS) - len(complete),
            "lag": sum(complete.values()), "consumption": n,
            "diff": {b: round(serving[b] - BASELINE_SERVING[b], 2) for b in BASELINE_SERVING}}


print("late-starting periods:", [p for p in PERIODS if PLAN[p]["start"] != PERIODS.index(p)])
print("periods arriving in two parts:", [p for p in PERIODS if PLAN[p]["parts"] == 2])
print("load completion lag (periods):", sum(PLAN[p]["done"] - PERIODS.index(p) for p in PERIODS))

late-starting periods: ['2026-02', '2026-10', '2026-12']
periods arriving in two parts: ['2026-01', '2026-02', '2026-06', '2026-09']
load completion lag (periods): 7

Three of the twelve periods start one step late, four are split into two parts, and a total lag of seven periods builds up. This is not a faulty source; collecting a part of the meter readings late and sending large loads in pieces are ordinary matters. What is to be measured is this: against the same source, how many tasks do the four trigger setups run, and what do they leave behind.

How Many Times and When the Same Job Ran

Four setups are compared. time looks at the calendar once. time + two periods adds a safety window, set by hand, to the same calendar: at each step it reprocesses both that period and the one before it. event runs on the notification of each load part. readiness waits until the two hundred rows of the period are complete.

print(f"{'trigger':<20}{'runs':>6}{'empty':>7}{'partial':>9}{'twice':>7}"
      f"{'missing':>9}{'lag':>9}{'consumption':>13}")
for mode, name in (("time", "time"), ("time_window", "time + two periods"),
                   ("event", "event"), ("readiness", "readiness")):
    o = measure(mode)
    print(f"{name:<20}{o['runs']:>6}{o['empty']:>7}{o['partial']:>9}{o['twice']:>7}"
          f"{o['missing']:>9}{o['lag']:>9}{o['consumption']:>13}")
print()
print("time deviation      :", measure("time")["diff"])
print("time + two periods  :", measure("time_window")["diff"])
print("event and readiness :", measure("readiness")["diff"])

trigger               runs  empty  partial  twice  missing      lag  consumption
time                    12      3        3      0        6        0          900
time + two periods      24      3        4     12        1        5         2000
event                   16      0        4      4        0        7         2200
readiness               12      0        0      0        0        7         2200

time deviation      : {'central': -317.28, 'north': -382.88, 'east': -175.69, 'west': -240.59, 'south': -251.72}
time + two periods  : {'central': -100.63, 'north': -145.19, 'east': -65.54, 'west': -86.82, 'south': -80.46}
event and readiness : {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0}

The first row is the baseline and the most misleading row in the table. The fixed calendar writes zero in the lag column; no period was processed late, because no period was waited for. In the same row six periods remain missing: three of the twelve runs finish empty, three partial. Consumption comes out as 900 rows instead of 2200, and serving collapses in all five regions (between −175.69 and −382.88). Zero lag is zero not because it measured waiting, but because there was no waiting at all.

The second row is the answer given in the field to this flaw: adding a safety window without leaving the calendar. The number of runs goes from 12 to 24, and all twelve periods are processed a second time — that is, half of the pipeline is unnecessary work. In return, the missing periods fall from six to one and consumption rises to 2000. But the one remaining period keeps making the pipeline pay: 2026-02 both starts one step late and is split into two parts, completes at the third step, and the two-period window no longer looks at it at the third step. Serving is still wrong in every region (between −65.54 and −145.19). Widening the window is choosing a boundary, and every boundary that is chosen leaves some arrival plan outside.

The third row shows both the strength and the flaw of the event trigger. The missing periods are zero, the serving deviation is empty, the result is complete. On the other hand, 16 runs were made: the four periods that arrive in two parts were processed twice, and the first run of those four is a partial run. Because this lesson keeps the write keyed, the partial run does no harm; the next lesson will measure what the same partial run does when the write is unkeyed.

The fourth row is the setup that is looked for, and it states plainly what is bought. The readiness trigger gives the complete result with 12 runs: no empty run, no partial run, no period processed a second time. The cost it pays is the seven in the lag column — because it waits for the load to complete, it has finished a total of seven periods late. This number is exactly the same as the lag of the source, that is, the trigger adds no lag of its own; it only makes the lag of the source visible.

The Flaw of the Readiness Criterion

The cost of the readiness trigger is not only lag. From now on a criterion determines when the task runs, and that criterion is written by hand. In the setup above the criterion was of the form “two hundred rows arrived”; the number two hundred is the current subscriber count of the source and is an assumption.

# Starting with the ninth period, four subscribers ended their subscriptions: the source sends 196 rows.
LEAVING = {"A197", "A198", "A199", "A200"}
SOURCE2 = [k for k in SOURCE if not (k["period"] >= "2026-09" and k["subscriber"] in LEAVING)]
COUNT = {p: sum(1 for k in SOURCE2 if k["period"] == p) for p in PERIODS}
print("rows per period:", sorted(set(COUNT.values())), " total:", len(SOURCE2))
print(f"{'readiness criterion':<28}{'periods run':>12}{'rows':>7}{'consumption':>13}")
results = {}
for name, threshold in (("source-reported parts", 0), ("fixed 200 rows", 200)):
    running = [p for p in PERIODS if COUNT[p] >= threshold]
    accepted = [k for k in SOURCE2 if k["period"] in running]
    serving, n = read_result(accepted)
    results[name] = serving
    print(f"{name:<28}{len(running):>12}{len(accepted):>7}{n:>13}")
a, b = results["source-reported parts"], results["fixed 200 rows"]
print("deviation of fixed threshold:", {x: round(b[x] - a[x], 2) for x in a})

rows per period: [196, 200]  total: 2384
readiness criterion          periods run   rows  consumption
source-reported parts                 12   2384         2184
fixed 200 rows                         8   1600         1400
deviation of fixed threshold: {'central': -37.59, 'north': -10.75, 'east': -18.5, 'west': 8.11, 'south': -13.98}

Because four subscribers ended their subscriptions, the source sends 196 rows after the ninth period. This is not a corruption; the source sends the correct data completely. The readiness criterion tied to a fixed threshold, however, waits for the second hundredth row, so it never triggers in four periods: the periods that run are 8 instead of 12, consumption is 1400 instead of 2184. No task raises an error, no log line is written, and the state of the pipeline stays waiting.

The deviation row says how this flaw looks in serving: while central falls by −37.59, west rises by +8.11. Because the lost periods are not distributed equally among the regions, even the sign of the deviation is not consistent; that is, it is impossible to conclude “four periods are missing” by looking at serving. The right criterion is not a fixed number but the completeness of parts reported by the source itself; when it is used, all twelve periods run and consumption stays at 2184 rows.

Summary

• The thirty-six tasks in the graph are triggered by dependency; a decision from outside starts the twelve ingest tasks, and this decision is a time, event or readiness trigger.
• The fixed calendar, the baseline, writes zero in the lag column but leaves six periods missing: 3 of 12 runs finish empty, 3 partial, and consumption is 900 rows instead of 2200.
• Adding a safety window of two periods to the calendar raises the runs to 24 and processes all twelve periods a second time; the missing periods fall to one, but serving is still wrong in every region.
• The event trigger gives the complete result; its cost is 16 runs and four periods processed a second time; the first run of the four periods that arrive in two parts is a partial run.
• The readiness trigger gives the complete result with 12 runs and without empty or partial runs; its cost is the total lag of seven periods, and this lag is exactly the same as that of the source.
• When the readiness criterion is tied to a fixed row count, four periods never trigger: consumption falls from 2184 to 1400, the sign of the deviation changes with the region, and no task raises an error.

Next Step

All the runs in this lesson had one thing in common: none of them failed. The task either ran or was never triggered; when it ran, it wrote all of the rows it saw. In a real pipeline there is a third outcome — the task runs, starts writing and fails in the middle. The partial run was harmless in this lesson, because the write was keyed and the missing part was completed in the next run. The next lesson removes that guarantee and measures two questions: what a partially written partition does to serving, and whether retrying the failed task repairs the pipeline or duplicates rows.