Durable Inbox and Outbox: Unit A Exercises

Student edition · 90 minutes of dedicated work · 2026-09-26

Runs on Google Colab as it ships today (Python 3.12), or on any local Python 3.12+ kernel. It needs nothing beyond Python's standard library.

This is the first of Chapter 8's two practical units, each a complete ninety-minute session.

• Unit A builds the inbox: requests admitted once per occurrence, bounded before any work runs, and moved through states that the database itself enforces.
• Unit B finishes work together with its report, and sends reports across a network that can lose a reply.

You bring Chapter 4's tables, constraints and transactions. Triggers, state machines and Little's law are introduced below, before the main task.

By the end you should be able to:

1. Predict what a restarted program knows about the requests it had accepted, and check the prediction.
2. Tell the text of a request from the identity of its occurrence, and say when a repeat is a duplicate and when it is a conflict.
3. Implement admit, which decides and records one request in one transaction and refuses work before storing anything when the inbox is full.
4. Make the database refuse a state transition that the state machine does not name, and read a capacity through Little's law, $L = \lambda W$.

These times are planning estimates, not measured completion times. Run All only checks that the notebook executes; the unfinished student functions deliberately report NEEDS_WORK. Keep your first attempt before you open the answers.

Run the self-contained setup

The unit needs only Python's standard library. The collapsed cell below creates your work folder and defines the supplied parts of the unit:

• The work table, with a trigger that enforces the state machine.
• A fresh-database helper.
• A supplied worker with two moves: claim, which takes the oldest pending request, and finish_plain, which marks it finished. finish_plain records the work as done but does not keep the answer anywhere. Unit B starts from exactly that gap.
• An independent observer, which reads the file through its own connection.

Run setup on every fresh kernel. Your saved work lives in practical-work/ch08-a. Restarting a kernel clears variables, not saved files.

Supplied setup, schema, worker and observer

The complete supplied setup cell is in the downloadable notebook. Run it there before attempting the cells below. This web edition omits only that compressed setup payload.

Commit to a prediction before the examples

Lucy taps "Prepare the opening brief" in the car park, the signal drops, and she taps it again. Her program keeps accepted requests in a Python list, and it restarts before the worker reaches them. Before you run anything below, write down:

• how many requests the restarted program knows about;
• how many briefs would be prepared if the program had not restarted.

prediction_notes = {
    "prediction": "Write what a restarted program knows, and how many briefs run otherwise.",
    "reason": "Name the rule behind that prediction.",
    "falsifier": "Name an observation that would prove the explanation wrong.",
    "revision": "After execution, explain what changed in your understanding.",
}

A list forgets; a table does not

A request the program has accepted is a promise. A promise kept only in a variable dies with the process. The example below keeps the same two requests in a list and in a table, then "restarts": a new list, and a new connection to the same file. Predict both lines.

with tempfile.TemporaryDirectory() as intro_folder:
    intro_list = ["Prepare the opening brief", "Prepare the opening brief"]
    intro_path = Path(intro_folder) / "inbox.sqlite3"
    intro_writer = connect(intro_path)
    intro_writer.execute("CREATE TABLE work (text TEXT NOT NULL)")
    intro_writer.executemany("INSERT INTO work VALUES (?)", [(text,) for text in intro_list])
    intro_writer.close()

    intro_list = []  # what a new process starts with
    intro_reader = connect(intro_path)
    intro_rows = intro_reader.execute("SELECT text FROM work").fetchall()
    intro_reader.close()
print("the list after a restart: ", intro_list)
print("the table after a restart:", intro_rows)

The table kept both requests, so it also kept the mistake: two taps became two rows, and a worker would prepare two briefs and pay the model twice. Durability alone does not answer the question "is this the same request?"

Identity belongs to the occurrence, not the text

Lucy's two taps carry the same source identity: the messaging service names each message, and her phone resent the same message. An hour later the till might send the same text, "Count vanilla", for a different reason. That is a different occurrence with its own identity. Text is not identity.

The work table therefore makes source_id unique. When a repeat arrives, the decision has three outcomes:

• duplicate: same identity, same content. Record nothing new.
• conflict: same identity, different content. Someone reused a name, so refuse the request and say so.
• new: an identity not seen before. Admit it.

INSERT OR IGNORE looks like a shortcut for the duplicate case. Predict what it does to the conflicting request below.

with tempfile.TemporaryDirectory() as intro_folder:
    intro_db = connect(Path(intro_folder) / "identity.sqlite3")
    intro_db.execute("CREATE TABLE work (source_id TEXT PRIMARY KEY, text TEXT NOT NULL)")
    for intro_request in [
        ("tg-101", "Prepare the opening brief"),
        ("tg-101", "Prepare the opening brief"),
        ("tg-101", "Order everything"),
        ("till-7", "Count vanilla"),
        ("till-8", "Count vanilla"),
    ]:
        intro_changed = intro_db.execute(
            "INSERT OR IGNORE INTO work VALUES (?, ?)", intro_request
        ).rowcount
        print(intro_request, "stored" if intro_changed else "ignored")
    intro_db.close()

INSERT OR IGNORE treated the conflict ("Order everything" under tg-101) exactly like the duplicate. It stored nothing and told nobody. A reused identity is a fault, whether it comes from a buggy client or a till whose counter reset. Your admit must read the existing row and compare the content, so that a conflict is reported rather than silently dropped.

States, and a trigger that enforces them

A work item is pending until a worker claims it, running while the worker is on it, and finished afterwards. A state machine lists the allowed moves:

Everything else is forbidden. Finished work does not go back to pending, and pending work does not become finished without having run.

The setup's trigger makes the database refuse any other update of state, whichever code issues it. A trigger runs implicitly: the UPDATE never mentions it. That lets it catch a bug you did not know you had, and it is also why this one enforces a single rule, which you can read in WORK_SCHEMA. Predict which of the three updates below the database refuses.

with tempfile.TemporaryDirectory() as intro_folder:
    intro_db = connect(Path(intro_folder) / "states.sqlite3")
    for intro_statement in WORK_SCHEMA:
        intro_db.execute(intro_statement)
    intro_db.execute(
        "INSERT INTO work (source_id, session_id, text, state)"
        " VALUES ('tg-101', 'lucy', 'Prepare the opening brief', 'pending')"
    )
    for intro_state in ["finished", "running", "pending"]:
        try:
            intro_db.execute("UPDATE work SET state = ? WHERE source_id = 'tg-101'", (intro_state,))
            print("to", intro_state, "-> allowed")
        except sqlite3.IntegrityError as intro_error:
            print("to", intro_state, "-> refused:", intro_error)
    print("allowed moves:", sorted(TRANSITIONS))
    intro_db.close()

How full the inbox gets: Little's law

An inbox that accepts everything will accept more than the worker can do on a bad morning. So admission is bounded. Once the number of open items (pending or running) reaches a capacity $C$, a new request is refused before anything is stored. A refusal the shop can explain is better than a promise it cannot keep.

To read a capacity, use Little's law. Let $L$ be the average number of open items, $\lambda$ the rate at which items are admitted, and $W$ the average time from admission to finish. Then

Why it holds. Draw the open count $L(t)$ over a period of length $T$, and count the area under it in two ways:

• Time slice by time slice, the area is $\bar{L}\,T$.
• Item by item, each item adds one unit of height for as long as it is open. So the area is also the sum of all items' times in the system, $\sum_i W_i = N\bar{W}$.

Set the two counts equal: $\bar{L} = (N/T)\,\bar{W} = \lambda\bar{W}$.

The example below counts both ways on a small timeline, in whole minutes, with every item finished inside the period. Predict whether the two areas agree exactly.

intro_timeline = [(0, 3), (1, 7), (2, 4), (6, 9), (6, 12), (10, 13)]  # (admitted, finished)
intro_period = 15
intro_area_by_time = sum(
    sum(1 for admitted, finished in intro_timeline if admitted <= minute < finished)
    for minute in range(intro_period)
)
intro_area_by_item = sum(finished - admitted for admitted, finished in intro_timeline)
intro_rate = len(intro_timeline) / intro_period
intro_wait = intro_area_by_item / len(intro_timeline)
print("area, time by time:", intro_area_by_time, " area, item by item:", intro_area_by_item)
print(f"L = {intro_area_by_time / intro_period:.3f}   lambda * W = {intro_rate * intro_wait:.3f}")

The two areas agree exactly, because every item's time in the system falls inside the period. On a real queue, items still open at the edges make the two counts differ slightly, and the difference shrinks as the period grows.

Applying it to Lucy's shop. Her worker takes three minutes per brief, and requests arrive at 0.3 per minute. The chapter's experiment measured about five open items for an unbounded inbox, each waiting about sixteen minutes: $0.3 \times 16.4 \approx 4.9$. The capacity is the number of promises the shop is willing to have outstanding at once.

Retrieval check: explain in your own words:

• a list versus a table, across a restart;
• a duplicate versus a conflict;
• why the trigger refuses pending → finished;
• what $L$, $\lambda$ and $W$ each measure.

Reference: SQLite's CREATE TRIGGER and Python's sqlite3 documentation.

Main practical: construct, connect and challenge

Every request that reaches Lucy's shop, from her phone, the till or a schedule, arrives with its source identity, a session and its text. You will implement admit, which decides what to do with one request and records the decision so that no later reader can see a half-made one.

1. The supplied schema

inbox_path = fresh_database("unit-a-inbox")
print(observe(inbox_path))

2. Construct admit

admit(connection, request, capacity) receives a connection from connect, a request dictionary with source_id, session_id and text, and an integer capacity. It must:

• read, decide and write in one transaction;
• return "duplicate", writing nothing, when the source_id exists with the same session_id and text. It answers this even when the inbox is full: Lucy's second tap is not new work.
• return "conflict", writing nothing, when the source_id exists with other content;
• return "refused", writing nothing, when the request is new but the number of open items (pending or running) has reached capacity. Finished work does not count.
• otherwise insert the request as pending and return "accepted";
• leave no transaction open, whatever it returns or raises.

The starter below counts the requests still waiting and inserts without asking who sent them. Run the visible cases to see where it fails, then repair it.

def admit(connection, request, capacity):
    """Return "accepted", "duplicate", "conflict" or "refused"."""
    (waiting,) = connection.execute("SELECT COUNT(*) FROM work WHERE state = 'pending'").fetchone()
    if waiting >= capacity:
        return "refused"
    connection.execute(
        "INSERT INTO work (source_id, session_id, text, state) VALUES (?, ?, ?, 'pending')",
        (request["source_id"], request["session_id"], request["text"]),
    )
    return "accepted"

Hint 1: the order of the questions

Ask about identity first and capacity second. A repeat of a known occurrence is a duplicate or a conflict whatever the inbox holds. Only a new occurrence asks for a place.

Hint 2: what counts as open

Count WHERE state IN ('pending', 'running'). Finished work has kept its promise and holds no place.

Hint 3: why one transaction

Chapters 9 and 10 add more producers: the phone adapter, the clock and the stock monitor. If two of them each read "two open items" and then each insert, a capacity of three admits four. BEGIN IMMEDIATE takes the write lock before the first read, so the count cannot change between the read and the insert. Use the same structure as Chapter 4: BEGIN IMMEDIATE, try: read, decide, write, except BaseException: ROLLBACK; raise, then COMMIT.

import copy

SEED = [
    ("done-1", "lucy", "Prepare yesterday's close", "finished"),
    ("tg-101", "lucy", "Prepare the opening brief", "pending"),
    ("till-7", "till", "Count vanilla", "running"),
]


def seeded():
    """A fresh inbox holding one finished, one pending and one running item."""
    path = fresh_database("visible")
    seed = connect(path)
    seed.executemany(
        "INSERT INTO work (source_id, session_id, text, state) VALUES (?, ?, ?, ?)", SEED
    )
    seed.close()
    return path


def with_rows(*extra):
    return [list(row) for row in SEED] + [list(row) for row in extra]


VISIBLE_CASES = [
    (
        "a new occurrence with room",
        {"source_id": "till-8", "session_id": "till", "text": "Count chocolate"},
        3,
        "accepted",
        with_rows(("till-8", "till", "Count chocolate", "pending")),
    ),
    (
        "the same text, a new occurrence",
        {"source_id": "till-9", "session_id": "till", "text": "Count vanilla"},
        3,
        "accepted",
        with_rows(("till-9", "till", "Count vanilla", "pending")),
    ),
    (
        "Lucy's second tap",
        {"source_id": "tg-101", "session_id": "lucy", "text": "Prepare the opening brief"},
        3,
        "duplicate",
        with_rows(),
    ),
    (
        "a reused identity",
        {"source_id": "tg-101", "session_id": "lucy", "text": "Order everything"},
        3,
        "conflict",
        with_rows(),
    ),
    (
        "a full inbox",
        {"source_id": "tg-102", "session_id": "lucy", "text": "Order everything"},
        2,
        "refused",
        with_rows(),
    ),
    (
        "Lucy's second tap on a full inbox",
        {"source_id": "tg-101", "session_id": "lucy", "text": "Prepare the opening brief"},
        2,
        "duplicate",
        with_rows(),
    ),
]


def grade_admit(candidate, cases):
    rows = []
    for label, request, capacity, expected, expected_work in cases:
        path = seeded()
        connection = connect(path)
        supplied = copy.deepcopy(request)
        try:
            observed = candidate(connection, supplied, capacity)
        except Exception as error:
            observed = type(error).__name__
        if connection.in_transaction:
            connection.execute("ROLLBACK")
            observed = f"{observed} (left a transaction open)"
        connection.close()
        work = observe(path)["work"]
        passed = observed == expected and work == expected_work and supplied == request
        rows.append(
            {
                "case": label,
                "expected": expected,
                "observed": observed,
                "status": "PASS" if passed else "FAIL",
            }
        )
    return rows


visible_results = grade_admit(admit, VISIBLE_CASES)
VISIBLE_PASSED = all(row["status"] == "PASS" for row in visible_results)
for visible_row in visible_results:
    print(visible_row["status"], visible_row["case"], "->", visible_row["observed"])
print("VISIBLE_CONTRACT", "PASSED" if VISIBLE_PASSED else "NEEDS_WORK")

3. Connect it to a morning of requests and a worker

Once the visible cases pass, the cell below runs a morning through your admit on a real file, with a capacity of three:

1. Lucy's tap and its repeat.
2. Two till counts with the same text.
3. A request that finds the inbox full.
4. The supplied worker claims and finishes the oldest item.
5. The refused request is sent again.

It also tries to move finished work back to pending. Then it closes the connection and asks the independent observer what the file holds.

Before running it, predict every outcome and the final state of each item.

MORNING = [
    {"source_id": "tg-101", "session_id": "lucy", "text": "Prepare the opening brief"},
    {"source_id": "tg-101", "session_id": "lucy", "text": "Prepare the opening brief"},
    {"source_id": "till-7", "session_id": "till", "text": "Count vanilla"},
    {"source_id": "till-8", "session_id": "till", "text": "Count vanilla"},
    {"source_id": "tg-102", "session_id": "lucy", "text": "Order everything"},
]
connected = None
if VISIBLE_PASSED:
    inbox_path = fresh_database("unit-a-inbox")
    inbox = connect(inbox_path)
    outcomes = [admit(inbox, request, 3) for request in MORNING]
    claimed = claim(inbox, "w1")
    finish_plain(inbox, claimed)
    outcomes.append(admit(inbox, MORNING[-1], 3))
    try:
        inbox.execute("UPDATE work SET state = 'pending' WHERE work_id = ?", (claimed,))
        backwards = "allowed"
    except sqlite3.IntegrityError as error:
        backwards = f"refused: {error}"
    inbox.close()
    connected = observe(inbox_path)
    print("outcomes:", outcomes)
    print("finished work back to pending:", backwards)
    print("reopened:", connected)
    assert outcomes == ["accepted", "duplicate", "accepted", "accepted", "refused", "accepted"]
    assert backwards == "refused: transition not allowed"
    assert [row[3] for row in connected["work"]] == ["finished", "pending", "pending", "pending"]
    assert connected["open"] == 3
else:
    print("CONNECTION_NOT_READY — repair admit, then run again.")

Trace tg-102 through the morning. Why was it refused the first time and accepted the second, although the request was identical? Which row changed state in between, and what did the worker do?

Then look at tg-101 in the observer's output. It is finished, but what did the brief say? Nothing on disk records it. Unit B starts from exactly this file.

4. Save the handoff

Unit B starts from this morning's inbox. The handoff records every item and what the observer saw, in a JSON file in your work folder.

ARTIFACT_PATH = Path("ch08-unit-a-handoff-v1.json")
artifact_status = "NOT_WRITTEN"
if connected is not None:
    handoff = {"unit": "ch08-a", "status": "COMPLETED", "work": connected["work"]}
    ARTIFACT_PATH.write_text(json.dumps(handoff, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    artifact_status = "WRITTEN"
    print(ARTIFACT_PATH)
else:
    print("HANDOFF_NOT_WRITTEN")

Exit ticket

Answer three questions:

• Why must the duplicate check come before the capacity check?
• Why is INSERT OR IGNORE not an answer to duplicates?
• Which observation proves that finished work cannot go back to pending?

exercise_report = {
    "unit": "ch08-a",
    "attempted": 1,
    "completed": int(VISIBLE_PASSED),
    "failed": int(not VISIBLE_PASSED),
    "skipped": 0,
    "connection": "PASSED" if connected else "NOT_READY",
    "handoff": artifact_status,
}
print("EXERCISE_REPORT=" + json.dumps(exercise_report, sort_keys=True))

Changed-constraint construction: admit by promised minutes

Allow fifteen minutes: three to predict, eight to implement and trace, and four for a case of your own.

A count of open items treats a one-minute stock count and a twenty-minute brief as equal promises. Lucy would rather bound the minutes of work she has promised. Implement that rule with no database.

transfer_admit(open_work, request, budget) receives:

• open_work: a dictionary from source_id to a dictionary with session_id, text, minutes and state;
• request: a dictionary with the same four keys, minus state, plus its own source_id;
• budget: an integer number of minutes.

Return "DUPLICATE" or "CONFLICT" by the same identity rule as admit. Content now means session, text and minutes. Otherwise:

• return "REFUSE" when the minutes of pending and running work, plus the request's own minutes, would exceed the budget;
• return "ACCEPT" otherwise.

Do not change either input. Write your expected values before you run the table. The driver passes your function in directly, copies the inputs, and checks that they come back unchanged.

Hint: what changed from a count A count asks "is there a free place?", so it compares the open count with the capacity. A size asks "does this request fit?", so it adds the request's minutes before comparing. A request exactly at the budget fits.

def transfer_admit(open_work, request, budget):
    raise NotImplementedError("Decide identity first, then whether the minutes fit")

BRIEF = {"session_id": "lucy", "text": "Prepare the opening brief", "minutes": 20}
COUNT = {"session_id": "till", "text": "Count vanilla", "minutes": 1}
TRANSFER_CASES = [
    (
        "an empty inbox",
        [{}, {"source_id": "tg-101", **BRIEF}, 30],
        "ACCEPT",
    ),
    (
        "a brief that does not fit",
        [
            {"tg-101": {**BRIEF, "state": "running"}, "till-7": {**COUNT, "state": "pending"}},
            {"source_id": "tg-102", **BRIEF},
            30,
        ],
        "REFUSE",
    ),
    (
        "a count that fits beside the brief",
        [{"tg-101": {**BRIEF, "state": "running"}}, {"source_id": "till-7", **COUNT}, 30],
        "ACCEPT",
    ),
    (
        "finished work holds no minutes",
        [{"tg-101": {**BRIEF, "state": "finished"}}, {"source_id": "tg-102", **BRIEF}, 30],
        "ACCEPT",
    ),
    (
        "exactly at the budget",
        [{"tg-101": {**BRIEF, "state": "pending"}}, {"source_id": "tg-102", **BRIEF}, 40],
        "ACCEPT",
    ),
    (
        "a repeat while full",
        [{"tg-101": {**BRIEF, "state": "pending"}}, {"source_id": "tg-101", **BRIEF}, 20],
        "DUPLICATE",
    ),
    (
        "same identity, other minutes",
        [
            {"tg-101": {**BRIEF, "state": "pending"}},
            {"source_id": "tg-101", **BRIEF, "minutes": 25},
            60,
        ],
        "CONFLICT",
    ),
]


def run_transfer(candidate, cases):
    observations = []
    for label, arguments, expected in cases:
        supplied = copy.deepcopy(arguments)
        try:
            actual = candidate(*supplied)
        except NotImplementedError:
            actual = {"unfinished": True}
        except Exception as error:
            actual = {"raises": type(error).__name__}
        passed = actual == expected and supplied == arguments
        observations.append(
            {"case": label, "expected": expected, "observed": actual, "passed": passed}
        )
        print("PASS" if passed else "NEEDS_WORK", label, "expected", expected, "observed", actual)
    return observations


transfer_observations = run_transfer(transfer_admit, TRANSFER_CASES)
TRANSFER_PASSED = all(row["passed"] for row in transfer_observations)
print("TRANSFER_STATUS", "PASS" if TRANSFER_PASSED else "NEEDS_WORK")

Design a counterexample and retrieve the mechanism

Add one case with an expected outcome you worked out independently, and rerun the driver.

Then, in a temporary copy, replace your function with one that checks the budget without adding the request's own minutes, and name the case that rejects it. Explain which is the better promise, and why:

• a capacity counted in items;
• a capacity counted in minutes.

Save your evidence and explain the result

Fill in the prediction notes and your explanation before saving. Include:

• the exact observed value, and the input that caused it;
• your code's invocation point;
• one failed hypothesis;
• the strongest claim the evidence still cannot support.

This unit ran one producer and one worker in one process. It did not run two producers at once, and it did not measure a real arrival rate.

explanation_notes = {
    "causal_trace": "Explain the input, learner invocation and observed result.",
    "failed_hypothesis": "Describe a prediction the evidence changed.",
    "remaining_limit": "Name the guarantee not established by this experiment.",
}
course_submission = {
    "unit": "ch08-a",
    "planned_minutes": 90,
    "starting_evidence": globals().get("HANDOFF_ORIGIN", "INDEPENDENT_UNIT_A"),
    "prediction": prediction_notes,
    "explanation": explanation_notes,
    "core_report": exercise_report,
    "transfer": transfer_observations,
    "explanation_review": "HUMAN_REVIEW_REQUIRED",
}
submission_path = COURSE_WORK / "ch08-a-submission-v1.json"
submission_path.write_text(
    json.dumps(course_submission, indent=2, sort_keys=True), encoding="utf-8"
)
print("Saved evidence:", submission_path)
print(
    "COURSE_REPORT="
    + json.dumps(
        {
            "unit": "ch08-a",
            "transfer_passed": TRANSFER_PASSED,
            "starting_evidence": course_submission["starting_evidence"],
            "edition": "student",
        },
        sort_keys=True,
    )
)

Keep building with Prof Rod

Found this material through a colleague, classroom or shared download? Get the complete book at profrod.ai/book and join the Prof Rod learner community. Bring one result, one question or one failure you learned from. Share this resource with another learner and keep its source links with it so they can find the full course and future updates.