A Changed-Page Export Can Quietly Shrink Your RAG Index

Your weekly refresh finishes successfully. Then the chatbot loses answers from pages nobody edited.

The suspicious line is downstream: replace_index(dataset). Once you enable crawlOnlyPagesChangedSince, that export contains changed pages. Unchanged pages are absent by design.

Lab note 047: can we make corpus refreshes cheaper without interpreting silence as permission to delete? The catch comes first: a changed-page delta cannot establish which documents were deleted.

What can an incremental export prove?

Our site crawler uses HTTP validators, sitemap lastmod, and content hashes to identify changes. That helps decide what to return. It doesn't turn an omitted URL into a tombstone.

There's another boundary: maxCrawlPages defaults to 20 and limits stored pages and files. The crawl stops at that limit, so a bounded export can't establish complete change coverage either.

The storage contract changes when you switch from full crawls to deltas:

changed-page export | v merge into document state --> replace affected chunk groups | +------------------> leave untouched documents alone independent inventory + checks --> deletion review queue

That distinction is the whole experiment. A successful run establishes successful execution within its scope. Your index still needs its own rules for continuity.

We'd rather keep a document pending review than let an empty export become a very efficient eraser.

A page owns its chunks

For retrieval, the embedding unit is a heading-aware chunk. The replacement unit is the source document.

That matters when an edited page gets shorter. Appending its new chunks leaves the old ending searchable, possibly answering questions with instructions the publisher removed.

Here's the corpus shape we want:

Corpus component	What it preserves
Document identity	An explicit policy based on the requested `url`
Source metadata	Available `loadedUrl`, `canonicalUrl`, and `referrerUrl`
Chunk record	Text, `headingPath`, document identity, representation version
Manifest	Source `contentHash`, fetch time, desired chunk IDs
Delivery queue	Previous chunk IDs to delete and replacement records
Review queue	Candidates requiring independent deletion or ingestion checks

We use the exact requested URL as identity and accept only URLs without credentials, queries, or fragments. That's a deliberately narrow policy for a public documentation corpus. Sites using query parameters as document identifiers need a different policy.

Canonical URLs remain provenance here. Automatically adopting each new canonical as the primary key can strand the old document when that declaration changes.

Likewise, referrerUrl preserves discovery context. It doesn't prove that a page is the semantic parent of another page.

The following code defines our local schema and prepares records. The database is created if it doesn't exist; there's no input file to supply.

import hashlib import json import os import sqlite3 import unicodedata from datetime import datetime, timezone from urllib.parse import urlsplit from apify_client import ApifyClient def pack(value): return json.dumps(value, ensure_ascii=False, sort_keys=True) def digest(value): return hashlib.sha256(pack(value).encode("utf-8")).hexdigest() def open_state(path): connection = sqlite3.connect(path) connection.executescript(""" CREATE TABLE IF NOT EXISTS documents ( doc_id TEXT PRIMARY KEY, body TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS outbox ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS review ( doc_id TEXT PRIMARY KEY, payload TEXT NOT NULL ); """) return connection def safe_url(value): parsed = urlsplit(value) allowed = ( parsed.scheme in {"http", "https"} and parsed.hostname and not parsed.username and not parsed.password and not parsed.query and not parsed.fragment ) return value if allowed else None def prepare(row): requested = row.get("url", "") if not safe_url(requested): raise ValueError("source identity needs review") fetched = datetime.fromisoformat( row.get("loadedTime", "").replace("Z", "+00:00") ) if fetched.tzinfo is None: raise ValueError("fetch time needs a timezone") raw_parts = row.get("chunks") or [ {"text": row.get("markdown") or row.get("text", "")} ] parts = [ { "text": unicodedata.normalize("NFC", part.get("text", "")).strip(), "headingPath": part.get("headingPath", ""), } for part in raw_parts ] if not parts or any(not part["text"] for part in parts): raise ValueError("empty content needs review") sources = {} for field in ("url", "loadedUrl", "canonicalUrl", "referrerUrl"): accepted = safe_url(row.get(field, "")) if accepted: sources[field] = accepted doc_id = digest(requested) source_hash = row.get("contentHash") version = digest({ "policy": "nfc-v1", "parts": parts, "sources": sources, "contentHash": source_hash, }) records = [ { "id": digest([doc_id, version, position]), "document_id": doc_id, "version": version, "text": part["text"], "headingPath": part["headingPath"], "sources": sources, "duplicate_group": source_hash or digest(parts), } for position, part in enumerate(parts) ] return { "url": requested, "contentHash": source_hash, "loadedTime": fetched.astimezone(timezone.utc).isoformat(), "version": version, "chunk_ids": [record["id"] for record in records], "records": records, }

The representation version includes chunk boundaries and heading paths because the source hash covers normalized text. Text identity alone doesn't describe everything an embedding pipeline stores.

The duplicate_group is a useful seed for identifying repeated content across URLs. We keep those sources separate rather than silently losing a citation.

Replace a document as a group

The manifest and delivery queue commit together.

That prevents a nasty interruption: recording a new version as processed, crashing before its index operations are saved, then skipping it on retry.

This function stages replacement batches and leaves every unmentioned document intact.

def merge(connection, rows): pending, batches, reviews = {}, [], [] with connection: for row in rows: doc_id = digest(row.get("url", "")) try: incoming = prepare(row) if doc_id in pending: previous = pending[doc_id] else: stored = connection.execute( "SELECT body FROM documents WHERE doc_id = ?", (doc_id,), ).fetchone() previous = json.loads(stored[0]) if stored else None if previous: if incoming["loadedTime"] < previous["loadedTime"]: continue if ( incoming["loadedTime"] == previous["loadedTime"] and incoming["version"] != previous["version"] ): raise ValueError("conflicting versions at one fetch time") pending[doc_id] = incoming if previous and incoming["version"] == previous["version"]: continue batches.append((pack({ "document_id": doc_id, "delete_ids": previous["chunk_ids"] if previous else [], "upserts": incoming["records"], }),)) except (ValueError, TypeError) as error: reviews.append((doc_id, pack({ "document_id": doc_id, "reason": str(error), }))) connection.executemany( """INSERT INTO documents VALUES (?, ?) ON CONFLICT(doc_id) DO UPDATE SET body = excluded.body""", [(key, pack(value)) for key, value in pending.items()], ) connection.executemany( "INSERT INTO outbox(payload) VALUES (?)", batches ) connection.executemany( "INSERT OR REPLACE INTO review VALUES (?, ?)", reviews ) def nominate(connection, urls, evidence): with connection: connection.executemany( "INSERT OR REPLACE INTO review VALUES (?, ?)", [ (digest(url), pack({"url": url, "evidence": evidence})) for url in urls ], )

This manifest records desired state. A queued replacement isn't evidence that your vector store has applied it.

Your delivery worker must process outbox.sequence in order, acknowledge only after successful application, and finish a pending batch before advancing. Stable chunk IDs make retrying a batch safe when the destination supports idempotent deletes and upserts.

For uninterrupted retrieval, stage the replacement and switch the active document version atomically where your index supports it. Otherwise, delete-then-upsert can briefly remove context; upsert-then-delete can briefly expose both versions.

Use a single writer for this compact planner. Multiple ingestion workers need explicit concurrency control around manifest reads and commits.

Give the merge a synthetic trap

The useful fixture is a shrinking document beside an untouched document.

Example: these synthetic records exercise replacement, preservation, repeated ingestion, and a review nomination without claiming a crawl result.

def fixture(path, texts, fetched): return { "url": "https://example.com" + path, "loadedTime": fetched, "contentHash": digest(texts), "chunks": [ {"text": text, "headingPath": "Manual"} for text in texts ], } baseline = [ fixture("/docs/install", ["Install the package.", "Legacy setup."], "2026-10-01T00:00:00Z"), fixture("/docs/reference", ["Reference material."], "2026-10-01T00:00:00Z"), ] delta = [ fixture("/docs/install", ["Install the current package."], "2026-10-16T12:00:05.123Z") ] test_db = open_state(":memory:") merge(test_db, baseline) original = dict(test_db.execute("SELECT doc_id, body FROM documents")) merge(test_db, delta) replacement = json.loads(test_db.execute( "SELECT payload FROM outbox ORDER BY sequence DESC LIMIT 1" ).fetchone()[0]) install_id = digest(baseline[0]["url"]) reference_id = digest(baseline[1]["url"]) assert replacement["delete_ids"] == json.loads(original[install_id])["chunk_ids"] assert len(replacement["upserts"]) < len(replacement["delete_ids"]) assert dict(test_db.execute( "SELECT doc_id, body FROM documents" ))[reference_id] == original[reference_id] queued_before_retry = test_db.execute( "SELECT COUNT(*) FROM outbox" ).fetchone()[0] merge(test_db, delta) assert test_db.execute( "SELECT COUNT(*) FROM outbox" ).fetchone()[0] == queued_before_retry nominate(test_db, [baseline[1]["url"]], "synthetic inventory omission") assert test_db.execute( "SELECT body FROM documents WHERE doc_id = ?", (reference_id,) ).fetchone() is not None test_db.close()

The assertions define the behavior your implementation should preserve. No printed row counts, no victory lap around a fixture.

Where deletion evidence comes from

An independent inventory can nominate missing URLs. It shouldn't immediately remove them.

Before approving a tombstone, check whether discovery covered the intended scope, whether limits or failures interrupted it, and whether redirects or identity changes explain the absence. A temporary fetch failure deserves a retry policy, not automatic historical revision.

That's how we'd use Domain-Scoped Site Crawler for Markdown Datasets: incremental ingestion for freshness, scheduled discovery and reconciliation for coverage.

Several tempting shortcuts still don't work:

Advancing the cutoff to the largest returned loadedTime doesn't establish complete coverage. Keep overlap and advance your refresh checkpoint only under an explicit coverage policy.
Randomly splitting chunks between training and evaluation leaks neighboring context. Group document versions, aliases, and duplicate-content families before assigning splits.
Treating a character target as a hard embedding budget fails on intact code blocks and tables. Measure with your embedding tokenizer and route oversized units deliberately.
Omitting author doesn't anonymize page text. Names, contact details, headings, and URL paths still need your privacy policy.

Our exported schema excludes raw author metadata, images, and arbitrary metadata objects. Its URL checks also exclude query strings. Neither choice replaces content review, including review of what persists in the database.

Changing the cleaner or chunking policy requires reprocessing retained documents. A delta can't deliver an unchanged page merely because our downstream rules changed.

Reproduce the refresh

The verdict is plain: merge changed documents, replace their chunk groups, and make deletion earn separate evidence.

You can run the cells above, then use this cell to ingest an export and write the current corpus plus pending work. The corpus contains desired document state; the delivery queue tells you what still needs applying.

Example: this input selects a documentation scope and enables structured chunks.

run_input = { "startUrls": [{"url": "https://example.com/docs"}], "includeUrlGlobs": ["https://example.com/docs*"], "useSitemaps": True, "maxCrawlPages": 500, "maxCrawlDepth": 5, "crawlOnlyPagesChangedSince": "2026-10-01", "outputFormat": "markdown", "chunkMarkdown": True, "chunkMaxChars": 1500, "chunkOverlapChars": 200, } client = ApifyClient(os.environ["APIFY_TOKEN"]) run = client.actor( "vexaloom-labs/domain-scoped-site-crawler-for-markdown-datasets" ).call(run_input=run_input) if run["status"] != "SUCCEEDED": raise RuntimeError(f"Refresh did not succeed: {run['status']}") db = open_state("corpus.sqlite3") merge(db, client.dataset(run["defaultDatasetId"]).iterate_items()) with open("corpus.jsonl", "w", encoding="utf-8") as corpus_file: for (body,) in db.execute("SELECT body FROM documents ORDER BY doc_id"): for record in json.loads(body)["records"]: corpus_file.write(pack(record) + "\n") for filename, table in [ ("refresh.jsonl", "outbox"), ("reconcile.jsonl", "review"), ]: with open(filename, "w", encoding="utf-8") as queue_file: for (payload,) in db.execute( f"SELECT payload FROM {table} ORDER BY rowid" ): queue_file.write(payload + "\n") db.close()