Design a Publish-to-Search Pipeline
Build a resilient publish-to-search pipeline with an outbox and safe replay.
Introduction
30 Second Summary
Publishing a learning project feels complete when the author clicks the final button. A learner searching moments later expects that project to appear.
In this project, you will build a local simulator that sends published projects from a source-of-truth database to a separate search index. You will expose a failed publish path before repairing it with reliable delivery and safe recovery.
What You'll Build
Your finished demo shows a project survive a publishing failure before becoming searchable through a replay-safe pipeline.
By the end of this project, you'll have:
- A working failure demonstration where a published project exists in the content database but remains missing from search.
- A durable publish-to-search pipeline that commits content with a transactional outbox before an asynchronous worker updates search.
- An interview-ready architecture explanation covering eventual consistency, idempotent replay, failure recovery, scaling, and observability.
- Secret Mission: Diagnose a growing indexing backlog before safely replaying events until every project becomes searchable.
Are there any prerequisites?
You need Visual Studio Code and Docker Desktop on a Mac. No local Python installation or cloud account is required.
Before We Start
Before the hands-on work begins, this step helps you lock in the system you are designing. You are building a local simulator of a hypothetical NextWork publish-to-search pipeline where newly published learning content must become searchable quickly and reliably.
Set Up the Local Simulator
A publish-to-search failure only teaches you something when you can reproduce it. Docker Desktop gives your Mac the pinned Python 3.14.7 environment without another local runtime.
Published projects live in one SQLite file. Search state lives in a second file.
This separation creates the system boundary that later failures will expose. The simulator first needs a repeatable workspace around that boundary.
In this step, get ready to:
- Prepare the local project workspace.
- Build the starter database schema.
- Initialize the two local databases through Docker.
Create the workspace
Visual Studio Code treats an open folder as a workspace. Keeping the script beside its database files makes the simulator easy to run from one integrated terminal.
- Press Cmd+Space to open macOS search.
- Type Visual Studio Code and press Enter to start it.
- Use Visual Studio Code's folder-opening flow.
- Select your Desktop in the macOS folder picker.
- Create a folder named nextwork-search-design.
- Open nextwork-search-design as the workspace.
Why Open a Folder as a Workspace?
A workspace gives Visual Studio Code one project boundary. Its Explorer and integrated terminal can now point at the same nextwork-search-design folder.
- Create search_pipeline.py inside nextwork-search-design from the Explorer sidebar.
- Open Visual Studio Code's integrated terminal for the current workspace.
- Press Cmd+Space to return to macOS search.
- Type Docker Desktop and press Enter to start it.
You should see nextwork-search-design at the top of the Explorer. The terminal should be ready inside that folder.
Add the database foundation
The starter script defines two independent database connections. Explicit commits make the source-side transaction boundary visible as the simulator grows.
- Add the imports, database filenames, and connection helper to search_pipeline.py by pasting this code:
import sqlite3
import sys
import time
CONTENT_DB = "content.db"
SEARCH_DB = "search.db"
def connect(path):
return sqlite3.connect(path, autocommit=False)What Does This Foundation Do?
- The sqlite3 module opens the local database files.
- The sys module provides the command-line arguments.
- The time module supports timestamps used by the complete simulator.
- The connect() helper uses autocommit=False so each transaction ends through an explicit commit or rollback.
- Save search_pipeline.py.
- Check the editor diagnostics for search_pipeline.py.
You should see no syntax problem in the imports, constants, or connection helper.
Seeing an Import or Syntax Problem?
Check that each import begins at the far left of the file. Confirm that CONTENT_DB and SEARCH_DB use the exact capitalization shown above.
Ask for help with the first section of the script if the diagnostic remains: help me check the imports, constants, and connect helper in my Python simulator
The content database needs the authoritative project row plus a durable record of work waiting for search. Those responsibilities belong in separate tables inside the same database.
- Add the source-database portion of initialize() below connect() by pasting this code:
def initialize(quiet=False):
content = connect(CONTENT_DB)
content.executescript(
"""
CREATE TABLE IF NOT EXISTS content (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS outbox (
event_id TEXT PRIMARY KEY NOT NULL,
slug TEXT NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL,
processed_at REAL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
"""
)
content.commit()
content.close()What Does the Source Schema Store?
- The content table stores each published project's slug, title, and publication time.
- The outbox table stores indexing events plus their processing state.
- The call to commit() persists both table definitions before the connection closes.
- Save search_pipeline.py.
- Check the editor diagnostics for initialize().
You should see no indentation or string-delimiter problem in the source-database section.
Seeing a Problem in the Source Schema?
Make sure the SQL stays between the matching triple quotes. Keep every line from content = connect(CONTENT_DB) through content.close() indented inside initialize().
Use this prompt if the indentation still looks wrong: help me check the source-database section of my initialize function
The search database is a separate read model. Its event ledger lets a future worker recognize an indexing event that it has already applied.
- Extend initialize() directly below content.close() by pasting this code with the indentation shown:
search = connect(SEARCH_DB)
search.executescript(
"""
CREATE TABLE IF NOT EXISTS documents (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
source_published_at REAL NOT NULL,
indexed_at REAL NOT NULL,
event_id TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS processed_events (
event_id TEXT PRIMARY KEY NOT NULL,
processed_at REAL NOT NULL
);
"""
)
search.commit()
search.close()
if not quiet:
print("Initialized content.db and search.db.")What Does the Search Schema Store?
- The documents table stores the project fields exposed to search.
- The processed_events table records event IDs already applied to the read model.
- The quiet flag lets other commands initialize missing tables without printing the setup confirmation.
- Save search_pipeline.py.
- Check the editor diagnostics for the complete initialize() function.
You should see no syntax problem across either schema definition.
Seeing an Indentation Problem in the Search Section?
Keep the search section aligned with content = connect(CONTENT_DB) inside initialize(). The line beginning with def initialize remains the only unindented line in this function.
Use this prompt for a line-by-line indentation check: help me align the search-database section inside initialize
Repeated experiments need a clean starting point. The reset path deletes rows while preserving the four table definitions.
- Add reset() below initialize() by pasting this code:
def reset():
content = connect(CONTENT_DB)
content.execute("DELETE FROM outbox")
content.execute("DELETE FROM content")
content.commit()
content.close()
search = connect(SEARCH_DB)
search.execute("DELETE FROM processed_events")
search.execute("DELETE FROM documents")
search.commit()
search.close()
print("Reset content.db and search.db.")What Does Reset Preserve?
The delete statements remove experiment data from all four tables. The database files plus their schemas remain available for another run.
- Save search_pipeline.py.
- Check the editor diagnostics for reset().
You should see no syntax problem in the reset path.
Seeing a Problem in Reset?
Confirm that each DELETE FROM statement names one of the four tables created by initialize(). Keep each commit before its matching connection closes.
Use this prompt if a table name or connection looks mismatched: help me compare reset with the four tables in initialize
A command-line simulator needs a clear response when the supplied command is missing or invalid. The usage function provides that response with a nonzero exit.
- Add usage() below reset() by pasting this code:
def usage(message=None):
if message:
print(message)
print(
"Usage: search_pipeline.py "
"init | reset | status | search TERM | "
"naive-publish SLUG TITLE [--fail] | "
"reliable-publish SLUG TITLE | work [--fail-after-index]"
)
sys.exit(2)Why Include the Full Command Surface?
The usage message documents the complete simulator interface. This starter version activates only the init and reset routes.
- Save search_pipeline.py.
- Check the editor diagnostics for usage().
You should see no problem with the adjacent usage strings or the closing parentheses.
Seeing a String Problem in Usage?
Keep each usage fragment inside matching double quotes. Python combines the adjacent strings inside print().
Use this prompt if the editor still highlights the function: help me find the unmatched quote or parenthesis in usage
The final starter function turns command-line arguments into actions. It initializes missing tables quietly before the reset path runs.
- Add main() below usage() by pasting this code:
def main():
if len(sys.argv) < 2:
usage()
command = sys.argv[1]
if command == "init" and len(sys.argv) == 2:
initialize()
return
initialize(quiet=True)
if command == "reset" and len(sys.argv) == 2:
reset()
else:
usage(f"Unknown command or invalid arguments: {command}")
if __name__ == "__main__":
main()How Does Command Routing Work?
- The init route creates the schemas with a visible confirmation.
- The reset route initializes missing schemas quietly before deleting any rows.
- An unsupported command reaches usage() with a description of the invalid input.
- The module guard calls main() when Python runs this file as a script.
- Save search_pipeline.py.
- Check the editor diagnostics for the complete starter script.
You should see no syntax problem in search_pipeline.py.
Seeing a Problem in Command Routing?
Keep the module guard at the far left of the file. Confirm that the body of main() uses four spaces for each indentation level.
Use this prompt if the route structure still fails inspection: help me check the init and reset routing in main
✔️ Awesome, I've got everything!
Your starter script now matches the setup state. The next check runs it inside the pinned Python container.
ⓧ I'd like to double check the full code
- Compare your complete search_pipeline.py with this starter version:
import sqlite3
import sys
import time
CONTENT_DB = "content.db"
SEARCH_DB = "search.db"
def connect(path):
return sqlite3.connect(path, autocommit=False)
def initialize(quiet=False):
content = connect(CONTENT_DB)
content.executescript(
"""
CREATE TABLE IF NOT EXISTS content (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS outbox (
event_id TEXT PRIMARY KEY NOT NULL,
slug TEXT NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL,
processed_at REAL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
"""
)
content.commit()
content.close()
search = connect(SEARCH_DB)
search.executescript(
"""
CREATE TABLE IF NOT EXISTS documents (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
source_published_at REAL NOT NULL,
indexed_at REAL NOT NULL,
event_id TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS processed_events (
event_id TEXT PRIMARY KEY NOT NULL,
processed_at REAL NOT NULL
);
"""
)
search.commit()
search.close()
if not quiet:
print("Initialized content.db and search.db.")
def reset():
content = connect(CONTENT_DB)
content.execute("DELETE FROM outbox")
content.execute("DELETE FROM content")
content.commit()
content.close()
search = connect(SEARCH_DB)
search.execute("DELETE FROM processed_events")
search.execute("DELETE FROM documents")
search.commit()
search.close()
print("Reset content.db and search.db.")
def usage(message=None):
if message:
print(message)
print(
"Usage: search_pipeline.py "
"init | reset | status | search TERM | "
"naive-publish SLUG TITLE [--fail] | "
"reliable-publish SLUG TITLE | work [--fail-after-index]"
)
sys.exit(2)
def main():
if len(sys.argv) < 2:
usage()
command = sys.argv[1]
if command == "init" and len(sys.argv) == 2:
initialize()
return
initialize(quiet=True)
if command == "reset" and len(sys.argv) == 2:
reset()
else:
usage(f"Unknown command or invalid arguments: {command}")
if __name__ == "__main__":
main()Start the simulator
The script now defines the schemas. The init route materializes them inside two database files.
The first run can take a few minutes while Docker downloads the pinned python:3.14.7-slim image. The command exits after initialization finishes.
Before you run it, which two database files do you expect the container to create?
- Initialize the simulator through Docker by running this command in the integrated terminal:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py initWhat Does This Command Do?
- The --rm flag removes the temporary container after the script exits.
- The -v flag mounts your current folder at /app inside the container.
- The -w flag makes /app the container's working directory.
- The pinned python:3.14.7-slim image runs the init route from your mounted script.
You should see Initialized content.db and search.db. in the terminal.
- Return to the Explorer sidebar in Visual Studio Code.
- Confirm that content.db appears beside search_pipeline.py.
- Confirm that search.db appears beside search_pipeline.py.
That is the setup complete. Your simulator now has two initialized stores plus a repeatable container command.
Which Database Is Authoritative?
The content.db file is the source of truth. Its empty content and outbox tables are ready for publications plus durable indexing work.
The search.db file is a disposable read model. Its empty documents and processed_events tables can be rebuilt from authoritative data as the design evolves.
Did the Databases Fail to Appear?
Confirm that Docker Desktop is running. Check that the integrated terminal is inside nextwork-search-design before retrying the command.
Compare your script with the full-code tab if Python reports a syntax problem. Use this prompt for environment-specific help: help me debug why the Docker initialization command did not create my SQLite files
Your local boundary is ready. Next, you will publish through a naive dual write and watch a project disappear from search.
Break the Naive Dual Write
Your SQLite source database and search read model are ready. You can now test what happens when one publication must reach both stores.
A dual write commits related data to separate systems. A crash between those commits can preserve the project in the source of truth while leaving it permanently absent from search.
In this step, get ready to:
- Add the naive publication path and search query.
- Prove that both writes succeed during the happy path.
- Trigger a crash between the writes and observe the missing search result.
Add the naive publish and search paths
The naive path treats the content commit and the search update as separate operations. The new functions make that boundary visible inside the simulator.
- In search_pipeline.py, find def usage(message=None):.
- Paste the first part of index_document() directly above that function:
def index_document(event_id, slug, title, published_at):
search = connect(SEARCH_DB)
duplicate = search.execute(
"SELECT event_id FROM processed_events WHERE event_id = ?",
(event_id,),
).fetchone()
if duplicate:
search.close()
print(f"Duplicate event ignored: {event_id}")
returnWhat does this code do?
- The function connects to SEARCH_DB because this operation updates the search read model.
- The lookup checks whether processed_events already contains the event ID.
- The early return prevents an event from being applied twice.
- Continue index_document() by pasting this code immediately below its return line:
indexed_at = time.time()
search.execute(
"""
INSERT INTO documents (
slug, title, source_published_at, indexed_at, event_id
) VALUES (?, ?, ?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
source_published_at = excluded.source_published_at,
indexed_at = excluded.indexed_at,
event_id = excluded.event_id
""",
(slug, title, published_at, indexed_at, event_id),
)
search.execute(
"INSERT INTO processed_events (event_id, processed_at) VALUES (?, ?)",
(event_id, indexed_at),
)
search.commit()
search.close()
lag = indexed_at - published_at
print(f"Indexed {slug}; publish-to-search lag was {lag:.3f}s.")How does indexing work?
- The upsert creates a search document for a new slug.
- The conflict clause replaces the indexed fields when the slug already exists.
- The processed event record gives the search-side consumer an idempotency key.
- The final calculation reports how long publication took to reach search.
- Save search_pipeline.py.
- Check that the indexing function has not broken the existing simulator by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py resetWhat does this check prove?
The pinned Python image loads the edited script before routing the reset command. A successful reset confirms that the file still parses.
You should see Reset content.db and search.db. in the terminal. Both databases are now empty.
Does the reset fail?
Check that the second code block remains indented inside index_document(). Its first line must align with the duplicate assignment.
Compare every parenthesis with the code above if Python reports a syntax problem.
help me debug the index_document function in my search pipeline simulator
The indexing function can update search. The publication function now shows how a direct call to it creates a fragile two-commit sequence.
- Paste naive_publish() between index_document() and usage() in search_pipeline.py:
def naive_publish(slug, title, fail_after_content=False):
published_at = time.time()
content = connect(CONTENT_DB)
content.execute(
"""
INSERT INTO content (slug, title, published_at) VALUES (?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
published_at = excluded.published_at
""",
(slug, title, published_at),
)
content.commit()
content.close()
print(f"Committed {slug} to the content database.")
if fail_after_content:
print("Simulated crash before the search-index write.")
return
event_id = f"naive:{slug}:{time.time_ns()}"
index_document(event_id, slug, title, published_at)Where is the consistency gap?
- The function upserts the project in CONTENT_DB first.
- The call to content.commit() makes that source record durable.
- The failure branch exits before index_document() can update search.
- The two stores therefore have no shared transaction.
- Paste search_projects() directly below naive_publish() in search_pipeline.py:
def search_projects(term):
search = connect(SEARCH_DB)
pattern = f"%{term}%"
rows = search.execute(
"""
SELECT slug, title
FROM documents
WHERE title LIKE ? OR slug LIKE ?
ORDER BY title
""",
(pattern, pattern),
).fetchall()
search.close()
if not rows:
print(f"No searchable projects matched '{term}'.")
return
for slug, title in rows:
print(f"{slug} | {title}")How does local search work?
- The pattern wraps the search term in % characters so LIKE can match text around it.
- The query reads only from the documents table in SEARCH_DB.
- An empty result produces a clear message for the designed failure later.
- Save search_pipeline.py.
- Check that both new functions parse by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py resetWhat does this second check prove?
The container loads both new functions before it resets the databases. Seeing the reset confirmation means their syntax is valid.
You should see Reset content.db and search.db. again. This clean state becomes the starting point for the happy path.
- In main(), find the branch that calls reset().
- Paste these command routes immediately below that branch:
elif command == "search" and len(sys.argv) == 3:
search_projects(sys.argv[2])
elif command == "naive-publish" and len(sys.argv) in (4, 5):
fail = len(sys.argv) == 5 and sys.argv[4] == "--fail"
if len(sys.argv) == 5 and not fail:
usage("The only optional naive-publish flag is --fail.")
naive_publish(sys.argv[2], sys.argv[3], fail)How are the commands routed?
- The search branch passes one search term to search_projects().
- The naive-publish branch passes the slug to naive_publish().
- The same branch passes the title as a separate argument.
- The optional --fail flag activates the simulated crash.
- Save search_pipeline.py.
✔️ Awesome, I've got everything!
Your naive publication and search commands are ready. Keep search_pipeline.py saved before testing them.
ⓧ I'd like to double check the full code
Compare your cumulative search_pipeline.py with this version:
import sqlite3
import sys
import time
CONTENT_DB = "content.db"
SEARCH_DB = "search.db"
def connect(path):
return sqlite3.connect(path, autocommit=False)
def initialize(quiet=False):
content = connect(CONTENT_DB)
content.executescript(
"""
CREATE TABLE IF NOT EXISTS content (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS outbox (
event_id TEXT PRIMARY KEY NOT NULL,
slug TEXT NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL,
processed_at REAL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
"""
)
content.commit()
content.close()
search = connect(SEARCH_DB)
search.executescript(
"""
CREATE TABLE IF NOT EXISTS documents (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
source_published_at REAL NOT NULL,
indexed_at REAL NOT NULL,
event_id TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS processed_events (
event_id TEXT PRIMARY KEY NOT NULL,
processed_at REAL NOT NULL
);
"""
)
search.commit()
search.close()
if not quiet:
print("Initialized content.db and search.db.")
def reset():
content = connect(CONTENT_DB)
content.execute("DELETE FROM outbox")
content.execute("DELETE FROM content")
content.commit()
content.close()
search = connect(SEARCH_DB)
search.execute("DELETE FROM processed_events")
search.execute("DELETE FROM documents")
search.commit()
search.close()
print("Reset content.db and search.db.")
def index_document(event_id, slug, title, published_at):
search = connect(SEARCH_DB)
duplicate = search.execute(
"SELECT event_id FROM processed_events WHERE event_id = ?",
(event_id,),
).fetchone()
if duplicate:
search.close()
print(f"Duplicate event ignored: {event_id}")
return
indexed_at = time.time()
search.execute(
"""
INSERT INTO documents (
slug, title, source_published_at, indexed_at, event_id
) VALUES (?, ?, ?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
source_published_at = excluded.source_published_at,
indexed_at = excluded.indexed_at,
event_id = excluded.event_id
""",
(slug, title, published_at, indexed_at, event_id),
)
search.execute(
"INSERT INTO processed_events (event_id, processed_at) VALUES (?, ?)",
(event_id, indexed_at),
)
search.commit()
search.close()
lag = indexed_at - published_at
print(f"Indexed {slug}; publish-to-search lag was {lag:.3f}s.")
def naive_publish(slug, title, fail_after_content=False):
published_at = time.time()
content = connect(CONTENT_DB)
content.execute(
"""
INSERT INTO content (slug, title, published_at) VALUES (?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
published_at = excluded.published_at
""",
(slug, title, published_at),
)
content.commit()
content.close()
print(f"Committed {slug} to the content database.")
if fail_after_content:
print("Simulated crash before the search-index write.")
return
event_id = f"naive:{slug}:{time.time_ns()}"
index_document(event_id, slug, title, published_at)
def search_projects(term):
search = connect(SEARCH_DB)
pattern = f"%{term}%"
rows = search.execute(
"""
SELECT slug, title
FROM documents
WHERE title LIKE ? OR slug LIKE ?
ORDER BY title
""",
(pattern, pattern),
).fetchall()
search.close()
if not rows:
print(f"No searchable projects matched '{term}'.")
return
for slug, title in rows:
print(f"{slug} | {title}")
def usage(message=None):
if message:
print(message)
print(
"Usage: search_pipeline.py "
"init | reset | status | search TERM | "
"naive-publish SLUG TITLE [--fail] | "
"reliable-publish SLUG TITLE | work [--fail-after-index]"
)
sys.exit(2)
def main():
if len(sys.argv) < 2:
usage()
command = sys.argv[1]
if command == "init" and len(sys.argv) == 2:
initialize()
return
initialize(quiet=True)
if command == "reset" and len(sys.argv) == 2:
reset()
elif command == "search" and len(sys.argv) == 3:
search_projects(sys.argv[2])
elif command == "naive-publish" and len(sys.argv) in (4, 5):
fail = len(sys.argv) == 5 and sys.argv[4] == "--fail"
if len(sys.argv) == 5 and not fail:
usage("The only optional naive-publish flag is --fail.")
naive_publish(sys.argv[2], sys.argv[3], fail)
else:
usage(f"Unknown command or invalid arguments: {command}")
if __name__ == "__main__":
main()Prove the happy path
A healthy run shows why direct dual writes can look convincing. The content record reaches the source database before the same command successfully indexes it.
- Publish Search System Basics without a simulated failure by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py naive-publish search-basics "Search System Basics"What does this publish command do?
The command runs naive_publish() without --fail. The function commits the source record before calling the search indexing function.
You should see one line confirming the content commit. A second line reports the publish-to-search lag.
- Before you search, consider whether the successful indexing message means the project is queryable.
- Search for the project by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py search SearchWhat does this search check?
The command queries only search.db. It does not consult the authoritative content table.
That separation mirrors a system where learner search traffic uses a dedicated read model.
You should see search-basics | Search System Basics. The direct dual write appears to work when neither operation is interrupted.
Is the happy-path result missing?
Confirm that the publish command printed both the content commit and indexing confirmation. A missing indexing confirmation points to the placement or indentation of index_document().
Check that the search route passes sys.argv[2] to search_projects().
help me debug why the happy-path project is missing from search
Trigger the dual-write failure
The happy path depends on both independent writes completing in order. You will now interrupt that sequence after the durable content commit.
- Reset both databases before the failure test by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py resetWhy reset again?
The reset removes the successful search-basics record from both stores. The next result therefore comes only from the failed publication.
You should see Reset content.db and search.db. in the terminal.
- Before you run the failed publish, predict which store will retain the project after the simulated crash.
- Publish Design Search Systems with the failure enabled by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py naive-publish search-design "Design Search Systems" --failWhat just failed?
The command commits search-design to content.db first. The --fail branch then returns before the search write.
You should see Committed search-design to the content database. followed by Simulated crash before the search-index write..
- Before you search, decide whether a durable content record is enough to make the project visible through the search read model.
- Search for the failed publication by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py search SearchWhat does the empty result prove?
The search command reads only from search.db. That database received no document for search-design.
You should see No searchable projects matched 'Search'. even though the project remains committed in the source database.
Why is a client retry incomplete?
A client that loses the original response cannot know whether the source commit completed. Repeating the request can create another attempt without recording which search update remains unfinished.
The system has no durable evidence that connects the committed project to its missing search write. The inconsistency can survive long after the original request disappears.
You have reproduced the exact gap that makes direct dual writes unsafe. Next, you will add a durable handoff that survives the same interruption.
Commit Publications with an Outbox
Your failed dual write proved that a published project can survive in the content database while remaining invisible to learners. Retrying an uncertain request cannot guarantee that the missing search write gets repaired.
A transactional outbox creates a durable handoff inside the source of truth. The project record shares one SQLite transaction with an indexing event. Search visibility then follows through eventual consistency.
In this step, get ready to:
- Commit each reliable publication with its outbox event.
- Add a status view across the content store and search read model.
- Prove that durable publication can precede search visibility.
Commit the project and event atomically
The reliable path uses one content-database connection for both writes. A single commit makes the project record and its indexing intent durable together.
- In search_pipeline.py, locate the naive_publish() function.
- Insert the reliable publication function directly below naive_publish() by pasting this code:
def reliable_publish(slug, title):
published_at = time.time()
event_id = f"publish:{slug}:{time.time_ns()}"
content = connect(CONTENT_DB)
try:
content.execute(
"""
INSERT INTO content (slug, title, published_at) VALUES (?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
published_at = excluded.published_at
""",
(slug, title, published_at),
)
content.execute(
"""
INSERT INTO outbox (
event_id, slug, title, published_at
) VALUES (?, ?, ?, ?)
""",
(event_id, slug, title, published_at),
)
content.commit()
except Exception:
content.rollback()
raise
finally:
content.close()What does this code do?
- The event ID combines the project slug with a nanosecond timestamp.
- Both inserts use the same content-database connection.
- The single commit() call makes both records durable as one unit.
- The exception path rolls back the whole transaction before closing the connection.
- Place this confirmation line directly below content.close() with four leading spaces:
print(f"Committed {slug} and outbox event {event_id} atomically.")Why print after the transaction?
The confirmation runs after the transaction has committed. It only reports success once the content row and outbox event are durable.
- Save search_pipeline.py.
- Confirm the updated script still initializes by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py initWhat does this check prove?
You should see Initialized content.db and search.db. in the terminal. The Docker container parsed the updated script successfully.
Does the script fail to initialize?
Check that except Exception: aligns with try:. Confirm that finally: uses the same indentation.
Make sure the confirmation line has four leading spaces. It belongs inside reliable_publish().
help me fix the indentation in my reliable publication function
The command router connects terminal input to the new publication path. The new branch accepts one slug and one title.
- In main(), locate the naive-publish branch.
- Insert this branch immediately below the call to naive_publish():
elif command == "reliable-publish" and len(sys.argv) == 4:
reliable_publish(sys.argv[2], sys.argv[3])How does this route work?
The router passes the third command-line argument into slug. It passes the fourth argument into title.
- Save search_pipeline.py.
- Publish the reliable project by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py reliable-publish event-driven-search "Event-Driven Search"What happened in the transaction?
You should see a confirmation that event-driven-search was committed with an outbox event. The generated event ID starts with publish:event-driven-search:.
The transaction inserted the publication into content.db. It also inserted the pending event into the outbox table.
Does reliable-publish fail?
Confirm that the new route appears before the final else: branch in main().
Check that the title remains inside double quotes in the terminal command. This keeps the three title words in one argument.
help me debug the reliable-publish command
Inspect the intentional consistency gap
A durable publication can exist before an asynchronous worker updates search. The status view compares both databases so you can see that boundary directly.
- In search_pipeline.py, locate the end of search_projects().
- Add the first half of status() directly below search_projects() by pasting this code:
def status():
content = connect(CONTENT_DB)
content_rows = content.execute(
"SELECT slug, title, published_at FROM content ORDER BY slug"
).fetchall()
pending_rows = content.execute(
"""
SELECT event_id, slug, attempts, published_at, last_error
FROM outbox
WHERE processed_at IS NULL
ORDER BY published_at, event_id
"""
).fetchall()
content.close()
search = connect(SEARCH_DB)
document_rows = search.execute(
"""
SELECT slug, title, source_published_at, indexed_at, event_id
FROM documents
ORDER BY slug
"""
).fetchall()
search.close()What does this half collect?
- The first query loads every authoritative content record.
- The second query loads outbox events whose processed_at value is still empty.
- The final query loads every document currently available in the search read model.
- Leave one blank line below search.close().
- Complete status() by pasting this code with four leading spaces:
documents = {row[0]: row[1:] for row in document_rows}
print(f"Content records: {len(content_rows)}")
print(f"Pending outbox events: {len(pending_rows)}")
print(f"Search documents: {len(document_rows)}")
for slug, title, published_at in content_rows:
document = documents.get(slug)
if document is None:
state = "NOT_SEARCHABLE"
lag_text = ""
else:
_, source_published_at, indexed_at, _ = document
if source_published_at == published_at:
state = "SEARCHABLE"
lag_text = f" lag={indexed_at - published_at:.3f}s"
else:
state = "STALE"
lag_text = ""
print(f" {slug}: {state}{lag_text} | {title}")
now = time.time()
for event_id, slug, attempts, published_at, last_error in pending_rows:
age = now - published_at
error_text = last_error if last_error else "none"
print(
f" pending {event_id} slug={slug} attempts={attempts} "
f"age={age:.3f}s last_error={error_text}"
)How are the states decided?
- A missing search document produces NOT_SEARCHABLE.
- Matching publication timestamps produce SEARCHABLE with a measured lag.
- Different publication timestamps produce STALE.
- Each pending event shows its age plus its attempt count.
- Save search_pipeline.py.
- Confirm the completed function parses by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py initWhat does this syntax check prove?
You should see Initialized content.db and search.db. again. The completed status function is valid Python.
- In main(), locate the reset command branch.
- Replace that branch with this expanded routing block:
if command == "reset" and len(sys.argv) == 2:
reset()
elif command == "status" and len(sys.argv) == 2:
status()What changed in the router?
The existing reset behavior remains in place. The next branch sends the status command to the new comparison function.
- Save search_pipeline.py.
✔️ Awesome, I've got everything!
Great. Your reliable publication path and cross-database status view are ready.
ⓧ I'd like to double check the full code
Compare your cumulative search_pipeline.py file with this reference:
import sqlite3
import sys
import time
CONTENT_DB = "content.db"
SEARCH_DB = "search.db"
def connect(path):
return sqlite3.connect(path, autocommit=False)
def initialize(quiet=False):
content = connect(CONTENT_DB)
content.executescript(
"""
CREATE TABLE IF NOT EXISTS content (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS outbox (
event_id TEXT PRIMARY KEY NOT NULL,
slug TEXT NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL,
processed_at REAL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
"""
)
content.commit()
content.close()
search = connect(SEARCH_DB)
search.executescript(
"""
CREATE TABLE IF NOT EXISTS documents (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
source_published_at REAL NOT NULL,
indexed_at REAL NOT NULL,
event_id TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS processed_events (
event_id TEXT PRIMARY KEY NOT NULL,
processed_at REAL NOT NULL
);
"""
)
search.commit()
search.close()
if not quiet:
print("Initialized content.db and search.db.")
def reset():
content = connect(CONTENT_DB)
content.execute("DELETE FROM outbox")
content.execute("DELETE FROM content")
content.commit()
content.close()
search = connect(SEARCH_DB)
search.execute("DELETE FROM processed_events")
search.execute("DELETE FROM documents")
search.commit()
search.close()
print("Reset content.db and search.db.")
def index_document(event_id, slug, title, published_at):
search = connect(SEARCH_DB)
duplicate = search.execute(
"SELECT event_id FROM processed_events WHERE event_id = ?",
(event_id,),
).fetchone()
if duplicate:
search.close()
print(f"Duplicate event ignored: {event_id}")
return
indexed_at = time.time()
search.execute(
"""
INSERT INTO documents (
slug, title, source_published_at, indexed_at, event_id
) VALUES (?, ?, ?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
source_published_at = excluded.source_published_at,
indexed_at = excluded.indexed_at,
event_id = excluded.event_id
""",
(slug, title, published_at, indexed_at, event_id),
)
search.execute(
"INSERT INTO processed_events (event_id, processed_at) VALUES (?, ?)",
(event_id, indexed_at),
)
search.commit()
search.close()
lag = indexed_at - published_at
print(f"Indexed {slug}; publish-to-search lag was {lag:.3f}s.")
def naive_publish(slug, title, fail_after_content=False):
published_at = time.time()
content = connect(CONTENT_DB)
content.execute(
"""
INSERT INTO content (slug, title, published_at) VALUES (?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
published_at = excluded.published_at
""",
(slug, title, published_at),
)
content.commit()
content.close()
print(f"Committed {slug} to the content database.")
if fail_after_content:
print("Simulated crash before the search-index write.")
return
event_id = f"naive:{slug}:{time.time_ns()}"
index_document(event_id, slug, title, published_at)
def reliable_publish(slug, title):
published_at = time.time()
event_id = f"publish:{slug}:{time.time_ns()}"
content = connect(CONTENT_DB)
try:
content.execute(
"""
INSERT INTO content (slug, title, published_at) VALUES (?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
published_at = excluded.published_at
""",
(slug, title, published_at),
)
content.execute(
"""
INSERT INTO outbox (
event_id, slug, title, published_at
) VALUES (?, ?, ?, ?)
""",
(event_id, slug, title, published_at),
)
content.commit()
except Exception:
content.rollback()
raise
finally:
content.close()
print(f"Committed {slug} and outbox event {event_id} atomically.")
def search_projects(term):
search = connect(SEARCH_DB)
pattern = f"%{term}%"
rows = search.execute(
"""
SELECT slug, title
FROM documents
WHERE title LIKE ? OR slug LIKE ?
ORDER BY title
""",
(pattern, pattern),
).fetchall()
search.close()
if not rows:
print(f"No searchable projects matched '{term}'.")
return
for slug, title in rows:
print(f"{slug} | {title}")
def status():
content = connect(CONTENT_DB)
content_rows = content.execute(
"SELECT slug, title, published_at FROM content ORDER BY slug"
).fetchall()
pending_rows = content.execute(
"""
SELECT event_id, slug, attempts, published_at, last_error
FROM outbox
WHERE processed_at IS NULL
ORDER BY published_at, event_id
"""
).fetchall()
content.close()
search = connect(SEARCH_DB)
document_rows = search.execute(
"""
SELECT slug, title, source_published_at, indexed_at, event_id
FROM documents
ORDER BY slug
"""
).fetchall()
search.close()
documents = {row[0]: row[1:] for row in document_rows}
print(f"Content records: {len(content_rows)}")
print(f"Pending outbox events: {len(pending_rows)}")
print(f"Search documents: {len(document_rows)}")
for slug, title, published_at in content_rows:
document = documents.get(slug)
if document is None:
state = "NOT_SEARCHABLE"
lag_text = ""
else:
_, source_published_at, indexed_at, _ = document
if source_published_at == published_at:
state = "SEARCHABLE"
lag_text = f" lag={indexed_at - published_at:.3f}s"
else:
state = "STALE"
lag_text = ""
print(f" {slug}: {state}{lag_text} | {title}")
now = time.time()
for event_id, slug, attempts, published_at, last_error in pending_rows:
age = now - published_at
error_text = last_error if last_error else "none"
print(
f" pending {event_id} slug={slug} attempts={attempts} "
f"age={age:.3f}s last_error={error_text}"
)
def usage(message=None):
if message:
print(message)
print(
"Usage: search_pipeline.py "
"init | reset | status | search TERM | "
"naive-publish SLUG TITLE [--fail] | "
"reliable-publish SLUG TITLE | work [--fail-after-index]"
)
sys.exit(2)
def main():
if len(sys.argv) < 2:
usage()
command = sys.argv[1]
if command == "init" and len(sys.argv) == 2:
initialize()
return
initialize(quiet=True)
if command == "reset" and len(sys.argv) == 2:
reset()
elif command == "status" and len(sys.argv) == 2:
status()
elif command == "search" and len(sys.argv) == 3:
search_projects(sys.argv[2])
elif command == "naive-publish" and len(sys.argv) in (4, 5):
fail = len(sys.argv) == 5 and sys.argv[4] == "--fail"
if len(sys.argv) == 5 and not fail:
usage("The only optional naive-publish flag is --fail.")
naive_publish(sys.argv[2], sys.argv[3], fail)
elif command == "reliable-publish" and len(sys.argv) == 4:
reliable_publish(sys.argv[2], sys.argv[3])
else:
usage(f"Unknown command or invalid arguments: {command}")
if __name__ == "__main__":
main()What should match?
The file now contains reliable_publish() before search_projects(). The new status() function appears before usage().
The router includes both status and reliable-publish branches.
Before you run the final check, do you expect the reliable project to be durable, searchable, or both?
- Inspect both database states by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py statusWhat should you see?
The summary reports Content records: 2. It also reports Pending outbox events: 1 plus Search documents: 0.
Both search-design and event-driven-search appear as NOT_SEARCHABLE. The pending event for event-driven-search shows attempts=0 because no worker has processed it.
That is the durable handoff working. Your reliable project is safely published while its search update waits in the outbox.
Does the pending count look wrong?
A count of zero usually means the reliable publication command did not run. Run it once with the exact slug from this step.
A count above one usually means the reliable publication command ran more than once. Resetting now would also erase the broken-path evidence from the previous step.
help me interpret the simulator status output
Your publication path now survives a missing search update. Next up, you will build the worker that consumes this durable handoff and recovers safely from a crash.
Replay Indexing Safely
Your reliable publication is durable in the transactional outbox. However, the project remains invisible to learners until a worker copies its search fields into the separate read model.
The risky moment comes after the search write succeeds but before the worker acknowledges the event. You will trigger that crash on purpose. You will then use idempotency to make the replay safe.
In this step, get ready to:
- Add a worker that selects the oldest pending event and tracks each attempt.
- Crash the worker after it updates search but before it acknowledges the event.
- Replay the event without creating a duplicate search-side record.
Add the worker and attempt tracking
The worker handles one pending event at a time. It increments the attempt count before touching search, which leaves an operational record even when processing stops halfway through.
- Switch back to search_pipeline.py in VS Code.
- Find def search_projects(term):.
- Insert the first half of process_one() directly above that function:
def process_one(fail_after_index=False):
content = connect(CONTENT_DB)
event = content.execute(
"""
SELECT event_id, slug, title, published_at
FROM outbox
WHERE processed_at IS NULL
ORDER BY published_at, event_id
LIMIT 1
"""
).fetchone()
if event is None:
content.close()
print("No pending outbox events.")
return
event_id, slug, title, published_at = event
content.execute(
"""
UPDATE outbox
SET attempts = attempts + 1, last_error = NULL
WHERE event_id = ?
""",
(event_id,),
)
content.commit()How does event selection work?
- The query selects the oldest row whose processed_at value is still empty.
- The LIMIT 1 clause keeps each worker run focused on one event.
- The attempt update commits before indexing starts. A crash therefore remains visible in the source database.
- Continue process_one() by adding this code directly below content.commit():
try:
index_document(event_id, slug, title, published_at)
if fail_after_index:
print("Simulated worker crash before the outbox acknowledgement.")
return
content.execute(
"""
UPDATE outbox
SET processed_at = ?, last_error = NULL
WHERE event_id = ?
""",
(time.time(), event_id),
)
content.commit()
print(f"Acknowledged outbox event {event_id}.")
except Exception as error:
content.execute(
"UPDATE outbox SET last_error = ? WHERE event_id = ?",
(str(error), event_id),
)
content.commit()
raise
finally:
content.close()What does the processing phase do?
- The worker passes the event to index_document(), which upserts the search document and records the event ID.
- The failure switch exits after the search-side commit. This deliberately leaves the outbox event pending.
- A successful run sets processed_at only after the search work finishes.
- A processing exception is stored in last_error before the exception continues upward.
The function now contains the full crash boundary. The command router still needs a path that can activate the normal run or the simulated failure.
- Find the reliable-publish branch inside main().
- Insert the worker branch directly after it and before the final else block:
elif command == "work" and len(sys.argv) in (2, 3):
fail = len(sys.argv) == 3 and sys.argv[2] == "--fail-after-index"
if len(sys.argv) == 3 and not fail:
usage("The only optional work flag is --fail-after-index.")
process_one(fail)How does the worker command behave?
The branch accepts a normal worker run or the exact failure switch. Any other optional value goes through usage() instead of silently changing worker behavior.
- Save search_pipeline.py.
Crash after the search-side effect
This run creates the dangerous partial state. Search accepts the event, while the source database receives no acknowledgement.
Before you run the worker, pause on this question: will the pending count fall after the simulated crash.
- Run the worker with the failure switch:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py work --fail-after-indexYou will see the project get indexed. You will then see Simulated worker crash before the outbox acknowledgement. The search-side effect has committed, but the source-side acknowledgement has not.
- Inspect both sides of the partial failure by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py statusYou will see event-driven-search reported as SEARCHABLE. You will also see Pending outbox events: 1 with the event attempt count increased to 1.
Why is this at-least-once delivery?
The pending event must be delivered again because the source database has no durable acknowledgement. This is at-least-once delivery, so the consumer must expect repeat attempts.
The search-side processed_events table stores each event ID after a successful write. That record turns the next attempt into a safe replay.
Worker command did not reach the simulated crash?
Check that both halves of process_one() remain indented inside the same function. Confirm that the work branch sits before the final else in main().
If the worker reports no pending events, use status to confirm that the reliable publication from the previous step still has an unprocessed outbox row.
Help me debug why the simulated indexing worker did not process the pending event.
Replay and acknowledge the event
The command help should describe the new worker path before you finish. This keeps the simulator's supported commands visible when someone supplies invalid input.
- Find def usage(message=None): in search_pipeline.py.
- Replace the complete function with this updated version:
def usage(message=None):
if message:
print(message)
print(
"Usage: search_pipeline.py "
"init | reset | status | search TERM | "
"naive-publish SLUG TITLE [--fail] | "
"reliable-publish SLUG TITLE | work [--fail-after-index]"
)
sys.exit(2)What changed in the command help?
The usage text now advertises work with its optional failure switch. The function still exits with a nonzero status when the command input is invalid.
- Save search_pipeline.py.
✔️ Awesome, I've got everything!
Great. Double check that search_pipeline.py is saved before you replay the event.
ⓧ I'd like to double check the full code
import sqlite3
import sys
import time
CONTENT_DB = "content.db"
SEARCH_DB = "search.db"
def connect(path):
return sqlite3.connect(path, autocommit=False)
def initialize(quiet=False):
content = connect(CONTENT_DB)
content.executescript(
"""
CREATE TABLE IF NOT EXISTS content (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS outbox (
event_id TEXT PRIMARY KEY NOT NULL,
slug TEXT NOT NULL,
title TEXT NOT NULL,
published_at REAL NOT NULL,
processed_at REAL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
"""
)
content.commit()
content.close()
search = connect(SEARCH_DB)
search.executescript(
"""
CREATE TABLE IF NOT EXISTS documents (
slug TEXT PRIMARY KEY NOT NULL,
title TEXT NOT NULL,
source_published_at REAL NOT NULL,
indexed_at REAL NOT NULL,
event_id TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS processed_events (
event_id TEXT PRIMARY KEY NOT NULL,
processed_at REAL NOT NULL
);
"""
)
search.commit()
search.close()
if not quiet:
print("Initialized content.db and search.db.")
def reset():
content = connect(CONTENT_DB)
content.execute("DELETE FROM outbox")
content.execute("DELETE FROM content")
content.commit()
content.close()
search = connect(SEARCH_DB)
search.execute("DELETE FROM processed_events")
search.execute("DELETE FROM documents")
search.commit()
search.close()
print("Reset content.db and search.db.")
def index_document(event_id, slug, title, published_at):
search = connect(SEARCH_DB)
duplicate = search.execute(
"SELECT event_id FROM processed_events WHERE event_id = ?",
(event_id,),
).fetchone()
if duplicate:
search.close()
print(f"Duplicate event ignored: {event_id}")
return
indexed_at = time.time()
search.execute(
"""
INSERT INTO documents (
slug, title, source_published_at, indexed_at, event_id
) VALUES (?, ?, ?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
source_published_at = excluded.source_published_at,
indexed_at = excluded.indexed_at,
event_id = excluded.event_id
""",
(slug, title, published_at, indexed_at, event_id),
)
search.execute(
"INSERT INTO processed_events (event_id, processed_at) VALUES (?, ?)",
(event_id, indexed_at),
)
search.commit()
search.close()
lag = indexed_at - published_at
print(f"Indexed {slug}; publish-to-search lag was {lag:.3f}s.")
def naive_publish(slug, title, fail_after_content=False):
published_at = time.time()
content = connect(CONTENT_DB)
content.execute(
"""
INSERT INTO content (slug, title, published_at) VALUES (?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
published_at = excluded.published_at
""",
(slug, title, published_at),
)
content.commit()
content.close()
print(f"Committed {slug} to the content database.")
if fail_after_content:
print("Simulated crash before the search-index write.")
return
event_id = f"naive:{slug}:{time.time_ns()}"
index_document(event_id, slug, title, published_at)
def reliable_publish(slug, title):
published_at = time.time()
event_id = f"publish:{slug}:{time.time_ns()}"
content = connect(CONTENT_DB)
try:
content.execute(
"""
INSERT INTO content (slug, title, published_at) VALUES (?, ?, ?)
ON CONFLICT(slug) DO UPDATE SET
title = excluded.title,
published_at = excluded.published_at
""",
(slug, title, published_at),
)
content.execute(
"""
INSERT INTO outbox (
event_id, slug, title, published_at
) VALUES (?, ?, ?, ?)
""",
(event_id, slug, title, published_at),
)
content.commit()
except Exception:
content.rollback()
raise
finally:
content.close()
print(f"Committed {slug} and outbox event {event_id} atomically.")
def process_one(fail_after_index=False):
content = connect(CONTENT_DB)
event = content.execute(
"""
SELECT event_id, slug, title, published_at
FROM outbox
WHERE processed_at IS NULL
ORDER BY published_at, event_id
LIMIT 1
"""
).fetchone()
if event is None:
content.close()
print("No pending outbox events.")
return
event_id, slug, title, published_at = event
content.execute(
"""
UPDATE outbox
SET attempts = attempts + 1, last_error = NULL
WHERE event_id = ?
""",
(event_id,),
)
content.commit()
try:
index_document(event_id, slug, title, published_at)
if fail_after_index:
print("Simulated worker crash before the outbox acknowledgement.")
return
content.execute(
"""
UPDATE outbox
SET processed_at = ?, last_error = NULL
WHERE event_id = ?
""",
(time.time(), event_id),
)
content.commit()
print(f"Acknowledged outbox event {event_id}.")
except Exception as error:
content.execute(
"UPDATE outbox SET last_error = ? WHERE event_id = ?",
(str(error), event_id),
)
content.commit()
raise
finally:
content.close()
def search_projects(term):
search = connect(SEARCH_DB)
pattern = f"%{term}%"
rows = search.execute(
"""
SELECT slug, title
FROM documents
WHERE title LIKE ? OR slug LIKE ?
ORDER BY title
""",
(pattern, pattern),
).fetchall()
search.close()
if not rows:
print(f"No searchable projects matched '{term}'.")
return
for slug, title in rows:
print(f"{slug} | {title}")
def status():
content = connect(CONTENT_DB)
content_rows = content.execute(
"SELECT slug, title, published_at FROM content ORDER BY slug"
).fetchall()
pending_rows = content.execute(
"""
SELECT event_id, slug, attempts, published_at, last_error
FROM outbox
WHERE processed_at IS NULL
ORDER BY published_at, event_id
"""
).fetchall()
content.close()
search = connect(SEARCH_DB)
document_rows = search.execute(
"""
SELECT slug, title, source_published_at, indexed_at, event_id
FROM documents
ORDER BY slug
"""
).fetchall()
search.close()
documents = {row[0]: row[1:] for row in document_rows}
print(f"Content records: {len(content_rows)}")
print(f"Pending outbox events: {len(pending_rows)}")
print(f"Search documents: {len(document_rows)}")
for slug, title, published_at in content_rows:
document = documents.get(slug)
if document is None:
state = "NOT_SEARCHABLE"
lag_text = ""
else:
_, source_published_at, indexed_at, _ = document
if source_published_at == published_at:
state = "SEARCHABLE"
lag_text = f" lag={indexed_at - published_at:.3f}s"
else:
state = "STALE"
lag_text = ""
print(f" {slug}: {state}{lag_text} | {title}")
now = time.time()
for event_id, slug, attempts, published_at, last_error in pending_rows:
age = now - published_at
error_text = last_error if last_error else "none"
print(
f" pending {event_id} slug={slug} attempts={attempts} "
f"age={age:.3f}s last_error={error_text}"
)
def usage(message=None):
if message:
print(message)
print(
"Usage: search_pipeline.py "
"init | reset | status | search TERM | "
"naive-publish SLUG TITLE [--fail] | "
"reliable-publish SLUG TITLE | work [--fail-after-index]"
)
sys.exit(2)
def main():
if len(sys.argv) < 2:
usage()
command = sys.argv[1]
if command == "init" and len(sys.argv) == 2:
initialize()
return
initialize(quiet=True)
if command == "reset" and len(sys.argv) == 2:
reset()
elif command == "status" and len(sys.argv) == 2:
status()
elif command == "search" and len(sys.argv) == 3:
search_projects(sys.argv[2])
elif command == "naive-publish" and len(sys.argv) in (4, 5):
fail = len(sys.argv) == 5 and sys.argv[4] == "--fail"
if len(sys.argv) == 5 and not fail:
usage("The only optional naive-publish flag is --fail.")
naive_publish(sys.argv[2], sys.argv[3], fail)
elif command == "reliable-publish" and len(sys.argv) == 4:
reliable_publish(sys.argv[2], sys.argv[3])
elif command == "work" and len(sys.argv) in (2, 3):
fail = len(sys.argv) == 3 and sys.argv[2] == "--fail-after-index"
if len(sys.argv) == 3 and not fail:
usage("The only optional work flag is --fail-after-index.")
process_one(fail)
else:
usage(f"Unknown command or invalid arguments: {command}")
if __name__ == "__main__":
main()Before you replay, pause on this question: do you expect another search document or a duplicate warning.
- Replay the pending event by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py workYou will see Duplicate event ignored: followed by the original event ID. You will then see the acknowledgement message because the replay safely completes the source-side work.
- Confirm that the outbox has drained by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py statusYou will see Pending outbox events: 0. The event-driven-search project remains SEARCHABLE, while the deliberately broken search-design record remains NOT_SEARCHABLE.
Before the final search, predict whether replay changed the visible title or preserved the existing document.
- Search for the recovered project by running:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py search Event-DrivenYou will see event-driven-search | Event-Driven Search. One document is searchable because the same processed event ID was accepted exactly once.
Replay did not show a duplicate warning?
Confirm that index_document() checks processed_events before writing the document. The first failed worker run must reach the search commit before returning.
If the event remains pending after the normal replay, check that the acknowledgement update uses the same event_id selected at the start of process_one().
Help me trace why my replay was not ignored or acknowledged safely.
That is the hardest failure boundary handled. Your simulator now demonstrates safe at-least-once processing, so the next step can turn its behavior into an interview-ready architecture explanation.
Document the Architecture
Your simulator now proves at-least-once delivery with idempotent replay. You have seen the worker recover from a crash without duplicating the search-side event.
A complete system design answer makes the requirements, consistency boundary, scale path, failure recovery, and operational signals explicit. You will capture that reasoning in a document you can present without consulting the Python code.
In this step, get ready to:
- Document the requirements and publish-to-search architecture.
- Defend the consistency model and production trade-offs.
- Validate the final state and rehearse the design answer.
Capture requirements and the architecture
The architecture file turns the behavior you observed in SQLite into a design that another engineer can review. It also makes clear that NextWork is a hypothetical example in this exercise.
- In the Explorer sidebar from earlier, use the new-file control beside the nextwork-search-design folder.
- Name the file architecture.md.
- Add the project context and requirements by pasting this first section:
# Hypothetical NextWork Publish-to-Search Design
This is an interview exercise inspired by NextWork's public project catalog. It does not describe NextWork's private production architecture.
## Requirements
Functional requirements:
- An author can publish or update a learning project.
- A learner can search published projects by relevant text and filters.
- A failed indexing attempt can be retried without losing the publication.
- An operator can identify missing, stale, delayed, and failed index updates.
Non-functional requirements:
- The content database is the source of truth.
- Publication must not be rolled back merely because the search system is temporarily unavailable.
- Search visibility should meet a product-defined publish-to-searchable objective.
- Duplicate delivery must be safe.
- Search and write capacity must scale independently.What do these requirements establish?
The functional requirements describe what authors, learners, and operators need from the system. They cover publication, discovery, retries, and investigation.
The non-functional requirements establish the consistency and scaling boundaries. The content database remains authoritative while search can recover independently.
- Save architecture.md.
- Confirm architecture.md appears beneath search_pipeline.py in the Explorer sidebar.
Architecture file missing?
Check that you created architecture.md inside nextwork-search-design. A file created beside that folder belongs to a different workspace location.
Ask for help if the file is not appearing in the Explorer sidebar:
The requirements define the contract. The next section shows how a publication travels from the author to a learner's search results.
- In architecture.md, place your cursor after the final non-functional requirement.
- Add the high-level flow by pasting this section:
## High-level architecture
Author
|
v
Publish API
|
+---- one database transaction ----+
| |
v v
Content table Outbox table
|
v
Indexing workers
| |
v v
Search index Dead-letter path
|
v
Search API
|
v
LearnerHow does the flow work?
The Publish API commits the content row and outbox event through one source-database transaction. That transaction is the strong consistency boundary.
Indexing workers move the event into the search index asynchronously. The Search API then serves the read model to learners.
- Save architecture.md.
- Confirm the editor shows the full path from Author to Learner.
Diagram difficult to read?
Keep the spaces at the start of every diagram line. Markdown uses that indentation to preserve the columns and arrows.
Ask for help checking the spacing:
Defend the trade-offs
The diagram shows the components. An interview answer also needs to explain why each boundary exists and how a request behaves across it.
- Continue architecture.md by pasting the write and search paths below the diagram:
## Write path
1. Validate and authorize the publication request.
2. Upsert the project and insert an outbox event in one content-database transaction.
3. Return publication success after the durable transaction commits.
4. Let an indexing worker claim the outbox event.
5. Transform search-relevant fields and upsert the search document using the event ID as an idempotency key.
6. Wait for the required search visibility guarantee, then acknowledge the outbox event.
For a search engine such as Elasticsearch, `refresh=wait_for` can make the indexing request wait until a refresh makes the change visible. It adds response latency, while `refresh=true` forces a refresh and can increase indexing and search cost. The product must decide whether every publish needs synchronous visibility confirmation or whether bounded eventual consistency is sufficient.
## Search path
1. The learner sends a query to the Search API.
2. The Search API queries only the search index for text matching, filters, ranking, and pagination.
3. The result returns project identifiers and display fields stored in the index.
4. The application may fetch authoritative detail from the content service when the learner opens a project.
The local simulator uses a second SQLite database and `LIKE` only to demonstrate consistency behavior. A production search engine would add analyzers, ranking, typo tolerance, facets, and horizontal scaling.Why keep search outside the transaction?
The source transaction protects the project row and its indexing intent. The search system stays outside that transaction because it is a separate system with its own availability and scaling behavior.
In Elasticsearch, refresh=wait_for waits for a refresh to expose the change. refresh=true forces a refresh with extra indexing and search cost.
- Save architecture.md.
- Confirm the write path ends with the outbox acknowledgement after search visibility is established.
Write and search paths mixed together?
Check that the write path ends before the Search path heading. The Search API queries the search index without joining the publication transaction.
Ask for help separating the paths:
The next section names the guarantees behind that flow. It also turns each interruption point into an observable result with a recovery path.
- Add the consistency contract and failure table below the search path by pasting:
## Consistency contract
- Content publication is strongly consistent inside the content database transaction.
- Propagation from the outbox to search is eventually consistent.
- The worker provides at-least-once delivery.
- Search-side idempotency makes replay safe.
- The publish-to-searchable objective bounds how long eventual consistency is allowed to last.
- For author read-your-writes, the authoring UI can read the project directly from the content database and show an `Indexing` state until global search confirms visibility.
## Failure modes
| Failure | Observable result | Recovery |
| --- | --- | --- |
| API fails before commit | No content and no outbox event | Client retries with a request id |
| API fails after commit | Content and outbox event remain durable | Client reads publication status or retries idempotently |
| Worker fails before search write | Outbox remains pending | Retry with backoff |
| Worker fails after search write | Document exists and event remains pending | Replay; consumer ignores the processed event ID |
| Search index is unavailable | Outbox age and pending count grow | Backoff, alert, and later drain |
| Event repeatedly fails transformation | One event blocks or churns | Move it to a dead-letter path and continue safely |
| Search mapping change breaks old events | Indexing failures rise | Version events and run a compatible reindex path |
| Update events arrive out of order | Search may show stale content | Carry a content version and reject older versions |Where is consistency strong?
The content row and outbox event commit together. That gives publication a durable handoff even when search is unavailable.
Search visibility uses eventual consistency. At-least-once delivery permits replay while search-side idempotency prevents the replay from duplicating the event.
- Save architecture.md.
- Confirm every row in the failure table names a visible symptom and a recovery action.
Failure table columns misaligned?
Check that each table row begins and ends with a vertical bar. Keep the separator row directly beneath the three column headings.
Ask for help repairing the Markdown table:
The final sections explain how the design grows and how operators know when search is falling behind. They also condense the design into an interview-ready summary.
- Complete architecture.md by pasting the scaling, observability, and interview sections:
## Scaling path
- Partition content by project identifier only when write volume requires it.
- Run multiple indexing workers and assign events so updates for one project preserve order.
- Batch index requests when throughput matters, while tracking each event outcome.
- Scale the search cluster independently for query traffic and index volume.
- Rebuild the search index from the source database or replayable change history because search is not the source of truth.
- Use CDC instead of application polling when the database exposes a reliable ordered change stream and the organization can operate that pipeline.
## Observability
Track:
- Publish request success and latency
- Outbox pending count
- Oldest pending event age
- Publish-to-searchable latency
- Worker attempt and failure counts
- Dead-letter count
- Search indexing throughput and errors
- Search zero-result rate
- Stale document count from reconciliation jobs
Alert first on the oldest pending event age because it directly measures how long users may be missing newly published content.
## Interview summary
I would keep the content database authoritative and use a dedicated search index as a read model. The naive synchronous dual write is unsafe because either side can succeed alone. I would commit the project and an outbox event atomically, then let replay-safe workers update search. This gives durable publication, independent scaling, and at-least-once delivery with idempotent consumption. Search visibility remains eventually consistent, so I would define a publish-to-searchable objective, monitor outbox age, and use a visibility acknowledgement when the product truly requires the publish call to return only after the document is searchable. I would also support dead-letter replay, reconciliation, and full reindexing from the source of truth.How does this become production-ready?
The scaling path preserves per-project ordering while allowing workers and search capacity to grow independently. Change data capture offers a production alternative when the database exposes a reliable ordered stream.
Observability turns consistency into measurable behavior. Oldest pending event age directly shows how long new publications may remain absent from learner searches.
- Save architecture.md.
- Confirm the final heading is Interview summary.
Missing part of the final section?
Check that Scaling path, Observability, and Interview summary each appear once. The oldest pending event age sentence belongs after the observability list.
Ask for help comparing the sections:
✔️ Awesome, I've got everything!
Great. Your architecture document now covers requirements, data flow, consistency, failure recovery, scaling, and observability.
ⓧ I'd like to double check the full code
Compare your saved architecture.md with this complete reference.
# Hypothetical NextWork Publish-to-Search Design
This is an interview exercise inspired by NextWork's public project catalog. It does not describe NextWork's private production architecture.
## Requirements
Functional requirements:
- An author can publish or update a learning project.
- A learner can search published projects by relevant text and filters.
- A failed indexing attempt can be retried without losing the publication.
- An operator can identify missing, stale, delayed, and failed index updates.
Non-functional requirements:
- The content database is the source of truth.
- Publication must not be rolled back merely because the search system is temporarily unavailable.
- Search visibility should meet a product-defined publish-to-searchable objective.
- Duplicate delivery must be safe.
- Search and write capacity must scale independently.
## High-level architecture
Author
|
v
Publish API
|
+---- one database transaction ----+
| |
v v
Content table Outbox table
|
v
Indexing workers
| |
v v
Search index Dead-letter path
|
v
Search API
|
v
Learner
## Write path
1. Validate and authorize the publication request.
2. Upsert the project and insert an outbox event in one content-database transaction.
3. Return publication success after the durable transaction commits.
4. Let an indexing worker claim the outbox event.
5. Transform search-relevant fields and upsert the search document using the event ID as an idempotency key.
6. Wait for the required search visibility guarantee, then acknowledge the outbox event.
For a search engine such as Elasticsearch, `refresh=wait_for` can make the indexing request wait until a refresh makes the change visible. It adds response latency, while `refresh=true` forces a refresh and can increase indexing and search cost. The product must decide whether every publish needs synchronous visibility confirmation or whether bounded eventual consistency is sufficient.
## Search path
1. The learner sends a query to the Search API.
2. The Search API queries only the search index for text matching, filters, ranking, and pagination.
3. The result returns project identifiers and display fields stored in the index.
4. The application may fetch authoritative detail from the content service when the learner opens a project.
The local simulator uses a second SQLite database and `LIKE` only to demonstrate consistency behavior. A production search engine would add analyzers, ranking, typo tolerance, facets, and horizontal scaling.
## Consistency contract
- Content publication is strongly consistent inside the content database transaction.
- Propagation from the outbox to search is eventually consistent.
- The worker provides at-least-once delivery.
- Search-side idempotency makes replay safe.
- The publish-to-searchable objective bounds how long eventual consistency is allowed to last.
- For author read-your-writes, the authoring UI can read the project directly from the content database and show an `Indexing` state until global search confirms visibility.
## Failure modes
| Failure | Observable result | Recovery |
| --- | --- | --- |
| API fails before commit | No content and no outbox event | Client retries with a request id |
| API fails after commit | Content and outbox event remain durable | Client reads publication status or retries idempotently |
| Worker fails before search write | Outbox remains pending | Retry with backoff |
| Worker fails after search write | Document exists and event remains pending | Replay; consumer ignores the processed event ID |
| Search index is unavailable | Outbox age and pending count grow | Backoff, alert, and later drain |
| Event repeatedly fails transformation | One event blocks or churns | Move it to a dead-letter path and continue safely |
| Search mapping change breaks old events | Indexing failures rise | Version events and run a compatible reindex path |
| Update events arrive out of order | Search may show stale content | Carry a content version and reject older versions |
## Scaling path
- Partition content by project identifier only when write volume requires it.
- Run multiple indexing workers and assign events so updates for one project preserve order.
- Batch index requests when throughput matters, while tracking each event outcome.
- Scale the search cluster independently for query traffic and index volume.
- Rebuild the search index from the source database or replayable change history because search is not the source of truth.
- Use CDC instead of application polling when the database exposes a reliable ordered change stream and the organization can operate that pipeline.
## Observability
Track:
- Publish request success and latency
- Outbox pending count
- Oldest pending event age
- Publish-to-searchable latency
- Worker attempt and failure counts
- Dead-letter count
- Search indexing throughput and errors
- Search zero-result rate
- Stale document count from reconciliation jobs
Alert first on the oldest pending event age because it directly measures how long users may be missing newly published content.
## Interview summary
I would keep the content database authoritative and use a dedicated search index as a read model. The naive synchronous dual write is unsafe because either side can succeed alone. I would commit the project and an outbox event atomically, then let replay-safe workers update search. This gives durable publication, independent scaling, and at-least-once delivery with idempotent consumption. Search visibility remains eventually consistent, so I would define a publish-to-searchable objective, monitor outbox age, and use a visibility acknowledgement when the product truly requires the publish call to return only after the document is searchable. I would also support dead-letter replay, reconciliation, and full reindexing from the source of truth.What does this reference contain?
This is the complete architecture document for the simulator. Use it to compare headings, lists, the failure table, and the final summary.
Run the acceptance scenario and rehearse
The document now describes the intended design. The simulator state provides the evidence that your explanation matches the behavior you built.
Before you run the final check, which project do you expect to remain missing from search?
- Check the final database and search state by running this command in the integrated terminal from earlier:
docker run --rm -v "$PWD":/app -w /app python:3.14.7-slim python search_pipeline.py statusWhat does this check prove?
You should see Pending outbox events: 0. The worker has acknowledged the reliable publication event.
You should also see search-design as NOT_SEARCHABLE and event-driven-search as SEARCHABLE. That contrast preserves the failed dual write beside the repaired outbox path.
Final status different?
If an outbox event is still pending, switch back to the integrated terminal and run the normal worker command from the previous step. If either project is missing from the content records, your databases no longer contain the acceptance scenario.
Confirm that Docker Desktop remains available if the container does not start. Ask for help diagnosing the current state:
That is the full consistency story working as designed. The durable path recovered while the deliberately broken path stayed visible as evidence.
Your final rehearsal should follow a stable order so the listener can connect requirements to mechanisms and operational behavior.
- Present the functional and non-functional requirements from architecture.md.
- State the traffic estimates and assumptions you would validate before sizing the production system.
- Define the publish and search APIs with their content and outbox data models.
- Walk through the high-level path from author publication to learner search.
- Deep dive into the transactional outbox, idempotent replay, and search visibility semantics.
- Explain the failure modes and their recovery paths.
- Close with the scaling path and observability signals.
- Give the five-minute explanation using architecture.md without looking at search_pipeline.py.
You now have an interview-ready architecture backed by a working failure and recovery demonstration. You can explain where the system is strongly consistent, where it is eventually consistent, and how operators prove recovery.
Secret mission
Run a Search-Indexing Incident Drill
Three durable publications are waiting while the indexing worker falls behind. Diagnose the growing backlog through its operational signals. Recover every event through replay before presenting the incident response you would give in a system design interview.
Clean Up Your Resources
Clean Up Your Resources
Everything from this project stays on your Mac with no cloud costs or paid persistent resources. Choose Keep, Pause, or Delete based on whether you want to rehearse the simulator again.
Resources you used:
- The nextwork-search-design project directory on your Mac.
- The search_pipeline.py simulator inside the project directory.
- The architecture.md interview guide inside the project directory.
- The content.db database containing source content plus processed outbox history.
- The search.db database containing indexed documents plus processed event IDs.
- The python:3.14.7-slim image stored by Docker Desktop.
Keep everything running
No action is required. Choose this option if you want to repeat the incident drill or rehearse your interview answer.
- Keep the nextwork-search-design directory to preserve the complete simulator.
- Keep content.db to preserve the source records plus acknowledged outbox history.
- Keep search.db to preserve the searchable documents plus idempotency records.
- Keep the python:3.14.7-slim image to avoid downloading the runtime during your next rehearsal.
- Remember that every simulator command used --rm. No project container remains after those commands.
Pause - I'll come back to this later
Pause the local tools while preserving the simulator state. Your files remain ready for another session.
- Quit Docker Desktop to stop its local engine.
- Quit Visual Studio Code after finishing your review.
- Leave the nextwork-search-design directory on your Mac.
Your two database files preserve the completed incident drill. Return to the same directory when you want to run the simulator again.
Delete - I don't want to use this again
Remove the local simulator when you no longer need its files. This choice becomes irreversible when Trash is emptied.
The project directory contains every script, document, and database created during the project. The downloaded runtime image remains separate inside Docker Desktop.
- Select Finder in the Dock.
- Enter nextwork-search-design in Finder's search field.
- Select the folder named nextwork-search-design in the results.
- Drag the nextwork-search-design folder to Trash in the Dock.
- Open Trash from the Dock.
That's the local workspace cleared. You should see nextwork-search-design inside Trash.
- Switch back to Docker Desktop.
- Select Images.
- Locate the python:3.14.7-slim image.
- Use the deletion control for that image.
- Approve the removal if a confirmation dialog appears.
- Confirm the image no longer appears in the Images view.
Your cleanup is complete. The project has no remaining cloud resources or paid persistent resources.
Nice Work!
Nice Work!
You did it! Your local simulator now proves how published content can disappear from search. You also built the recovery path that makes publication durable before search catches up.
You've learned how to:
- Demonstrated a dual-write failure that left a published project missing from search. The project remained durable in content.db while search.db had no matching document.
- Implemented a transactional outbox that commits each content record with its indexing event. An asynchronous worker delivers those events with replay-safe idempotency.
- Used architecture.md to deliver a five-minute system design answer. You explained eventual consistency through a clear consistency boundary. You also defended the scale path. You connected failure recovery to observability.
- Completed the Secret Mission by diagnosing a three-event indexing backlog. You tracked the oldest pending event age. You replayed the crashed event safely. You proved recovery when the pending count returned to 0. All three backlog projects became searchable.
Ready to quiz yourself?