The harness in code · Code exercises

A webhook deduper

Implement the deduper that applies at-least-once, any-order deliveries in sequence order with one cursor, and hands a gap to the backfill after a set window.

codemediumExercise · 4 min · 43 of 75

On this page

A cursor is a smaller memory than a set of everything seen, and unlike the set it knows what is missing. This deduper keeps one number and a short hold, and it is what makes an at-least-once, any-order webhook safe to apply in order. Webhooks versus polling covers senders without sequence numbers; why at-least-once forces deduplication at all is idempotency.

The situation

The 40-person wholesaler this module uses in place of the agency, from why we write these five by hand, supplies 900 cafés. Its accounting system numbers every event on its ledger (invoice.issued, invoice.updated, payment.received, credit_note.issued) with a sequence number, contiguous from 1, and delivers each by webhook at least once, in no guaranteed order. The automation that keeps the CRM in step applied deliveries as they arrived.

Last month about 10,300 events arrived as 11,200 deliveries: 900 duplicates, and 180 that landed after a later event. Eleven payment.received events were applied before their invoice existed, failed and were dropped, and eleven cafés were chased for invoices they had paid. One event never arrived at all.

Of the 180 late deliveries, 176 landed within a minute of their successor and 4 within 9 minutes. The sender retries a delivery it believes failed for up to 48 hours. A missing event can be fetched from the API by its sequence number, one call each. At the Friday peak about 300 deliveries arrive in an hour.

The question

Implement the deduper so the examples below hold. The window it runs with in production is yours to set from the figures above.

def solve(deliveries, start, window):
    """deliveries: [(t, seq)] in arrival order, t non-decreasing, in seconds; seq is the
    sender's sequence number, contiguous per ledger. start: the first seq not yet applied.
    window: the longest a delivery may be held, in seconds.
    A delivery below the cursor, or one already held, is a duplicate and is dropped.
    Otherwise it is held, and everything contiguous from the cursor is applied. Then, while
    the delivery held longest has waited at least window, every seq below it that has not
    arrived is given up on and the cursor moves past it.
    Returns (applied, missing): the seqs applied, in order, and the seqs handed to the
    backfill, ascending. Deliveries still held after the last one are in neither list."""

solve([(0, 1), (2, 3), (3, 2), (4, 3), (5, 4)], 1, 600)      -> ([1, 2, 3, 4], [])
solve([(0, 1), (10, 3), (40, 4), (70, 5)], 1, 60)            -> ([1, 3, 4, 5], [2])
solve([(0, 1), (10, 3), (40, 4), (70, 5), (80, 2)], 1, 60)   -> ([1, 3, 4, 5], [2])
solve([(0, 1), (10, 3), (70, 2)], 1, 60)                     -> ([1, 2, 3], [])

A strong answer · Rubric · Where answers go wrong · Solution

Write your answer first, then open

A strong answer

Keep a cursor, the next sequence number to apply, and a dict of held deliveries from sequence number to arrival time. A delivery below the cursor, or already held, is dropped. The cursor replaces a set of every id seen, so memory does not grow with the stream, and a late copy of a number already handed to the backfill is dropped too (example 3), so the backfill is that event's only writer. Otherwise hold it and drain, applying from the cursor while the next number is held: example 1 holds 3 until 2 arrives, then applies both.

The clock runs after the arrival is handled, so a missing event that lands at the deadline is applied, not given up on (example 4). Then, while the delivery held longest has waited at least the window, record every absent number below it as missing, move the cursor past them and drain (example 2). The hold never exceeds one window of traffic: about 50 deliveries at the Friday peak with a 10-minute window.

Set the window to 600 seconds. Every late delivery last month landed within 9 minutes; a later retry, up to the sender's 48 hours, falls below the cursor and is dropped. A 60-second window would also send the 4 slow deliveries to the backfill, 5 fetches a month instead of 1; both are cheap. What 600 costs is staleness: events behind a gap wait up to 10 minutes, a few times a month, and nobody reads the CRM ledger to the minute. The evidence that would change it is a step that needs a payment within a minute, such as releasing a held order, which buys the 60-second window and its five fetches.

Rubric

drops anything below the cursor or already held, without a set of seen ids that grows forever
applies in sequence order, draining the contiguous run after each arrival
handles the arrival before the clock, so a missing event that lands at the deadline is applied
gives up on a gap once the delivery held longest has waited the window, and records the missing numbers for the backfill
sets the production window from last month's lateness, not the sender's 48-hour retry span, and names what it costs in staleness

Where answers go wrong

"Keep a set of every event id and drop repeats." The duplicates go and the order problem stays: example 1 applies 3 before 2, the eleven payments still land before their invoices, and the set grows by 10,300 ids a month.

"Hold every delivery until its gap fills." The one event that never arrives stalls the ledger. From then on every delivery is held, and by Monday morning the CRM is frozen at Friday afternoon with hundreds of events in memory.

"Apply anything newer than the last applied number and drop the rest." That is a cursor with no hold. The 180 late deliveries are thrown away, payments among them, and nothing records them as missing, so the backfill never fetches them.

Solution

Python 3 · standard library44 lines · 10 of 10 tests passing

"""A webhook deduper — a cursor and a bounded hold put at-least-once deliveries back in sequence order."""


def solve(deliveries, start, window):
    """deliveries: [(t, seq)] in arrival order, t non-decreasing, in seconds; seq is the
    sender's sequence number, contiguous per ledger. start: the first seq not yet applied.
    window: the longest a delivery may be held, in seconds.
    A delivery below the cursor, or one already held, is a duplicate and is dropped.
    Otherwise it is held, and everything contiguous from the cursor is applied. Then, while
    the delivery held longest has waited at least window, every seq below it that has not
    arrived is given up on and the cursor moves past it.
    Returns (applied, missing): the seqs applied, in order, and the seqs handed to the
    backfill, ascending. Deliveries still held after the last one are in neither list."""
    held = {}                          # seq -> arrival time; only seqs above the cursor
    cursor = start                     # invariant: every seq below the cursor is applied or missing
    applied, missing = [], []

    def drain():
        nonlocal cursor
        while cursor in held:          # apply the contiguous run that starts at the cursor
            del held[cursor]
            applied.append(cursor)
            cursor += 1

    for t, seq in deliveries:
        if seq < cursor or seq in held:
            continue                   # a duplicate, or a seq the backfill already owns: the cursor
                                       # replaces a seen-set, so memory never grows with the stream
        held[seq] = t
        drain()                        # the arrival is handled before the clock, so a missing
                                       # seq that lands at the deadline is applied, not given up on
        while held:
            oldest = min(held, key=lambda s: (held[s], s))
            if t - held[oldest] < window:
                break                  # nothing has waited a full window yet
            while cursor < oldest:     # give up on every gap below the delivery held longest
                if cursor in held:
                    del held[cursor]
                    applied.append(cursor)
                else:
                    missing.append(cursor)
                cursor += 1
            drain()
    return applied, missing

The cases that ran

TESTS = [
    # the examples on the page
    (([(0, 1), (2, 3), (3, 2), (4, 3), (5, 4)], 1, 600), ([1, 2, 3, 4], [])),
    (([(0, 1), (10, 3), (40, 4), (70, 5)], 1, 60), ([1, 3, 4, 5], [2])),
    (([(0, 1), (10, 3), (40, 4), (70, 5), (80, 2)], 1, 60), ([1, 3, 4, 5], [2])),
    (([(0, 1), (10, 3), (70, 2)], 1, 60), ([1, 2, 3], [])),
    # deliveries below the backfill's cursor are duplicates
    (([(0, 4), (1, 5), (2, 6)], 5, 600), ([5, 6], [])),
    # a gap that is still inside its window holds everything behind it
    (([(0, 2), (30, 3)], 1, 60), ([], [])),
    # two gaps closed by one late delivery; the newest held seq stays held
    (([(0, 2), (5, 4), (70, 6)], 1, 60), ([2, 4], [1, 3])),
    # one second short of the window: still held
    (([(0, 1), (10, 3), (69, 4)], 1, 60), ([1], [])),
    # a duplicate of a held seq is dropped and does not restart its clock
    (([(0, 2), (50, 2), (60, 3)], 1, 60), ([2, 3], [1])),
    (([], 1, 600), ([], [])),
]