A slow worker may finish indexing an old version after a withdrawal or a new version. An at-least-once queue may replay the same event. A vector index that does not check the source state can then reintroduce an outdated document.
A local mini document-management system in PostgreSQL/pgvector maintains versions and events, computes real embeddings, and exposes authorised searches. A step-by-step scenario shows obsolete jobs being rejected and indexing lag without bringing withdrawn documents back to life.
Download the complete project, revision r003
Data, corpus, and model
Vesta-Sync contains 200 original English texts, twenty case files, and four fictional profiles. A text has one to three paragraphs retained as fragments, giving 399 initial fragments. Case-file references and business terms are strong synthetic signals. One hundred queries fixed before execution have relevance defined by topic, case file, and profile permission; cases with no relevant document remain separate in the metrics. No labels are mounted in the inference service.
E5-small is pinned to revision 614241f622f53c4eeff9890bdc4f31cfecc418b3. Queries use query: and fragments use passage: . The code loads the acquired tensors, checks keys and bytes, averages the actual states using attention_mask, then applies L2 normalisation in 384 dimensions. Real special tokens take part in pooling. The unused CLS pooler is explicitly excluded. A fragment longer than 512 tokens is rejected, never silently truncated. Cosine measures proximity, not permission.
One transaction before, another after computation
The documents_current table carries the version, hash, withdrawal state, and ACL. Each mutation creates a new revision, keeps the history, and adds the outbox event in the same commit. A rollback must cancel both. The SQL sequence can have gaps after rollback: this is not the loss of a committed event.
A worker briefly reserves an event with a lock and SKIP LOCKED. It closes the connection before E5 inference. The final transaction takes locks in epoch then document order, and compares version, hash, withdrawal status, and active epoch. It inserts fragments and a receipt atomically only if all checks pass; otherwise the receipt is STALE. A redelivery that was already received returns DUPLICATE with no new effect. Computation may be repeated: exactly-once execution is not promised.
The abandoned demonstration worker is recovered by an explicit redelivery action. There is no automatically expiring lease or distributed coordinator here. The exercise shows the guards separately, without confusing rejection at write time with authorisation at read time.
The authorised set before distances
Search uses a read-only repeatable-read transaction. Its first SELECT fixes the snapshot; no source access occurs before it in that transaction. It follows the commits recorded in the scenarios. A MATERIALIZED CTE joins fragments to the current version/hash, active epoch, and ACLs, excluding withdrawals. Exact distance calculation then uses that set, and the best fragment distance produces the document ranking with deterministic ties.
The practical exercise retains EXPLAIN ANALYZE VERBOSE JSON and eligible vectors to verify the plan and recalculate distances. It does not assume that the textual order of SQL clauses imposes execution order. No HNSW or response cache is used. A result received before a revocation remains in the client memory; the guarantee does not erase it retroactively.
A withdrawal is a tombstone, not physical erasure. An ACL change also increments the revision in this minimal model. Until reindexing, coverage may therefore decrease even for a profile that remains authorised. Search does not reuse an outdated version to hide that lag.
Demonstration and scenarios
An isolated naive_index table really receives an old vector after withdrawal; its query, which lacks a source join, exposes the document. The same data submitted to the conditional check is rejected. This table is never used by the correct workflow.
Twelve independent schemas each start from the same 200 sources: two paused jobs then withdrawal; v2 completed before v1; duplicate; abandonment after computation; effect/receipt rollback; revocation between embedding and SELECT; an earlier response retained then revocation; old epoch; rebuild and ACL; concurrent reservation of the same event; source/outbox rollback; and all documents forbidden to the profile. Explicit threads and barriers reproduce the orders, not sleeps. Barrier timeouts detect a blocked scenario and do not define its order.
What the outputs measure
Recall@5 is averaged only over queries with an authorised relevant item. The number with no relevant item and their empty-response rate are separate. Without a rejection threshold, an out-of-corpus query may return authorised but irrelevant documents. Initial coverage reports the documents and fragments actually indexed.
Safety is measured across the twelve scenarios, independently of the one hundred relevance queries. The criterion is zero forbidden results in the post-commit snapshot within this scope. APPLIED/STALE receipts, DUPLICATE redeliveries, query times, and delays from outbox-row creation to receipt are retained. Row creation precedes its commit, so this delay is not an exact commit-to-index measure and also includes that part of the transaction. Delays include queue waiting, computation, and imposed pauses; they do not predict a production SLA. Scenario 1 also measures the two client returns from commit (v2, then index) with a monotonic clock; this observed duration includes encoding and coordination without claiming to observe the internal WAL instant. Scenarios 3 and 4 time redelivery, its recomputation, and the final commit. Recovery costs must be read within the scenario timeline, without confusing no duplicate effect with no recomputation.
The exercise retains v1, changes the text, revokes one profile, then rebuilds the epoch with the same pinned E5. Version, hash, and epoch reject the old job; withdrawal is not active, so that guard passes. Search then distinguishes the revoked profile from the profile that remains authorised. The rebuild tests a state protocol; it does not measure a migration to another model.
Step 01 — Build the catalogue, versions, and outbox
SQL atomicity and source schema
Error: The event is written after the document commit
Correction: Put both writes in the same transaction and test a rollback.
File src/ged_index_transactionnel/schema.py
"""Source, immutable events and derived effects; every SQL value is parameterized."""
import hashlib, json, os, re
from contextlib import contextmanager
import psycopg
from psycopg import sql
from psycopg.rows import dict_row
def content_hash(text):
return hashlib.sha256(text.encode()).hexdigest()
def schema_name(name):
if not re.fullmatch(r"tp38_[a-z0-9_]{1,48}", name):
raise ValueError("Invalid scenario schema")
return name
@contextmanager
def connection(name):
schema_name(name)
with psycopg.connect(
os.environ.get(
"DATABASE_URL", "postgresql://reader:local-demo-only@db:5432/tutorial"
),
autocommit=True,
row_factory=dict_row,
) as conn:
conn.execute(
sql.SQL("SET search_path TO {}, public").format(sql.Identifier(name))
)
yield conn
DDL = """
CREATE TABLE epoch(singleton boolean PRIMARY KEY DEFAULT true CHECK(singleton), active integer NOT NULL CHECK(active>0));
INSERT INTO epoch VALUES(true,1);
CREATE TABLE documents_current(document_id text PRIMARY KEY, dossier text NOT NULL, revision integer NOT NULL CHECK(revision>0), content_hash text NOT NULL, title text NOT NULL, text_content text NOT NULL, withdrawn boolean NOT NULL DEFAULT false, acl integer[] NOT NULL, changed_at timestamptz NOT NULL DEFAULT clock_timestamp());
CREATE TABLE document_history(document_id text NOT NULL, revision integer NOT NULL, content_hash text NOT NULL, title text NOT NULL, text_content text NOT NULL, withdrawn boolean NOT NULL, acl integer[] NOT NULL, PRIMARY KEY(document_id,revision));
CREATE TABLE outbox(seq bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, document_id text NOT NULL REFERENCES documents_current, revision integer NOT NULL, content_hash text NOT NULL, model_epoch integer NOT NULL, text_content text NOT NULL, title text NOT NULL, created_at timestamptz NOT NULL DEFAULT clock_timestamp(), claimed_by text, attempts integer NOT NULL DEFAULT 0);
CREATE TABLE receipts(seq bigint PRIMARY KEY REFERENCES outbox, status text NOT NULL CHECK(status IN ('APPLIED','STALE')), checks jsonb NOT NULL, completed_at timestamptz NOT NULL DEFAULT clock_timestamp(), lag_ms double precision NOT NULL);
CREATE TABLE fragments(document_id text NOT NULL, revision integer NOT NULL, model_epoch integer NOT NULL, fragment_no integer NOT NULL, content_hash text NOT NULL, title text NOT NULL, text_content text NOT NULL, embedding vector(384) NOT NULL, event_seq bigint NOT NULL REFERENCES outbox, PRIMARY KEY(document_id,revision,model_epoch,fragment_no));
"""
def migrate_schema(name):
# Refuse reuse; no DROP, no reset of an existing scenario or historical evidence.
schema_name(name)
with connection(name) as conn:
with conn.transaction():
conn.execute("CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA public")
conn.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(name)))
conn.execute(DDL)
return {
"server_version": conn.execute("SHOW server_version").fetchone()[
"server_version"
],
"pgvector": conn.execute(
"SELECT extversion FROM pg_extension WHERE extname='vector'"
).fetchone()["extversion"],
"schema": name,
}
def append_event(conn, document, epoch):
return conn.execute(
"INSERT INTO outbox(document_id,revision,content_hash,model_epoch,text_content,title) VALUES(%s,%s,%s,%s,%s,%s) RETURNING seq,created_at",
(
document["document_id"],
document["revision"],
document["content_hash"],
epoch,
document["text_content"],
document["title"],
),
).fetchone()
def history(conn, d):
conn.execute(
"INSERT INTO document_history VALUES(%s,%s,%s,%s,%s,%s,%s)",
tuple(
d[k]
for k in [
"document_id",
"revision",
"content_hash",
"title",
"text_content",
"withdrawn",
"acl",
]
),
)
def seed_demo_catalog(name, documents):
with connection(name) as conn, conn.transaction():
events = []
for doc in documents:
d = dict(
doc,
revision=1,
withdrawn=False,
content_hash=content_hash(doc["text_content"]),
)
conn.execute(
"INSERT INTO documents_current(document_id,dossier,revision,content_hash,title,text_content,withdrawn,acl) VALUES(%s,%s,%s,%s,%s,%s,%s,%s)",
tuple(
d[k]
for k in [
"document_id",
"dossier",
"revision",
"content_hash",
"title",
"text_content",
"withdrawn",
"acl",
]
),
)
history(conn, d)
events.append(append_event(conn, d, 1))
return events
def commit_document_change(
name, identifier, *, text=None, withdrawn=None, acl=None, force_rollback=False
):
with connection(name) as conn, conn.transaction():
# Same lock order as worker finalization/rebuild: epoch, then document.
epoch = conn.execute("SELECT active FROM epoch FOR SHARE").fetchone()["active"]
old = conn.execute(
"SELECT * FROM documents_current WHERE document_id=%s FOR UPDATE",
(identifier,),
).fetchone()
if old is None:
raise ValueError("Unknown document")
new = dict(old, revision=old["revision"] + 1)
if text is not None:
new["text_content"] = text
new["content_hash"] = content_hash(text)
if withdrawn is not None:
new["withdrawn"] = bool(withdrawn)
if acl is not None:
if any(type(p) is not int or p not in range(4) for p in acl):
raise ValueError("Unknown profile")
new["acl"] = sorted(set(acl))
conn.execute(
"UPDATE documents_current SET revision=%s,content_hash=%s,text_content=%s,withdrawn=%s,acl=%s,changed_at=clock_timestamp() WHERE document_id=%s",
tuple(
new[k]
for k in [
"revision",
"content_hash",
"text_content",
"withdrawn",
"acl",
"document_id",
]
),
)
history(conn, new)
event = append_event(conn, new, epoch)
if force_rollback:
raise RuntimeError("Injected source/outbox rollback")
return event
def rebuild_model_epoch(name):
with connection(name) as conn, conn.transaction():
value = conn.execute(
"UPDATE epoch SET active=active+1 RETURNING active"
).fetchone()["active"]
docs = conn.execute(
"SELECT * FROM documents_current ORDER BY document_id FOR UPDATE"
).fetchall()
events = [append_event(conn, d, value) for d in docs if not d["withdrawn"]]
return {
"model_epoch": value,
"events": events,
"acl": {d["document_id"]: d["acl"] for d in docs},
}
Step 02 — Compute real embeddings and exact search
E5 prefixes and authorised subset
Error: The vector carries copied ACLs that have become incorrect
Correction: Join the source state on every SELECT; do not delegate the decision to the vector.
File src/ged_index_transactionnel/retrieval.py
import time
from .schema import connection
from .worker import vector_literal
QUERY = """
WITH authorized AS MATERIALIZED (
SELECT f.document_id,f.fragment_no,f.title,f.text_content,f.embedding
FROM fragments f JOIN documents_current d USING(document_id)
CROSS JOIN epoch e
WHERE NOT d.withdrawn AND %s=ANY(d.acl)
AND f.revision=d.revision AND f.content_hash=d.content_hash AND f.model_epoch=e.active
), distances AS MATERIALIZED (
SELECT document_id,fragment_no,title,text_content,embedding <=> %s::vector AS distance FROM authorized
), best AS (
SELECT DISTINCT ON(document_id) * FROM distances ORDER BY document_id,distance,fragment_no
)
SELECT * FROM best ORDER BY distance,document_id LIMIT %s
"""
def search_authorized_exact(name, vector, profile, top_k=5, explain=False):
if type(profile) is not int or profile not in range(4):
raise ValueError("Unknown profile")
if type(top_k) is not int or top_k < 1:
raise ValueError("Invalid top_k")
parameters = (profile, vector_literal(vector), top_k)
started = time.perf_counter()
with connection(name) as conn:
# Read-only repeatable-read makes results, eligible evidence and EXPLAIN
# use one snapshot, acquired by the first SELECT below.
with conn.transaction():
conn.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY")
snapshot = conn.execute(
"SELECT pg_current_snapshot()::text AS snapshot,clock_timestamp() AS observed_at"
).fetchone()
rows = conn.execute(QUERY, parameters).fetchall()
eligible = conn.execute(
"SELECT f.document_id,f.fragment_no,f.revision,f.model_epoch,f.embedding::text AS embedding FROM fragments f JOIN documents_current d USING(document_id) CROSS JOIN epoch e WHERE NOT d.withdrawn AND %s=ANY(d.acl) AND f.revision=d.revision AND f.content_hash=d.content_hash AND f.model_epoch=e.active ORDER BY f.document_id,f.fragment_no",
(profile,),
).fetchall()
plan = (
conn.execute(
"EXPLAIN (ANALYZE,VERBOSE,FORMAT JSON) " + QUERY, parameters
).fetchone()
if explain
else None
)
return {
"rows": rows,
"eligible": eligible,
"snapshot": snapshot,
"seconds": time.perf_counter() - started,
"plan": plan,
"query_vector": list(map(float, vector)),
"profile": profile,
"response_cache": False,
}
Step 03 — Walk through a stale worker write
Concurrency and revision comparison
Error: The worker keeps a transaction open during inference
Correction: Release it before computation, then lock and check in a short final transaction.
File src/ged_index_transactionnel/worker.py
import json, math
from .schema import connection
def vector_literal(vector):
if len(vector) != 384 or any(not math.isfinite(float(x)) for x in vector):
raise ValueError("Expected finite 384-vector")
norm = math.sqrt(sum(float(x) ** 2 for x in vector))
if abs(norm - 1) > 1e-5:
raise ValueError("Expected L2-normalized vector")
return "[" + ",".join(format(float(x), ".9g") for x in vector) + "]"
def split_fragments(text):
values = text.split("\n\n")
if not 1 <= len(values) <= 3 or any(not t.strip() for t in values):
raise ValueError("Expected 1 to 3 nonempty original paragraphs")
return values
def claim_event(name, worker, seq=None):
with connection(name) as conn, conn.transaction():
row = conn.execute(
"SELECT o.* FROM outbox o LEFT JOIN receipts r USING(seq) WHERE r.seq IS NULL AND o.claimed_by IS NULL AND (%s::bigint IS NULL OR o.seq=%s) ORDER BY o.seq FOR UPDATE OF o SKIP LOCKED LIMIT 1",
(seq, seq),
).fetchone()
if row is None:
return None
conn.execute(
"UPDATE outbox SET claimed_by=%s, attempts=attempts+1 WHERE seq=%s",
(worker, row["seq"]),
)
return dict(row, claimed_by=worker, attempts=row["attempts"] + 1)
def release_for_redelivery(name, seq):
# Explicit recovery action in this fixture; no implicit claim expiration promise.
with connection(name) as conn, conn.transaction():
return conn.execute(
"UPDATE outbox SET claimed_by=NULL WHERE seq=%s AND NOT EXISTS(SELECT 1 FROM receipts WHERE seq=%s) RETURNING seq",
(seq, seq),
).fetchone()
def compute_embeddings(event, encoder):
# No SQL connection or transaction survives claim_event's context manager.
fragments = split_fragments(event["text_content"])
vectors = encoder.encode_passages(fragments)
return [
{
"fragment_no": i,
"text_content": text,
"embedding": v.tolist() if hasattr(v, "tolist") else list(v),
}
for i, (text, v) in enumerate(zip(fragments, vectors, strict=True))
]
def conditional_commit(name, event, fragments, force_rollback=False):
# Validate dimensions before SQL; no partial effect if one vector is invalid.
literals = [vector_literal(f["embedding"]) for f in fragments]
if [f["text_content"] for f in fragments] != split_fragments(event["text_content"]):
raise ValueError("Fragment/event mismatch")
with connection(name) as conn, conn.transaction():
epoch = conn.execute("SELECT active FROM epoch FOR SHARE").fetchone()["active"]
current = conn.execute(
"SELECT * FROM documents_current WHERE document_id=%s FOR UPDATE",
(event["document_id"],),
).fetchone()
existing = conn.execute(
"SELECT * FROM receipts WHERE seq=%s", (event["seq"],)
).fetchone()
if existing is not None:
return {
"status": "DUPLICATE",
"original_status": existing["status"],
"seq": event["seq"],
"logical_writes": 0,
}
checks = {
"revision": current["revision"] == event["revision"],
"content_hash": current["content_hash"] == event["content_hash"],
"not_withdrawn": not current["withdrawn"],
"model_epoch": epoch == event["model_epoch"],
}
applied = all(checks.values())
if applied:
for f, value in zip(fragments, literals):
conn.execute(
"INSERT INTO fragments VALUES(%s,%s,%s,%s,%s,%s,%s,%s::vector,%s) ON CONFLICT(document_id,revision,model_epoch,fragment_no) DO NOTHING",
(
event["document_id"],
event["revision"],
event["model_epoch"],
f["fragment_no"],
event["content_hash"],
event["title"],
f["text_content"],
value,
event["seq"],
),
)
if force_rollback:
raise RuntimeError("Injected effect/receipt rollback")
receipt = conn.execute(
"INSERT INTO receipts(seq,status,checks,lag_ms) SELECT seq,%s,%s::jsonb,EXTRACT(EPOCH FROM(clock_timestamp()-created_at))*1000 FROM outbox WHERE seq=%s RETURNING *",
("APPLIED" if applied else "STALE", json.dumps(checks), event["seq"]),
).fetchone()
return dict(receipt, logical_writes=len(fragments) if applied else 0)
Step 04 — Run repeated processing without duplicating effects
Idempotency, receipts, and the epoch model
Error: An old model mixes its vectors with the new one
Correction: Check model_epoch and search only the active epoch, with a documented switch.
File src/ged_index_transactionnel/service.py
import json, threading, time
from datetime import datetime
from pathlib import Path
from .schema import (
connection,
migrate_schema,
seed_demo_catalog,
commit_document_change,
rebuild_model_epoch,
)
from .worker import (
claim_event,
compute_embeddings,
conditional_commit,
release_for_redelivery,
)
from .retrieval import search_authorized_exact
from .config import save_json, read_json
def serializable(value):
if isinstance(value, datetime):
return value.isoformat()
if isinstance(value, dict):
return {k: serializable(v) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [serializable(v) for v in value]
return value
def record(path, value):
save_json(path, serializable(value))
def run_worker_loop(name, encoder, worker="sync"):
rows = []
while True:
event = claim_event(name, worker)
if event is None:
break
vectors = compute_embeddings(event, encoder)
rows.append(
{
"event": event,
"fragments": vectors,
"receipt": conditional_commit(name, event, vectors),
}
)
return rows
class PausedWorker:
"""Actual worker thread: committed claim, real encoding, then explicit barrier."""
def __init__(self, name, encoder, seq, worker):
self.name = name
self.encoder = encoder
self.seq = seq
self.worker = worker
self.ready = threading.Event()
self.resume = threading.Event()
self.error = None
self.result = None
self.thread = threading.Thread(target=self.execute, name=worker, daemon=True)
def execute(self):
try:
event = claim_event(self.name, self.worker, self.seq)
if event is None:
raise ValueError("Expected pending event")
vectors = compute_embeddings(event, self.encoder)
self.result = {
"event": event,
"fragments": vectors,
"computed_at_ns": time.monotonic_ns(),
}
self.ready.set()
if not self.resume.wait(120):
raise TimeoutError("Scenario coordinator did not release worker")
self.result["receipt"] = conditional_commit(self.name, event, vectors)
except BaseException as exc:
self.error = exc
self.ready.set()
def start_paused(self):
self.thread.start()
if not self.ready.wait(120):
raise TimeoutError("Worker failed to reach barrier")
if self.error:
raise self.error
return self
def finish(self):
self.resume.set()
self.thread.join(120)
if self.thread.is_alive():
raise TimeoutError("Worker failed to finish")
if self.error:
raise self.error
return self.result
def snapshot_state(name):
with connection(name) as conn, conn.transaction():
conn.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY")
return {
table: conn.execute(
"SELECT * FROM " + table + " ORDER BY " + order
).fetchall()
for table, order in [
("epoch", "active"),
("documents_current", "document_id"),
("document_history", "document_id,revision"),
("outbox", "seq"),
("receipts", "seq"),
("fragments", "document_id,revision,model_epoch,fragment_no"),
]
}
def new_scenario(label, documents):
name = "tp38_" + label
versions = migrate_schema(name)
events = seed_demo_catalog(name, documents)
return name, versions, events
def replay_event(name, encoder, event):
vectors = compute_embeddings(event, encoder)
return {
"event": event,
"fragments": vectors,
"receipt": conditional_commit(name, event, vectors),
}
Step 05 — Observe safety and lag instead of promising real time
Snapshots and commit timeline
Error: A search started before the commit is counted as after it
Correction: Record the snapshot point and explain the scope of the guarantee.
File src/ged_index_transactionnel/evaluate.py
from .config import read_json, save_json
def relevance_metrics(rows, gold):
references = {r["query_id"]: r["relevant"] for r in gold}
recalls = []
empty = []
for row in rows:
relevant = set(references[row["query"]["query_id"]])
found = {r["document_id"] for r in row["search"]["rows"]}
if relevant:
recalls.append(len(relevant & found) / len(relevant))
else:
empty.append(not found)
return {
"queries": len(rows),
"queries_with_relevant": len(recalls),
"queries_without_relevant": len(empty),
"recall_at_5": sum(recalls) / len(recalls) if recalls else None,
"empty_result_rate_without_relevant": (
sum(empty) / len(empty) if empty else None
),
"rejection_policy": "none; top5 may return irrelevant authorized documents",
"search_seconds": [r["search"]["seconds"] for r in rows],
}
def evaluate(config):
baseline = read_json("outputs/baseline.json")
gold = read_json("data/generated/labels/gold.json")
metrics = relevance_metrics(baseline["queries"], gold)
state = baseline["state"]
ids = {f["document_id"] for f in state["fragments"]}
metrics["initial_document_coverage"] = len(ids) / 200
metrics["initial_fragment_count"] = len(state["fragments"])
scenarios = [read_json(f"outputs/scenarios/{i:02d}.json") for i in range(12)]
safety = []
for s in scenarios:
receipts = s["state"]["receipts"]
safety.append(
{
"scenario": s["scenario"],
"timings": s["timings"],
"forbidden": s["forbidden_count"],
"checks": s["checks"],
"applied": sum(r["status"] == "APPLIED" for r in receipts),
"stale": sum(r["status"] == "STALE" for r in receipts),
"duplicate_deliveries": sum(
j["receipt"]["status"] == "DUPLICATE" for j in s["jobs"]
),
"event_created_to_receipt_ms": [
r["lag_ms"] for r in receipts if r["status"] == "APPLIED"
],
"post_commit_search_seconds": s["query"]["seconds"],
}
)
save_json(
"outputs/evaluation.json",
{
"relevance": metrics,
"safety_population": 12,
"safety": safety,
"forbidden_total": sum(s["forbidden"] for s in safety),
"limits": [
"100 synthetic relevance queries and12 fixed schedules are distinct populations",
"event_created_to_receipt_ms starts before source commit; it is not exact commit-to-index latency; includes transaction tail, queue wait, encoding and barriers",
"snapshot taken after recorded source commit; no retroactive erasure of prior responses",
"no response cache, no ANN, no real identity provider",
"ACL update increments revision, temporarily reducing coverage until reindexing; no stale fallback",
"one model epoch can be rebuilt with same frozen model to test state protocol; not a measured model migration",
],
},
)
Step 06 — Delay a job and rebuild the index
Rebuild without reappearance
Error: A global purge deletes volumes from the workstation
Correction: Clean up only this practical exercise's Compose project and named volumes after agreement.
File src/ged_index_transactionnel/exercise.py
from .run import exercise
Sources
pgvector: exact vector search and operators. PostgreSQL 18 SELECT and isolation: materialisation and snapshots. Psycopg3: binary installation, version 3.3.5 pinned here. Pinned E5. Debezium Outbox: source of the pattern; Debezium is not a dependency of this practical exercise.
Precondition for the already-received response scenario
Scenario 6 selects the first document actually returned by authorised search, then revokes its permissions. The dense Top 5 is not required to contain the first document in the corpus. The response and pre-revocation state are retained and checked separately from the following query: a revocation cannot erase a response already delivered to the client. This fixture tests temporal consistency, not ranking relevance.
Authorisation and ranking in the exercise
After rebuilding, I-038 must belong to the admissible candidates for profile 1 and remain excluded from those for profile 0. Being authorised does not guarantee appearing in the Top 5: ranking remains evaluated separately on the 100 queries. The exercise retains candidates and results for both profiles, along with checks for the old job. Evidence is written before any final assertion so that a failure can be inspected.
Limits measured on this corpus
The r003 development run passes 27 tests and 12 scenarios, with no forbidden result in those scenarios. The deliberately naive table returns a withdrawn document. This consistency check does not guarantee relevance: across 43 queries with a relevant reference, average recall at 5 is 0.697674. All 57 queries with no relevant reference receive authorised results, because this practical exercise defines no rejection threshold. A document's admissibility therefore guarantees neither a Top 5 position nor a relevant response.
Development observations
Values recalculated from the raw outputs of this run. They are not improvement thresholds; the workflow from the ZIP recalculates them over the same populations.
{
"status": "PASS",
"documents": 200,
"fragments": 399,
"queries": 100,
"scenarios": 12,
"forbidden": 0,
"relevance": {
"queries": 100,
"queries_with_relevant": 43,
"queries_without_relevant": 57,
"recall_at_5": 0.6976744186046512,
"empty_result_rate_without_relevant": 0.0,
"rejection_policy": "none; top5 may return irrelevant authorized documents",
"search_seconds": [
0.017054370138794184,
0.014858815120533109,
0.014735099859535694,
0.014504152117297053,
0.014803188852965832,
0.01480566500686109,
0.014747765148058534,
0.014471150003373623,
0.014671219047158957,
0.016347639029845595,
0.015716912923380733,
0.01424545212648809,
0.014667331939563155,
0.01463680388405919,
0.014291898813098669,
0.014762402046471834,
0.014397970866411924,
0.014162834966555238,
0.01443811715580523,
0.015051970956847072,
0.014726764988154173,
0.014461921993643045,
0.014894993044435978,
0.014633486978709698,
0.015179472975432873,
0.01620633201673627,
0.014789002016186714,
0.01429196004755795,
0.014399623963981867,
0.0143572639208287,
0.014292029896751046,
0.01434827595949173,
0.0146823909599334,
0.014421104919165373,
0.014470439171418548,
0.014435563003644347,
0.014820301905274391,
0.014615984167903662,
0.0141976207960397,
0.015787536976858974,
0.014346914133056998,
0.014198421966284513,
0.014846672071143985,
0.015042924089357257,
0.01434057205915451,
0.014739780221134424,
0.01414010114967823,
0.014250982087105513,
0.014625121839344501,
0.014820992946624756,
0.014611405087634921,
0.014185687992721796,
0.014868692960590124,
0.01457401504740119,
0.014393582008779049,
0.014304412994533777,
0.014290888095274568,
0.014231255045160651,
0.014433527830988169,
0.014618369052186608,
0.014457924058660865,
0.014126676833257079,
0.014289675047621131,
0.014253918081521988,
0.014481578953564167,
0.01433459110558033,
0.014310816070064902,
0.014589583966881037,
0.014125314075499773,
0.01457697106525302,
0.014417767990380526,
0.014702448854222894,
0.014919290086254478,
0.014304623939096928,
0.014253486879169941,
0.014490996953099966,
0.014279616996645927,
0.014611155027523637,
0.014485295861959457,
0.014198743039742112,
0.014697218080982566,
0.014369997894391418,
0.014320624060928822,
0.014170920010656118,
0.01442647515796125,
0.014156041899695992,
0.014695435063913465,
0.014143798034638166,
0.015373692149296403,
0.014168125111609697,
0.01460188813507557,
0.014278363902121782,
0.014524701051414013,
0.014531494118273258,
0.014358945889398456,
0.014269687933847308,
0.014337607193738222,
0.014196869917213917,
0.014903279952704906,
0.01656491099856794
],
"initial_document_coverage": 1.0,
"initial_fragment_count": 399
},
"actual_database": {
"server_version": "18.6 (Debian 18.6-1.pgdg12+2)",
"pgvector": "0.8.6",
"schema": "tp38_initial"
},
"exercise": {
"version_guard": true,
"hash_guard": true,
"epoch_guard": true,
"not_withdrawn_guard_passes": true,
"denied_profile_no_target": true,
"denied_profile_target_ineligible": true,
"allowed_profile_target_eligible": true,
"main_outputs_unchanged": true,
"rebuild_retained_acl": true
},
"naive_forbidden": 1,
"latency_limit": "outbox creation is before commit; separate scenario1 observed commit-return interval and scenarios3/4 recovery timings are recorded"
}
