--- title: 'Delivery Guarantees' source: 'https://academia.sh/en/courses/data-pipelines/delivery-guarantees' course: 'Data Pipelines and Orchestration' language: en updated: '2026-10-21T05:02:58+00:00' license: 'CC BY-SA 4.0' objectives: - 'Runs the same pipeline in three delivery modes and counts the lost rows, the duplicated rows, and the consumption rows separately.' - 'Compares what loss and duplication do to the five region values served.' - 'Reads which check fires in each mode and shows with threshold numbers why the loss stayed silent.' - 'Turns off the dedupe stage to measure where the duplication is repaired and what that costs.' --- # Delivery Guarantees Turning the three delivery modes into numbers on the same pipeline: with at-most-once delivery 19 rows of period 2026-07 are lost, consumption falls from 2200 to 2162, 10 of the 55 summary cells are polluted, and the serving shifts in every region (−9.18 … −28.36); with at-least-once delivery 19 rows arrive twice but consumption stays at 2200 and the serving difference comes out empty, because the dedupe stage swallows the duplication; with exactly-once delivery the keyed write zeroes out both. The real finding is in the check column: the uniqueness check catches the duplication at stage 0, while the loss slips past every check — the row ratio of 0.0079 stays below the 0.05 threshold and the largest cell deviation of 0.2488 stays below the 0.25 threshold. Loss is silent, duplication is repairable. The previous lesson measured one form of loss, but there the record had **reached** the pipeline: it had only reached a window that had closed late, and it was dropped by a decision. This lesson removes that assumption: a row is sent and **never arrives**, or it is sent once and **arrives twice**. These two events are the two unavoidable ends of transport. When the sender gets no acknowledgement it either gives up or tries again; if it gives up the result is loss, if it tries again the result is duplication. The promise that transport makes on this point is called a **delivery guarantee**: **at-most-once** is the mode in which a row can be lost, **at-least-once** the mode in which it can arrive twice, and **exactly-once** the mode in which neither happens. The third often sits not in transport but in the **write**: the row is written with a key, and when the same key comes a second time the write is not made. The lesson runs the three modes on the same pipeline and asks: which one corrupts the served number, and which one is caught by a check. - **DI49.** The code is **fictional**; no messaging system, stream processor, or orchestrator is called. The source is the fictional meter readings of the Data Engineering and MLOps curriculum: 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DI50. The procedure of delivery semantics is not repeated**; it was measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum. The only thing added is what the mode does to **the pipeline's serving**. - **DI51.** Loss and duplication are modeled in a single period, 2026-07, with a share of 0.10 per row. The generator is seeded; 19 of the 200 rows of that period are affected. - **DI52.** The delivery mode is a property of **transport**; the five stages of the pipeline are the same in all three modes. - **DI53. Exactly-once** is modeled here not as a transport property but as a **keyed write**: if the (subscriber, period) pair has been written before, the second write is not made at all. - **DI54.** The check set is taken from the common definition; the row count tolerance is 0.05 and the distribution tolerance is 0.25. - **DI55.** The dedupe stage of the pipeline is **an option** in this lesson; where the duplication goes when it is turned off is measured. - **DI56.** The units counted are rows, lost rows, duplicated rows, polluted cells, and the serving deviation. The step of the row ratio is 1/2400, that is 0.0004; no ratio is printed on the serving. ## Three Modes, One Pipeline The block below sets up the common definition and adds the `deliver` function and the common check set to it. `deliver` touches only period 2026-07: it drops a row, adds a copy of it, or does not write a row whose key has been seen at all. ```python # FICTIONAL. No real messaging system , stream processor or orchestrator is called; # the meter readings of the measurement network are fictional , seed 20260218. 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(): # one reading per subscriber per period 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): # consumption from the period difference 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 out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) 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): # per-region mean over the periods 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 deliver(records, mode, period="2026-07", share=0.10): """At most once: loss. At least once: duplication. Exactly once: keyed write.""" r = rng(SEED + 7) out, lost, duplicates = [], 0, 0 seen = set() for k in records: if k["period"] != period: out.append(k) continue if mode == "at_most_once": if r() < share: lost += 1 continue out.append(k) elif mode == "at_least_once": out.append(k) if r() < share: out.append(dict(k)) duplicates += 1 else: # exactly_once: keyed write key = (k["subscriber"], k["period"]) if r() < share and key in seen: duplicates += 0 continue seen.add(key) out.append(k) return out, lost, duplicates def c_schema(data, fields=("subscriber", "period", "region", "index", "unit")): return any(any(a not in k for a in fields) for k in data) def c_uniqueness(data): key = [(k["subscriber"], k["period"]) for k in data] return len(key) != len(set(key)) def c_row_count(data, expected, tolerance=0.05): return abs(len(data) - expected) / expected > tolerance def c_partition_completeness_period(data): seen = set((k["region"], k["period"]) for k in data) return seen != set((b, d) for b in REGION for d in PERIODS) def c_distribution(summary, baseline, tolerance=0.25): for key, v in summary.items(): t = baseline.get(key) if t and abs(v - t) / max(abs(t), 1) > tolerance: return True return False raw = generate_source() def run(records, dedupe=True): v, _ = s1_screen(records) if dedupe: v, _ = s2_dedupe(v) v, _ = s3_consumption(v) summ, _ = s4_summary(v) serving, _ = s5_serve(summ) return len(v), summ, serving N, BASELINE, BASE_SERVING = run(raw) def dirty(summ): return sum(1 for name in set(summ) | set(BASELINE) if summ.get(name) != BASELINE.get(name)) def diff(serving): d = {b: round(serving[b] - BASE_SERVING[b], 2) for b in BASE_SERVING if round(serving.get(b, 0) - BASE_SERVING[b], 2)} return d if d else "empty" CHECKS = [("schema", lambda c, o: c_schema(c)), ("uniqueness", lambda c, o: c_uniqueness(c)), ("row", lambda c, o: c_row_count(c, 2400)), ("partition_period", lambda c, o: c_partition_completeness_period(c)), ("distribution", lambda c, o: c_distribution(o, BASELINE))] print("baseline: rows 2400 , consumption", N, ", summary cells", len(BASELINE)) print("baseline serving", BASE_SERVING) print() print("mode rows lost duplicates consumption dirty/55 firing check") for mode in ("at_most_once", "at_least_once", "exactly_once"): c, lost, dup = deliver(raw, mode) n, summ, serving = run(c) fired = [name for name, f in CHECKS if f(c, summ)] or ["none"] print(f"{mode:<13} {len(c):>5} {lost:>5} {dup:>10} {n:>11} {dirty(summ):>9} {','.join(fired)}") print(" serving diff", diff(serving)) print() print("keys held by the keyed write: per period", len({(k["subscriber"], k["period"]) for k in raw if k["period"] == "2026-07"}), ", for the whole run", len({(k["subscriber"], k["period"]) for k in raw})) ``` ``` baseline: rows 2400 , consumption 2200 , summary cells 55 baseline serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} mode rows lost duplicates consumption dirty/55 firing check at_most_once 2381 19 0 2162 10 none serving diff {'central': -9.18, 'north': -22.64, 'east': -9.81, 'west': -28.36, 'south': -20.28} at_least_once 2419 0 19 2200 0 uniqueness serving diff empty exactly_once 2400 0 0 2200 0 none serving diff empty keys held by the keyed write: per period 200 , for the whole run 2400 ``` ## The Asymmetry of Loss and Duplication Two rows of the table are mirror images of each other: one removes 19 rows, the other multiplies 19 rows. Their results are nothing alike. In **at-most-once** delivery, 19 rows belonging to period 2026-07 never arrive. The consumption rows fall from 2200 to **2162**, that is the loss costs **38** rows: a missing reading lowers the consumption of its own period, and because it removes the previous index it also makes the consumption of the following period impossible to compute. **10** of the 55 summary cells are polluted and **all** five region values served shift: −28.36 in the west, −9.18 in the center. In the previous lesson exactly 10 cells were polluted too, but the serving did not move at all; the difference is that there what fell short in July was in excess in August, while here **both periods fall short**. In **at-least-once** delivery the same 19 rows arrive twice. The dedupe stage of the pipeline drops the second copy; consumption **stays at 2200**, the polluted cells are **0**, and the serving difference is **empty**. In **exactly-once** mode the keyed write does not make the second write of the same key at all. Since every (subscriber, period) pair is in the source once, the output stays at 2400 rows and there is no duplication to count — the duplication counter of the mode is therefore zero. This mode is a property not of transport but of the **write**; transport can still deliver twice, the write makes it invisible. That it is not free is written on the last line. For the write to recognize the same key a second time, it has to **remember** it: **200** keys for one period, **2400** for the whole run. Exactly-once is not a transport assurance but an assurance **bought by keeping state**; for a key that is not kept there is no guarantee. ## Why the Loss Is Silent The real finding is in the last column. The mode that shifts the five values served in every region is **caught by no check**; the mode that changes no value fires the **uniqueness check**. The block below writes under which thresholds the loss slips through. ```python c, lost, _ = deliver(raw, "at_most_once") n, summ, serving = run(c) print("lost", lost, "rows ; row ratio", round(abs(len(c) - 2400) / 2400, 4), "< threshold 0.05") print("largest summary cell deviation", round(max(abs(summ[a] - BASELINE[a]) / abs(BASELINE[a]) for a in BASELINE if a in summ), 4), "< threshold 0.25") print("region set complete:", not c_partition_completeness_period(c)) print("same check at tolerance 0.005:", c_row_count(c, 2400, tolerance=0.005)) print() print("loss share lost consumption largest cell deviation row ratio firing") for p in (0.02, 0.05, 0.10, 0.20, 1.00): c, lost, _ = deliver(raw, "at_most_once", share=p) n, summ, serving = run(c) mx = round(max(abs(summ[a] - BASELINE[a]) / abs(BASELINE[a]) for a in BASELINE if a in summ), 4) fired = [name for name, f in CHECKS if f(c, summ)] or ["none"] print(f"{p:<11} {lost:>5} {n:>12} {mx:>23} {round(abs(len(c) - 2400) / 2400, 4):>10}" f" {','.join(fired)}") print() print("dedupe stage off:") print("mode consumption dirty/55 serving diff") for mode in ("at_most_once", "at_least_once", "exactly_once"): c, _, _ = deliver(raw, mode) n, summ, serving = run(c, dedupe=False) print(f"{mode:<13} {n:>11} {dirty(summ):>9} {diff(serving)}") ``` ``` lost 19 rows ; row ratio 0.0079 < threshold 0.05 largest summary cell deviation 0.2488 < threshold 0.25 region set complete: True same check at tolerance 0.005: True loss share lost consumption largest cell deviation row ratio firing 0.02 4 2192 0.1441 0.0017 none 0.05 10 2180 0.2094 0.0042 none 0.1 19 2162 0.2488 0.0079 none 0.2 39 2122 0.3953 0.0163 distribution 1.0 200 1800 0.0 0.0833 row,partition_period dedupe stage off: mode consumption dirty/55 serving diff at_most_once 2162 10 {'central': -9.18, 'north': -22.64, 'east': -9.81, 'west': -28.36, 'south': -20.28} at_least_once 2219 5 {'central': 6.19, 'north': 12.27, 'east': 3.64, 'west': 18.36, 'south': 6.45} exactly_once 2200 0 empty ``` The first four lines show how the silence is built. The row count check expects 2400 and finds 2381; the ratio is **0.0079** against a threshold of 0.05, that is the check stays six times below the threshold. The distribution check sees a deviation of **0.2488** in the most corrupted cell and stays **by a hair** below the threshold of 0.25. Partition completeness can see nothing, because the lost rows are spread over the five regions. The loss deletes no field, breaks no type, duplicates no key; it only leaves **fewer** rows — and the checks look at the shape of the row. The flaw is not in the code but in the **tolerance**: the fourth line shows the same check firing at tolerance 0.005. The loss is not unmeasurable, it is unmeasured. The second table sweeps the loss share and shows that the silence is not a point but a **band**. As the share goes from 0.02 to 0.20, the lost rows climb from 4 to 39 and the largest cell deviation from 0.1441 to 0.3953: the effect grows steadily with the share, but the distribution check sees it only at **0.20**. The row count check fires **nowhere** in the band; the reason is arithmetic, the loss is confined to a single period and that period is 200/2400 of the source, that is 0.0833. The last row is the most anomalous finding of the lesson. When 2026-07 never arrives at all, two checks fire **at once** and the deviation of the remaining cells is **0.00**: the cells of 2026-07 and 2026-08 have not been computed wrongly, they have not been written at all, and the remaining 45 cells are flawless. **Total loss is noisier than partial loss.** ## The Place and Cost of Repair The third table explains why the duplication stayed harmless. When the dedupe is turned off, at-least-once delivery is no longer absorbed: consumption rises to **2219**, 5 cells are polluted, and the serving shifts **upward** in every region. Duplication is not harmless by itself; it is the stage that makes it harmless, and the cost of that is again a set of keys. The first row of the same table completes the asymmetry. With the dedupe off, the result of at-most-once delivery **does not change at all**: 2162 rows, 10 polluted cells, the same deviations. No stage of the pipeline repairs the loss, because there is nothing to repair — the lost row leaves no trace anywhere. A duplicated row, on the other hand, does leave a trace: it is an extra record, it has a key, it can be counted and deleted. The backbone of the lesson is this sentence: **loss is silent, duplication is repairable**. Duplication is noisy; it both fires a check and can be undone by a stage. Loss is noiseless and cannot be undone; the only way to see it is **to know the number of rows sent**, that is to keep a counter outside the pipeline. This is why at-least-once is preferred: accepting duplication and making the write resistant to repetition is cheaper than accepting loss and searching for it. ## Summary - In at-most-once delivery 19 rows of period 2026-07 are lost, consumption falls from 2200 to 2162, 10 of the 55 cells are polluted, and all five values served shift (−9.18 … −28.36). - In at-least-once delivery 19 rows arrive twice, the dedupe stage drops the second copy, and the consumption, the polluted cells, and the serving difference stay at their baseline values. - No check catches the loss: the row ratio of 0.0079 stays below the 0.05 tolerance and the largest cell deviation of 0.2488 stays below the 0.25 tolerance. The uniqueness check catches the duplication at the entrance of the pipeline. - When the loss share is swept, the silence turns out to be a band: the distribution check fires only at 0.20, and the row count check fires at no share, because one period is only 0.0833 of the source. When the whole period is lost two checks fire at once — total loss is noisier than partial loss. - When the dedupe is turned off, the duplication stops being harmless (consumption 2219, serving upward in every region); the result of the loss does not change at all. The cost of exactly-once sits in the same place: 200 keys per period, 2400 per run. - Loss is silent, duplication is repairable: the duplication has a trace and can be deleted, the loss has no trace and can be seen only with a counter kept outside the pipeline. ## Next Step In five lessons the data always flowed in **one direction**: from the source system to the pipeline, from the pipeline to the summary, from the summary to the serving. Every measure assumed this direction — the ingestion method read the source, the delivery mode carried the row, the window decision gathered it in a period. One question was not asked: what happens when the value the pipeline produces **returns**. A number in the serving layer often does not stay there; it is written to an operational system or to an application that reads in the field, because that is where it does its real work. At the moment that write touches the pipeline's own input, the arrow closes into a circle. The next lesson measures the circle: after how many stages the value written back returns to its own input, and what it does to the number in the next run.