Un worker lent peut terminer l'indexation d'une ancienne version après un retrait ou une nouvelle version. Une file au moins une fois peut rejouer le même événement. Un index vectoriel qui ne vérifie pas l'état source peut alors réintroduire une pièce périmée.
Une mini-GED locale en PostgreSQL/pgvector maintient versions et événements, calcule de vrais embeddings et expose des recherches autorisées. Un scénario pas à pas montre les jobs obsolètes rejetés et le retard d'indexation sans ressusciter les pièces retirées.
Télécharger le projet complet, révision r003
Données, corpus et modèle
Vesta-Sync contient 200 textes anglais originaux, vingt dossiers et quatre profils fictifs. Un texte compte un à trois paragraphes conservés comme fragments, soit 399 fragments initiaux. Les références de dossier et termes métier sont des signaux synthétiques forts. Cent requêtes fixées avant exécution ont une pertinence définie par sujet, dossier et droit du profil ; les cas sans document pertinent restent distincts dans les métriques. Aucun label n'est monté dans le service d'inférence.
E5-small est gelé à la révision614241f622f53c4eeff9890bdc4f31cfecc418b3. Les requêtes portent query: , les fragments passage: . Le code charge les tenseurs acquis, vérifie les clés et octets, calcule une moyenne des états réels par attention_mask, puis normalise L2 en 384 dimensions. Les tokens spéciaux réels participent au pooling. Le CLS pooler inutilisé est exclu explicitement. Un fragment dépassant512 tokens est refusé, jamais tronqué silencieusement. Le cosinus mesure une proximité, pas une autorisation.
Une transaction avant, une autre après le calcul
La table documents_current porte version, hash, état de retrait et ACL. Chaque mutation crée une nouvelle révision, garde l'historique et ajoute l'événement outbox dans le même commit. Un rollback doit annuler les deux. La séquence SQL peut avoir des trous après rollback : ce n'est pas une perte d'événement validé.
Un worker réserve brièvement un événement avec verrou et SKIP LOCKED. Il ferme la connexion avant l'inférence E5. La transaction finale prend les verrous dans l'ordre epoch puis document, compare version, hash, retrait et epoch active. Elle insère fragments et reçu atomiquement seulement si tous ces contrôles passent ; sinon le reçu est STALE. Une redélivrance déjà reçue donne DUPLICATE sans nouvel effet. Le calcul peut être refait : on ne promet pas une exécution exactement une fois.
Le worker de démonstration abandonné est récupéré par une action explicite de redélivrance. Il n'existe pas ici de bail expirant automatiquement ni de coordinateur distribué. L'exercice montre les gardes séparément, sans confondre rejet à l'écriture et autorisation à la lecture.
L'ensemble autorisé avant les distances
La recherche lit une transaction read-only repeatable-read. Son premier SELECT fixe le snapshot ; aucun accès source n'a lieu avant lui dans cette transaction. Il suit les commits enregistrés dans les scénarios. Un CTE MATERIALIZED joint les fragments à la version/hash courants, à l'epoch active et aux ACL, en excluant les retraits. Le calcul exact des distances utilise ensuite cet ensemble, et la meilleure distance de fragment donne le classement documentaire avec égalités déterministes.
Le TP conserve EXPLAIN ANALYZE VERBOSE JSON et les vecteurs éligibles pour vérifier le plan et recalculer les distances. Il ne suppose pas que l'ordre textuel des clauses SQL impose l'ordre d'exécution. Aucun HNSW ni cache de réponse n'est utilisé. Un résultat déjà reçu avant une révocation reste dans la mémoire du client ; la garantie ne l'efface pas rétroactivement.
Un retrait est un tombstone, pas un effacement physique. Une modification d'ACL incrémente aussi la révision dans ce modèle minimal. Jusqu'à réindexation, la couverture peut donc baisser même pour un profil encore autorisé. La recherche ne réutilise pas de version périmée pour masquer ce retard.
Démonstration et scénarios
Une table naive_index isolée reçoit réellement un ancien vecteur après retrait ; sa requête dépourvue de jointure source expose la pièce. Les mêmes données soumises au contrôle conditionnel sont refusées. Cette table n'est jamais utilisée par le parcours correct.
Douze schémas indépendants repartent chacun des 200 sources : deux jobs en pause puis retrait ; v2 terminé avant v1 ; duplicata ; abandon après calcul ; rollback effet/reçu ; révocation entre embedding et SELECT ; réponse antérieure conservée puis révocation ; ancienne epoch ; rebuild et ACL ; réservation concurrente du même événement ; rollback source/outbox ; tous les documents interdits au profil. Des threads et barrières explicites reproduisent les ordres, pas des sleeps. Les timeouts des barrières détectent un scénario bloqué et ne définissent pas son ordre.
Ce que mesurent les sorties
Recall@5 est moyenné seulement sur les requêtes ayant un pertinent autorisé. Le nombre sans pertinent et leur taux de réponse vide sont séparés. Sans seuil de rejet, une requête hors corpus peut renvoyer des documents autorisés mais non pertinents. La couverture initiale rapporte les documents et fragments effectivement indexés.
La sûreté est mesurée sur les douze scénarios, indépendamment des cent requêtes de pertinence. Le critère est zéro résultat interdit au snapshot post-commit sur ce périmètre. Les reçus APPLIED/STALE, redélivrances DUPLICATE, temps de requête et délais création de ligne outbox→reçu sont conservés. La création de ligne précède son commit : ce délai n’est donc pas une mesure exacte commit→index, et inclut aussi cette portion de transaction. Les délais incluent attente dans la file, calcul et pauses imposées ; ils ne prédisent pas un SLA de production. Le scénario1 mesure aussi les deux retours client de commit (v2 puis index) avec une horloge monotone ; cette durée observée inclut l'encodage et la coordination, sans prétendre observer l'instant interne WAL. Les scénarios 3et4 chronomètrent la redélivrance, son recalcul et le commit final. Les coûts de récupération doivent être lus dans la chronologie du scénario, sans confondre absence de doublon et absence de recalcul.
L'exercice retient v1, change le texte, révoque un profil puis reconstruit l'epoch avec le même E5 gelé. Version, hash et epoch refusent l'ancien job ; le retrait n'est pas actif et cette garde passe. La recherche distingue ensuite le profil révoqué du profil toujours autorisé. Le rebuild teste un protocole d'état ; il ne mesure pas une migration vers un autre modèle.
Étape 01 — Construire le catalogue, les versions et l'outbox
Atomicité SQL et schéma source
Erreur : L'événement est écrit après le commit documentaire
Correction : Déplacer les deux écritures dans la même transaction et tester un rollback.
Fichier 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},
}
Étape 02 — Calculer des embeddings réels et une recherche exacte
Préfixes E5 et sous-ensemble autorisé
Erreur : Le vecteur porte des ACL copiées devenues fausses
Correction : Joindre l'état source à chaque SELECT ; ne pas déléguer la décision au vecteur.
Fichier 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,
}
Étape 03 — Dérouler une écriture de worker obsolète
Concurrence et comparaison de révision
Erreur : Le worker garde une transaction ouverte pendant l'inférence
Correction : Libérer avant calcul, puis verrouiller/vérifier dans une transaction finale courte.
Fichier 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)
Étape 04 — Exécuter le traitement répété sans doublonner les effets
Idempotence, reçus et modèle epoch
Erreur : Un ancien modèle mélange ses vecteurs avec le nouveau
Correction : Vérifier model_epoch et rechercher uniquement dans l'epoch active, avec bascule documentée.
Fichier 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),
}
Étape 05 — Observer sûreté et retard au lieu de promettre temps réel
Snapshots et chronologie des commits
Erreur : Une recherche partie avant le commit est comptée après
Correction : Enregistrer le point de snapshot et expliquer le périmètre de la garantie.
Fichier 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",
],
},
)
Étape 06 — Retarder un job et reconstruire l'index
Rebuild sans réapparition
Erreur : Une purge globale supprime les volumes du poste
Correction : Nettoyer uniquement le projet Compose de ce TP et ses volumes nommés après accord.
Fichier src/ged_index_transactionnel/exercise.py
from .run import exercise
Sources
pgvector : recherche vectorielle exacte et opérateurs. PostgreSQL18 SELECT et isolation : matérialisation et snapshots. Psycopg3 : installation binaire, version 3.3.5 verrouillée ici. E5 figé. Outbox Debezium : source du pattern ; Debezium n'est pas une dépendance du TP.
Précondition du scénario de réponse déjà reçue
Le scénario 6 choisit le premier document réellement retourné par la recherche autorisée, puis révoque ses droits. Le Top 5 dense n’est pas obligé de contenir le premier document du corpus. La réponse et l’état avant révocation sont conservés et vérifiés séparément de la requête suivante : une révocation ne peut pas effacer une réponse déjà remise au client. Cette fixture teste la cohérence temporelle, pas la pertinence du classement.
Autorisation et classement dans l’exercice
Après reconstruction, I-038 doit appartenir aux candidats admissibles du profil 1 et rester exclu de ceux du profil 0. Être autorisé ne garantit pas de figurer dans le Top 5 : le classement reste évalué séparément sur les 100 requêtes. L’exercice conserve les candidats et les résultats des deux profils, ainsi que les contrôles du job ancien. Les preuves sont écrites avant toute assertion finale afin de rendre un échec inspectable.
Limites mesurées sur ce corpus
Le parcours de développement r003 réussit 27 tests et 12 scénarios, sans résultat interdit dans ces scénarios. La table volontairement naïve retourne une pièce retirée. Ce contrôle de cohérence ne garantit pas la pertinence : sur 43 requêtes possédant une référence pertinente, le rappel moyen à 5 vaut 0,697674. Les 57 requêtes sans référence pertinente reçoivent toutes des résultats autorisés, puisque ce TP ne définit aucun seuil de rejet. L’admissibilité d’une pièce ne garantit donc ni sa présence dans le Top 5 ni la pertinence de sa réponse.
Observations de développement
Valeurs recalculées depuis les sorties brutes de ce parcours. Elles ne sont pas des seuils de gain ; le parcours depuis le ZIP les recalcule sur les mêmes 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"
}
