Resilience to Repetition and Reprocessing

The previous lesson showed that the contract catches the corruption at stage 0, but catching is not repairing: the 200 blocked rows wait somewhere, the producer fixes them, and that period is resent to the pipeline. In none of the measurements so far has the pipeline seen the same data twice.

The second run is not an exception; it is the most frequent thing a pipeline does. A producer corrects a load, a stage fails and is retried, a past period is backfilled. This lesson counts what happens to the row count of a pipeline that processes the same load twice, which stage absorbs the surplus, and whether the served value shows it. The answer is the cleanest example of the third claim of the course.

• QP14. The source, the pipeline and the store are fictional; no real store, file format or orchestrator is called. The seed is 20260218, and the measurement does not depend on the run.
• QP15. The equivalent of idempotent in this curriculum is idempotent. The concept itself was measured in the Shell Programming course, and the retry and queue procedure in the Caching, Queues and Asynchronous Processing course, and is not repeated. The only question here is whether reprocessing duplicates the row.
• QP16. Reprocessing is not an accident: a corrected load, a retried stage and a backfilled period are ordinary jobs.
• QP17. The value of a resent row stays the same, only the load stamp changes; the one exception is the load that carries a correction. This detail determines the whole of the last two measurements.
• QP18. The baseline is the single run: raw rows 2400, duplicated rows 0.
• QP19. The units counted: reprocessed rows, duplicated rows, rows dropped by deduplication and the difference at serving. No raw time is written, the unit is written as run.
• QP20. Resolution: the step of the row share is 0.0004, of a summary cell 0.0182, of the serving value 0.2000. No share is printed on the serving output.

Two Write Modes

Whether a pipeline is idempotent is decided by a single choice: how the incoming row is written to the store. An appending write puts the row at the end of the store; when the same row arrives twice, it sits in the store twice. A keyed write overwrites the row at a key; when the same key arrives twice, it sits in the store once. The key here is the pair of subscriber and period, which is already what the pipeline uses as its uniqueness criterion.

The second part is how reprocessing looks. When a corrected load arrives, the value of the row is the same; the only thing that changes is the load stamp. This detail is often overlooked, and the whole of the last measurement rests on it.

# FICTIONAL. Meter readings, the five-stage pipeline and the store are defined in the lesson;
# no real store, file format or orchestrator 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


# ---- PIPELINE: five stages, each counts the rows it drops
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


STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption),
          ("summary", s4_summary), ("serve", s5_serve)]


def pipeline(records, stages=None):
    trace, data = [], records
    for name, f in (stages or STAGES):
        data, dropped = f(data)
        trace.append({"stage": name, "remaining": len(data), "dropped": dropped})
    return data, trace


# ---- INGESTION and REPROCESSING
def ingest(records, load):
    """Ingestion stage: each load is given a stamp."""
    return [dict(k, load=load) for k in records]


def resend(records, periods, load):
    """The corrected load is resent ; the stamp changes, the value does not."""
    return [dict(k, load=load) for k in records if k["period"] in periods]


def store_append(store, new):
    """NON-idempotent write: the incoming row is appended to the end of the store."""
    return store + new


def store_keyed(store, new):
    """IDEMPOTENT write: the (subscriber , period) key is overwritten."""
    table = {(k["subscriber"], k["period"]): k for k in store}
    for k in new:
        table[(k["subscriber"], k["period"])] = k
    return list(table.values())


def s2_dedupe_full(records):
    """FLAWED DEDUPLICATION: looks at the whole row, not at the key."""
    seen, keep = set(), []
    for k in records:
        stamp = tuple(sorted((a, str(v)) for a, v in k.items()))
        if stamp in seen:
            continue
        seen.add(stamp)
        keep.append(k)
    return keep, len(records) - len(keep)


FLAWED = [("screen", s1_screen), ("dedupe", s2_dedupe_full), ("consumption", s3_consumption),
          ("summary", s4_summary), ("serve", s5_serve)]


def c_uniqueness(data):
    return len(data) != len(set((k["subscriber"], k["period"]) for k in data))


def c_row_count(data, expected=2400, tolerance=0.05):
    return abs(len(data) - expected) / expected > tolerance

Duplicated Rows and an Unchanged Serving

The second block applies five setups to the same source. The raw column gives the row count in the store, dedupe the rows dropped by that stage, summary the cell count of the fourth stage, and north one of the five served values.

SOURCE_RECORDS = generate_source()
FIRST = ingest(SOURCE_RECORDS, 1)
BASELINE, _ = pipeline(FIRST)


def report(name, store, stages=None):
    serving, trace = pipeline(store, stages)
    print(f"{name:<30}{len(store):>6}{trace[1]['dropped']:>9}{trace[3]['remaining']:>9}"
          f"{serving['north']:>11.2f}{'same' if serving == BASELINE else 'changed':>9}")


one_period = resend(SOURCE_RECORDS, {"2026-07"}, 2)
full = resend(SOURCE_RECORDS, set(PERIODS), 2)
print("baseline serving:", BASELINE)
print("resent rows: one period", len(one_period), " full run", len(full))
print()
print(f"{'setup':<30}{'raw':>6}{'dedupe':>9}{'summary':>9}{'north':>11}{'serving':>9}")
report("single run", FIRST)
report("appending store , one period", store_append(FIRST, one_period))
report("appending store , full run", store_append(FIRST, full))
report("keyed store , full run", store_keyed(FIRST, full))
report("flawed deduplication , full", store_append(FIRST, full), FLAWED)
print()
d = store_append(FIRST, full)
print("uniqueness check:", c_uniqueness(d), " row count check:", c_row_count(d))

baseline serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55}
resent rows: one period 200  full run 2400

setup                            raw   dedupe  summary      north  serving
single run                      2400        0       55    1212.55     same
appending store , one period    2600      200       55    1212.55     same
appending store , full run      4800     2400       55    1212.55     same
keyed store , full run          2400        0       55    1212.55     same
flawed deduplication , full     4800        0       55    2425.09  changed

uniqueness check: True  row count check: True

The second row is the headline of the lesson: when one period is backfilled, the appending store carries 2600 rows, which means 200 rows were duplicated. The deduplication stage drops exactly those 200 rows, and the served value stays at 1212.55. The third row shows the same thing at a larger scale: a full rerun takes the store to 4800 rows, deduplication drops 2400 rows, and the serving is again the same.

This is the cleanest example of the third claim of the course. The size of the store has doubled, the pipeline has done the same work twice, yet none of the five served values has moved. An observation that looks at the output concludes that the pipeline is idempotent. But what is idempotent is not the pipeline, it is a single stage inside the pipeline: deduplication absorbs the surplus and tells nobody while doing so.

The fourth row shows what real idempotence is. After the second run the keyed store still carries 2400 rows: duplicated rows 0, rows dropped by deduplication 0. The difference is not in the store but beyond the pipeline; while producing the right number, the appending store takes up twice the space and cannot be used without a deduplication stage.

The last two rows are what the checks say. The uniqueness check fires on the appending store, and the row count check fires too — 4800 is far outside the five percent band around the expected 2400. So the detection lag of the corruption is 0, and the check that catches it is not a check that looks at the serving.

The Absorbed Correction

So far the value of the resent row was the same as the first one’s, so which one deduplication kept did not matter. In real backfills the situation is the reverse: the load is resent because it was corrected. The deduplication stage keeps the row it sees first, that is, the old one. In the setup below, the producer read the last period short, then corrected it and resent it. The corruption is deliberately placed in the last period: a shift in the middle of the series returns with the opposite sign in the following period and the two cancel out in the average.

def short_reading(records, period="2026-12", diff=100):
    """The producer read the last period short ; it will then correct it and resend."""
    return [dict(k, index=k["index"] - diff) if k["period"] == period else k
            for k in records]


SHORT = ingest(short_reading(SOURCE_RECORDS), 1)
CORRECTION = resend(SOURCE_RECORDS, {"2026-12"}, 2)
print("corrected rows:", len(CORRECTION))
print(f"{'setup':<32}{'raw':>6}{'dedupe':>9}{'north':>10}{'baseline':>10}")
for name, store in (("short load , no correction", SHORT),
                    ("appending store , correction", store_append(SHORT, CORRECTION)),
                    ("keyed store , correction", store_keyed(SHORT, CORRECTION))):
    serving, trace = pipeline(store)
    print(f"{name:<32}{len(store):>6}{trace[1]['dropped']:>9}{serving['north']:>10.2f}"
          f"{BASELINE['north']:>10.2f}")

corrected rows: 200
setup                              raw   dedupe     north  baseline
short load , no correction        2400        0    794.36   1212.55
appending store , correction      2600      200    794.36   1212.55
keyed store , correction          2400        0   1212.55   1212.55

The second row is the most expensive finding of the lesson. The correction arrived at the pipeline, 200 rows were reprocessed, the store went up to 2600 rows, deduplication dropped 200 rows — and the served value stayed at 794.36, that is, exactly the same as before the correction. The 200 rows that were dropped were the corrected ones. The deduplication stage absorbed the second run; here, absorbing is losing.

This is the exact opposite reading of the previous section, and both come from the same behavior. When the resent row is the same as the first, deduplication saves the pipeline; when it is corrected, the same deduplication throws the correction away. The stage cannot tell the two apart, because all it has is the key and the key of the two rows is the same. The third row shows the solution: because the keyed store overwrites the old row with the new one, the serving returns to 1212.55, that is, to the baseline.

For a pipeline to count as idempotent, it is not enough that processing the same load twice leaves the result unchanged; processing a corrected load a second time must also correct the result. The appending store meets the first condition and does not meet the second.

When the Key Is Chosen Wrongly

The last row of the table opens up the most fragile assumption of the pipeline. The deduplication stage looks at the (subscriber, period) key. Had the same stage been written to look at the whole row, the resent row would look different because of the load stamp and would not be eliminated. This is not a made-up flaw: asking whether a load is entirely the same seems intuitively safer, and that is exactly why it gets written.

appending, keyed = FIRST, FIRST
print(f"{'run':<6}{'appending raw':>15}{'north':>9}{'flawed north':>15}"
      f"{'keyed raw':>18}{'north':>9}")
for run in range(1, 4):
    if run > 1:
        new = resend(SOURCE_RECORDS, set(PERIODS), run)
        appending = store_append(appending, new)
        keyed = store_keyed(keyed, new)
    serve_append, _ = pipeline(appending)
    serve_flawed, _ = pipeline(appending, FLAWED)
    serve_keyed, _ = pipeline(keyed)
    print(f"{run:<6}{len(appending):>15}{serve_append['north']:>9.2f}"
          f"{serve_flawed['north']:>15.2f}{len(keyed):>18}"
          f"{serve_keyed['north']:>9.2f}")
print()
summ, _ = s4_summary(s3_consumption(s2_dedupe_full(s1_screen(appending)[0])[0])[0])
summ_base, _ = s4_summary(s3_consumption(s2_dedupe(s1_screen(FIRST)[0])[0])[0])
print("summary ('north', '2026-07') baseline:", summ_base[("north", "2026-07")],
      " flawed deduplication:", summ[("north", "2026-07")])
print("keys kept by deduplication:", len(set((k["subscriber"], k["period"]) for k in FIRST)))

run     appending raw    north   flawed north         keyed raw    north
1                2400  1212.55        1212.55              2400  1212.55
2                4800  1212.55        2425.09              2400  1212.55
3                7200  1212.55        3637.64              2400  1212.55

summary ('north', '2026-07') baseline: 1251  flawed deduplication: 3753
keys kept by deduplication: 2400

The three-run table sets two sentences side by side. In the pipeline that works with the right key, the raw store goes from 2400 to 4800 to 7200 and the served value is 1212.55 in all three runs. In the pipeline that works with the flawed key, the same raw store and the same reruns make the served value grow as 1212.55 → 2425.09 → 3637.64: in direct proportion to the number of runs. The same shows in the summary layer: the 2026-07 cell of the north is 1251 at the baseline and 3753 after the third run.

The only difference between the two pipelines is the key choice of one stage. No corruption class, no field name changed, no type drifted; the only thing that happened on the producer side is that the same correct load was sent a second time. This is the form that the second claim of the course takes in this lesson: a pipeline’s own stage, with a flawed key, can triple the served number without any source corruption.

The column of the keyed store says where the cost lies. Idempotence is not free: at write time a key table of 2400 entries is kept and every row is looked up before it is written. The deduplication stage carries a set of the same size. In return, the store stays at 2400 rows regardless of the number of runs, and the correctness of the pipeline no longer depends on the key of a single stage.

Summary

• Reprocessing is an ordinary job; in an appending store, backfilling one period takes the raw rows from 2400 to 2600, a full rerun to 4800, and the third run to 7200.
• None of the five served values shows this growth: the north is 1212.55 in all three runs, because the deduplication stage absorbs the surplus and does not say so while absorbing.
• What is idempotent is not the pipeline but a single stage inside it; when keyed writing is used, duplicated rows fall to 0 and the store stays independent of the number of runs.
• When the resent load carries a correction, the same deduplication throws it away: the served value stays at 794.36, whereas the keyed store returns to 1212.55 with the same correction. Idempotence is not that the second run leaves the result unchanged, but that it corrects when it should correct.
• When deduplication looks at the whole row instead of (subscriber, period), the same reruns carry the served value from 1212.55 to 2425.09 and then to 3637.64; the summary cell goes from 1251 to 3753.
• The checks that catch duplication are uniqueness and row count, and both fire at stage 0; no check that looks at the serving can see it.

Next Step

In three lessons the check always looked at the same place: the row itself. Does it have its fields, is its type right, is its key unique, is its value in line with the previous periods. These questions give no answer when a serving value is corrupted, because none of them says where the row came from. Someone looking at the figure 3753 produced by the flawed deduplication has to read the whole pipeline from the beginning to find which loads contributed to that cell and at which link of the chain the error is. The next lesson counts this search: how many stages and how many rows one has to look at to go from a corrupted serving value back to the source, with a record kept and without one.