Un worker lento puede terminar la indexación de una versión antigua tras una retirada o versión nueva. Una cola al menos una vez puede reproducir el mismo evento. Un índice vectorial que no verifica el estado fuente puede reintroducir un documento obsoleto.
Una mini-GED local en PostgreSQL/pgvector mantiene versiones y eventos, calcula embeddings reales y expone búsquedas autorizadas. Un escenario paso a paso muestra trabajos obsoletos rechazados y retraso de indexación sin resucitar documentos retirados.
Descargar el proyecto completo, revisión r003
Datos, corpus y modelo
Vesta-Sync contiene 200 textos ingleses originales, veinte expedientes y cuatro perfiles ficticios. Cada texto cuenta uno a tres párrafos conservados como fragmentos, 399 fragmentos iniciales. Referencias de expediente y términos de negocio son señales sintéticas fuertes. Cien consultas fijadas antes de ejecución tienen pertinencia definida por tema, expediente y derecho de perfil; casos sin documento pertinente se separan en métricas. Ningún label se monta en inferencia.
E5-small queda congelado en revisión 614241f622f53c4eeff9890bdc4f31cfecc418b3. Consultas llevan query: y fragmentos passage: . El código carga tensores adquiridos, verifica claves y bytes, calcula media de estados reales por attention_mask y normaliza L2 en 384 dimensiones. Tokens especiales reales participan en pooling. El pooler CLS sin usar se excluye explícitamente. Un fragmento que supera 512 tokens se rechaza sin truncar. El coseno mide proximidad, no autorización.
Una transacción antes y otra después del cálculo
La tabla documents_current lleva versión, hash, estado de retirada y ACL. Cada mutación crea revisión, conserva historial y añade evento outbox en el mismo commit. Un rollback debe anular ambos. La secuencia SQL puede tener huecos tras rollback: no implica pérdida de evento validado.
Un worker reserva brevemente evento con bloqueo y SKIP LOCKED y cierra conexión antes de inferencia E5. La transacción final bloquea en orden epoch y documento, compara versión, hash, retirada y epoch activa. Inserta fragmentos y recibo atómicamente solo si pasan controles; de otro modo recibo es STALE. Una redelivery recibida es DUPLICATE sin efecto nuevo. El cálculo puede repetirse: no se promete exactamente una vez.
El worker de demostración abandonado se recupera mediante acción explícita de redelivery. No existe aquí lease que expire automáticamente ni coordinador distribuido. El ejercicio muestra guardas por separado, sin confundir rechazo de escritura con autorización de lectura.
El conjunto autorizado antes de las distancias
La búsqueda lee una transacción read-only repeatable-read. Su primer SELECT fija snapshot; antes no hay acceso fuente en esa transacción. Un CTE MATERIALIZED une fragmentos a versión/hash actual, epoch activa y ACL, excluyendo retiradas. El cálculo exacto de distancias usa este conjunto y la mejor distancia de fragmento clasifica documentos con empates deterministas.
El TP conserva EXPLAIN ANALYZE VERBOSE JSON y vectores elegibles para verificar plan y recalcular distancias. No supone que orden textual de cláusulas SQL imponga orden de ejecución. No usa HNSW ni caché de respuesta. Un resultado recibido antes de revocación permanece en memoria cliente; la garantía no lo borra retroactivamente.
Una retirada es tombstone, no borrado físico. Una ACL también incrementa revisión en este modelo mínimo. Hasta reindexación puede bajar cobertura incluso para perfil aún autorizado. La búsqueda no reutiliza versión obsoleta para ocultar retraso.
Demostración y escenarios
Una tabla naive_index aislada recibe realmente vector antiguo tras retirada; su consulta sin unión fuente expone el documento. Los mismos datos sometidos al control condicional se rechazan. Esta tabla nunca se usa en el recorrido correcto.
Doce esquemas independientes parten de las 200 fuentes: jobs pausados y retirada, v2 antes de v1, duplicado, abandono tras cálculo, rollback, revocación entre embedding y SELECT, respuesta previa y revocación, epoch antigua, rebuild y ACL, reserva concurrente, rollback fuente/outbox y documentos prohibidos. Threads y barreras explícitas reproducen órdenes, no sleeps. Timeouts de barrera detectan escenario bloqueado y no definen orden.
Qué miden las salidas
Recall@5 se promedia solo en consultas con pertinente autorizado. Número sin pertinente y tasa de respuesta vacía se separan. Sin umbral de rechazo, consulta fuera de corpus puede devolver documentos autorizados no pertinentes. Cobertura inicial informa documentos y fragmentos indexados.
La seguridad se mide en doce escenarios independientemente de cien consultas de pertinencia. Criterio: cero resultado prohibido en snapshot post-commit. Se conservan recibos APPLIED/STALE, redeliveries DUPLICATE, tiempos de consulta y demora outbox→recibo. La línea outbox precede commit y la demora no mide exactamente commit→índice; incluye parte de transacción, espera de cola, cálculo y pausas. No predice SLA. Escenario 1 mide retornos cliente v2 e índice con reloj monotónico, incluidos encoding y coordinación, sin afirmar observar instante WAL. Escenarios 3 y 4 cronometran redelivery, recálculo y commit final.
El ejercicio retiene v1, cambia texto, revoca un perfil y reconstruye epoch con el mismo E5 congelado. Versión, hash y epoch rechazan trabajo antiguo; retirada no está activa y esa guarda pasa. La búsqueda distingue perfil revocado del aún autorizado. Rebuild prueba protocolo de estado, no migración a otro modelo.
Paso 01 — Construir catálogo, versiones y outbox
Atomicidad SQL y esquema fuente
Error: El evento se escribe después del commit documental
Corrección: Mover ambas escrituras a la misma transacción y probar rollback.
Archivo 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},
}
Paso 02 — Calcular embeddings reales y búsqueda exacta
Prefijos E5 y subconjunto autorizado
Error: El vector lleva ACL copiadas que se volvieron falsas
Corrección: Unir estado fuente a cada SELECT; no delegar decisión al vector.
Archivo 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,
}
Paso 03 — Desarrollar una escritura de worker obsoleta
Concurrencia y comparación de revisión
Error: El worker mantiene transacción abierta durante inferencia
Corrección: Liberar antes de calcular y bloquear/verificar en transacción final corta.
Archivo 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)
Paso 04 — Ejecutar el procesamiento repetido sin duplicar efectos
Idempotencia, recibos y modelo de epoch
Error: Un modelo antiguo mezcla sus vectores con el nuevo
Corrección: Verificar model_epoch y buscar solo en epoch activa, con cambio documentado.
Archivo 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),
}
Paso 05 — Observar seguridad y retraso en vez de prometer tiempo real
Snapshots y cronología de commits
Error: Una búsqueda iniciada antes del commit se cuenta después
Corrección: Registrar punto de snapshot y explicar el perímetro de garantía.
Archivo 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",
],
},
)
Paso 06 — Retrasar un job y reconstruir el índice
Rebuild sin reaparición
Error: Una purga global elimina volúmenes del equipo
Corrección: Limpiar solo el proyecto Compose de este TP y sus volúmenes nombrados tras acuerdo.
Archivo src/ged_index_transactionnel/exercise.py
from .run import exercise
Fuentes
pgvector: búsqueda vectorial exacta y operadores. PostgreSQL18 SELECT y aislamiento: materialización y snapshots. Psycopg3: instalación binaria, versión 3.3.5 bloqueada aquí. E5 congelado. Outbox Debezium: fuente del patrón; Debezium no es dependencia del TP.
Precondición del escenario de respuesta ya recibida
El escenario 6 elige el primer documento realmente devuelto por búsqueda autorizada y después revoca sus derechos. Top 5 denso no debe contener el primer documento del corpus. La respuesta y estado antes de revocación se conservan y verifican aparte de la consulta siguiente: una revocación no puede borrar una respuesta ya entregada. Esta fixture prueba coherencia temporal, no pertinencia de clasificación.
Autorización y clasificación en el ejercicio
Tras reconstrucción, I-038 debe pertenecer a candidatos admisibles del perfil 1 y seguir excluido de los del perfil 0. Estar autorizado no garantiza figurar en Top 5: clasificación se evalúa aparte sobre 100 consultas. El ejercicio conserva candidatos y resultados de ambos perfiles, además de controles del job antiguo. Las pruebas se escriben antes de toda afirmación final para hacer un fallo inspeccionable.
Límites medidos en este corpus
El recorrido de desarrollo r003 supera 27 pruebas y 12 escenarios, sin resultado prohibido en ellos. La tabla deliberadamente ingenua devuelve un documento retirado. Este control de coherencia no garantiza pertinencia: en 43 consultas con referencia pertinente, recall medio a 5 es 0,697674. Las 57 consultas sin referencia pertinente reciben todas resultados autorizados porque el TP no define umbral de rechazo. La admisibilidad no garantiza presencia en Top 5 ni pertinencia.
Observaciones de desarrollo
Valores recalculados a partir de las salidas brutas de este recorrido. No son umbrales de ganancia; el recorrido desde el ZIP los recalcula sobre las mismas poblaciones.
{
"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"
}
