Build an ingestion job that survives an interrupted source write, closes and reopens its SQLite database, and resumes without duplicating committed chunks.
When you need this. Use it when a stopped ingestion job must continue from intact committed source units without skipping incomplete chunks.
Prerequisites. Complete the preceding ingestion module and use the supplied local files. These programs use local files and the Python standard library.
Pin the job before processing sources
A job is one attempt to ingest a fixed source snapshot. The job row records a job ID, the snapshot ID and a canonical hash of the complete supplied snapshot. The complete input hash catches changes to unprocessed text, metadata or chunker fields even if someone leaves the claimed snapshot ID unchanged.
The first input is a trusted local artifact from the versioning lab. Pinning it binds subsequent attempts to that same payload; it does not authenticate its author or validate an external upload. This lab allows one job per database. A different corpus revision belongs in a new job database or the separate update workflow.
Changed job input. Restarting with a different supplied payload raises IngestionError before source processing. Restore the pinned snapshot or deliberately start a new job. Do not edit the checkpoint to make changed input look like a retry.
Commit one complete source unit
The persistent source unit consists of a source revision, all its chunk rows and its completion checkpoint. A checkpoint records that this entire unit committed. The writer places the inserts in one SQLite transaction and writes the checkpoint before that transaction commits.
Interrupted write. The controlled run raises InterruptedError after inserting the next source’s rows but before committing its checkpoint. The transaction rolls those rows back. Previously committed source units remain in the database. This is an injected Python exception, not a simulated power failure or a measurement of crash durability.
A with db block commits a successful transaction or rolls it back when the exception leaves the block. Closing the connection is separate; the driver explicitly closes and reopens it before resuming. The database remains a local file, with one writer and no background workers.
Skip only an intact committed unit
On restart, the writer checks a completed source’s revision and complete sorted chunk ID/text set against the supplied snapshot. Only a matching unit is skipped. A checkpoint row alone is insufficient.
Stale checkpoint. A missing or changed stored chunk causes IngestionError instead of a silent skip. Repair the database from a reviewed snapshot. A reading marker or a partial row count does not show that ingestion is complete.
After finishing, another attempt skips the same intact units without adding rows. That is idempotency for this fixed job payload. It does not provide distributed exactly-once execution, remote deduplication or a concurrent-worker lease.
Execute the local corpus operation
Download the complete lab. The implementation, fixtures, tests, README and raw execution record travel together.
"""Persist a pinned ingestion job, one atomic source unit at a time."""
import hashlib
import json
import sqlite3
class IngestionError(ValueError):
pass
def connect(path):
db = sqlite3.connect(path)
db.execute("PRAGMA foreign_keys=ON")
db.executescript("""
CREATE TABLE IF NOT EXISTS jobs(job_id TEXT PRIMARY KEY, snapshot_id TEXT NOT NULL, input_sha256 TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS sources(source_id TEXT PRIMARY KEY, document_revision TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS chunks(chunk_id TEXT PRIMARY KEY, source_id TEXT NOT NULL REFERENCES sources(source_id), text TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS checkpoints(job_id TEXT NOT NULL REFERENCES jobs(job_id), source_id TEXT NOT NULL REFERENCES sources(source_id), document_revision TEXT NOT NULL, PRIMARY KEY(job_id, source_id));
""")
return db
def ingest(db, snapshot, job_id, interrupt_source=None):
"""An injected interruption raises before commit, leaving no partial source."""
input_digest = hashlib.sha256(json.dumps(snapshot, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")).hexdigest()
with db:
other = db.execute("SELECT job_id FROM jobs WHERE job_id<>?", (job_id,)).fetchone()
if other:
raise IngestionError("database belongs to another job; use a new database")
previous = db.execute("SELECT snapshot_id,input_sha256 FROM jobs WHERE job_id=?", (job_id,)).fetchone()
if previous and previous != (snapshot["snapshot_id"], input_digest):
raise IngestionError("job snapshot changed; start a new job")
db.execute("INSERT OR IGNORE INTO jobs VALUES (?,?,?)", (job_id, snapshot["snapshot_id"], input_digest))
actions = []
for source in snapshot["documents"]:
sid, revision = source["source_id"], source["document_revision"]
with db:
done = db.execute("SELECT document_revision FROM checkpoints WHERE job_id=? AND source_id=?", (job_id, sid)).fetchone()
stored = db.execute("SELECT document_revision FROM sources WHERE source_id=?", (sid,)).fetchone()
expected = sorted((c["chunk_id"], sid, c["text"]) for c in source["chunks"])
actual = list(db.execute("SELECT chunk_id,source_id,text FROM chunks WHERE source_id=? ORDER BY chunk_id", (sid,)))
if done:
if done[0] != revision or stored != (revision,) or actual != expected:
raise IngestionError("checkpoint does not match stored source")
actions.append({"source_id": sid, "action": "skipped"})
continue
if stored and stored != (revision,):
raise IngestionError("existing source revision differs; use update workflow")
if stored and actual != expected:
raise IngestionError("existing chunks differ from supplied source")
db.execute("INSERT OR IGNORE INTO sources VALUES (?,?)", (sid, revision))
if not stored:
for chunk in source["chunks"]:
db.execute("INSERT INTO chunks VALUES (?,?,?)", (chunk["chunk_id"], sid, chunk["text"]))
if sid == interrupt_source:
raise InterruptedError("injected interruption before source commit")
db.execute("INSERT INTO checkpoints VALUES (?,?,?)", (job_id, sid, revision))
actions.append({"source_id": sid, "action": "committed"})
return actions
def state(db):
return {table: [list(row) for row in db.execute(f"SELECT * FROM {table} ORDER BY 1,2")] for table in ("jobs", "sources", "chunks", "checkpoints")}python3 run.py --database /tmp/resume-ingestion-demo.sqlite
python3 -m unittest -v test_ingest.py[
{
"operation": "interrupt-before-second-source-commit",
"status": "interrupted",
"error": "injected interruption before source commit"
},
{
"operation": "close-reopen-and-resume",
"status": "completed"
},
{
"operation": "retry-completed-job",
"status": "completed"
}
]Inspect the committed rows and operation outcomes in the complete saved result.
Inspect the complete saved result
{
"execution_mode": "offline-sqlite-source-maintenance",
"model_called": false,
"sqlite_version": "3.53.4",
"snapshot_id": "0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"steps": [
{
"operation": "interrupt-before-second-source-commit",
"status": "interrupted",
"error": "injected interruption before source commit",
"state": {
"jobs": [
[
"python-docs-v1",
"0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"53a35a4f444bf0ea58c4561b94d9b468a611c0439be1e0006467d306afb02bed"
]
],
"sources": [
[
"json-text",
"0c4b565b2b1900787d6da9f8539c471f8cb2245c00abf887a57d460364abc57a"
]
],
"chunks": [
[
"e5455737b11664449f653280d89dfc2faaa9a14de16e9c3eb24296ca2ceea51c",
"json-text",
"json.dumps(obj, ensure_ascii=True) returns a JSON string. Setting ensure_ascii=False leaves non-ASCII characters in that string. It does not write a file or select the encoding used by a file writer."
]
],
"checkpoints": [
[
"python-docs-v1",
"json-text",
"0c4b565b2b1900787d6da9f8539c471f8cb2245c00abf887a57d460364abc57a"
]
]
}
},
{
"operation": "close-reopen-and-resume",
"status": "completed",
"actions": [
{
"source_id": "json-text",
"action": "skipped"
},
{
"source_id": "path-read",
"action": "committed"
},
{
"source_id": "path-read-3-13",
"action": "committed"
},
{
"source_id": "path-write",
"action": "committed"
}
],
"state": {
"jobs": [
[
"python-docs-v1",
"0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"53a35a4f444bf0ea58c4561b94d9b468a611c0439be1e0006467d306afb02bed"
]
],
"sources": [
[
"json-text",
"0c4b565b2b1900787d6da9f8539c471f8cb2245c00abf887a57d460364abc57a"
],
[
"path-read",
"b58385b9bf597a951c16c035653d639929924e53eae3e5ada255afd6ee1592f2"
],
[
"path-read-3-13",
"d68df3e7ea1e24af3f40366a61402fc80558fefe2ef5f3eda928de0e01820db9"
],
[
"path-write",
"28c6349eaf80b629322edb310bb871dc45a04af9fad3b752bc75d12f0d7314e7"
]
],
"chunks": [
[
"bdaeff5ae3ab88b70b12b9b6d4a0a46de45ea14520b971a7756cb03bbd384f35",
"path-read-3-13",
"In Python 3.13, Path.read_text accepts encoding, errors and newline. The newline parameter was added in Python 3.13. The method returns decoded file text and closes the file."
],
[
"c66643a4609c6ee157458bf4353f7446b9c6a200a3ef453846d6a18d5b5069a7",
"path-write",
"Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written."
],
[
"c737db948e8111ae00e3442d8a22c80c876e43446dc1ee176ffe8ce142a41900",
"path-read",
"Path.read_text(encoding=None, errors=None) opens a file, returns its text as a string and closes the file. Set encoding=\"utf-8\" when reading a UTF-8 file."
],
[
"e5455737b11664449f653280d89dfc2faaa9a14de16e9c3eb24296ca2ceea51c",
"json-text",
"json.dumps(obj, ensure_ascii=True) returns a JSON string. Setting ensure_ascii=False leaves non-ASCII characters in that string. It does not write a file or select the encoding used by a file writer."
]
],
"checkpoints": [
[
"python-docs-v1",
"json-text",
"0c4b565b2b1900787d6da9f8539c471f8cb2245c00abf887a57d460364abc57a"
],
[
"python-docs-v1",
"path-read",
"b58385b9bf597a951c16c035653d639929924e53eae3e5ada255afd6ee1592f2"
],
[
"python-docs-v1",
"path-read-3-13",
"d68df3e7ea1e24af3f40366a61402fc80558fefe2ef5f3eda928de0e01820db9"
],
[
"python-docs-v1",
"path-write",
"28c6349eaf80b629322edb310bb871dc45a04af9fad3b752bc75d12f0d7314e7"
]
]
}
},
{
"operation": "retry-completed-job",
"status": "completed",
"actions": [
{
"source_id": "json-text",
"action": "skipped"
},
{
"source_id": "path-read",
"action": "skipped"
},
{
"source_id": "path-read-3-13",
"action": "skipped"
},
{
"source_id": "path-write",
"action": "skipped"
}
],
"state": {
"jobs": [
[
"python-docs-v1",
"0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"53a35a4f444bf0ea58c4561b94d9b468a611c0439be1e0006467d306afb02bed"
]
],
"sources": [
[
"json-text",
"0c4b565b2b1900787d6da9f8539c471f8cb2245c00abf887a57d460364abc57a"
],
[
"path-read",
"b58385b9bf597a951c16c035653d639929924e53eae3e5ada255afd6ee1592f2"
],
[
"path-read-3-13",
"d68df3e7ea1e24af3f40366a61402fc80558fefe2ef5f3eda928de0e01820db9"
],
[
"path-write",
"28c6349eaf80b629322edb310bb871dc45a04af9fad3b752bc75d12f0d7314e7"
]
],
"chunks": [
[
"bdaeff5ae3ab88b70b12b9b6d4a0a46de45ea14520b971a7756cb03bbd384f35",
"path-read-3-13",
"In Python 3.13, Path.read_text accepts encoding, errors and newline. The newline parameter was added in Python 3.13. The method returns decoded file text and closes the file."
],
[
"c66643a4609c6ee157458bf4353f7446b9c6a200a3ef453846d6a18d5b5069a7",
"path-write",
"Path.write_text(data, encoding=None, errors=None, newline=None) writes text to a file and closes it. Set encoding=\"utf-8\" to select UTF-8. An existing file is overwritten. The method returns the number of characters written."
],
[
"c737db948e8111ae00e3442d8a22c80c876e43446dc1ee176ffe8ce142a41900",
"path-read",
"Path.read_text(encoding=None, errors=None) opens a file, returns its text as a string and closes the file. Set encoding=\"utf-8\" when reading a UTF-8 file."
],
[
"e5455737b11664449f653280d89dfc2faaa9a14de16e9c3eb24296ca2ceea51c",
"json-text",
"json.dumps(obj, ensure_ascii=True) returns a JSON string. Setting ensure_ascii=False leaves non-ASCII characters in that string. It does not write a file or select the encoding used by a file writer."
]
],
"checkpoints": [
[
"python-docs-v1",
"json-text",
"0c4b565b2b1900787d6da9f8539c471f8cb2245c00abf887a57d460364abc57a"
],
[
"python-docs-v1",
"path-read",
"b58385b9bf597a951c16c035653d639929924e53eae3e5ada255afd6ee1592f2"
],
[
"python-docs-v1",
"path-read-3-13",
"d68df3e7ea1e24af3f40366a61402fc80558fefe2ef5f3eda928de0e01820db9"
],
[
"python-docs-v1",
"path-write",
"28c6349eaf80b629322edb310bb871dc45a04af9fad3b752bc75d12f0d7314e7"
]
]
}
}
]
}Compare the state before and after the operation
Run the demonstration with a new database path. Compare the state after interruption with the state after close/reopen/resume. The interrupted source must have no committed rows or checkpoint, while earlier source units remain. Compare the final retry: every intact unit should be skipped and the stored rows should stay unchanged. The driver refuses an existing database path so each demonstration starts from a known state.
Read the explanation
Check the specific stored state and input contract; a completed command alone does not prove corpus consistency.
What’s next
Continue with Update and Delete Indexed Content.
References
APIs were checked on 23 October 2026 against official Python hashlib, Python sqlite3, SQLite transactions and SQLite FTS5 documentation as applicable. The source summaries retain their version-specific Python documentation URLs. The changed fixture is a labelled local teaching revision, not an upstream documentation update.