Un worker lento può completare l'indicizzazione di una vecchia versione dopo un ritiro o un aggiornamento. Una coda con consegna almeno una volta può riprodurre lo stesso evento. Un indice vettoriale che non controlla lo stato della sorgente può quindi reintrodurre un documento obsoleto.
Una mini-GED locale in PostgreSQL/pgvector conserva versioni ed eventi, calcola embedding reali ed espone ricerche autorizzate. Uno scenario passo passo mostra i job obsoleti rifiutati e il ritardo di indicizzazione senza far ricomparire i documenti ritirati.
Scaricare il progetto completo, revisione r003
Dati, corpus e modello
Vesta-Sync contiene 200 testi originali in inglese, venti cartelle e quattro profili fittizi. Un testo contiene da uno a tre paragrafi conservati come frammenti, per 399 frammenti iniziali. I riferimenti alle cartelle e i termini professionali sono segnali sintetici forti. Cento query fissate prima dell'esecuzione hanno una pertinenza definita per argomento, cartella e diritto del profilo; i casi senza documento pertinente restano distinti nelle metriche. Nessuna etichetta viene caricata nel servizio di inferenza.
E5-small è fissato alla revisione614241f622f53c4eeff9890bdc4f31cfecc418b3. Le query usano query: , i frammenti passage: . Il codice carica i tensori acquisiti, controlla chiavi e byte, calcola la media degli stati reali tramite attention_mask, poi normalizza L2 in 384 dimensioni. I token speciali reali partecipano al pooling. Il pooler CLS inutilizzato viene escluso esplicitamente. Un frammento oltre512 token viene rifiutato, mai troncato in silenzio. Il coseno misura una vicinanza, non un'autorizzazione.
Una transazione prima, un'altra dopo il calcolo
La tabella documents_current contiene versione, hash, stato di ritiro e ACL. Ogni modifica crea una nuova revisione, conserva lo storico e aggiunge l'evento outbox nello stesso commit. Un rollback deve annullare entrambe le scritture. La sequenza SQL può avere dei vuoti dopo un rollback: non è una perdita di evento convalidato.
Un worker riserva brevemente un evento con lock e SKIP LOCKED. Chiude la connessione prima dell'inferenza E5. La transazione finale acquisisce i lock nell'ordine epoch poi documento, confronta versione, hash, ritiro ed epoch attiva. Inserisce frammenti e ricevuta in modo atomico solo se tutti questi controlli passano; altrimenti la ricevuta è STALE. Una riconsegna già ricevuta dà DUPLICATE senza nuovo effetto. Il calcolo può essere ripetuto: non si promette un'esecuzione esattamente una volta.
Il worker dimostrativo abbandonato viene recuperato con un'azione esplicita di riconsegna. Non esistono qui né un lease che scade automaticamente né un coordinatore distribuito. L'esercizio mostra le protezioni separatamente, senza confondere il rifiuto in scrittura con l'autorizzazione in lettura.
L'insieme autorizzato prima delle distanze
La ricerca legge una transazione read-only repeatable-read. Il primo SELECT fissa lo snapshot; in questa transazione nessun accesso alla sorgente avviene prima. Segue i commit registrati negli scenari. Un CTE MATERIALIZED unisce i frammenti alla versione/hash corrente, all'epoch attiva e agli ACL, escludendo i ritiri. Il calcolo esatto delle distanze usa poi questo insieme, e la migliore distanza di frammento stabilisce la classifica dei documenti con parità deterministiche.
Il TP conserva EXPLAIN ANALYZE VERBOSE JSON e i vettori ammissibili per verificare il piano e ricalcolare le distanze. Non presume che l'ordine testuale delle clausole SQL imponga l'ordine di esecuzione. Non usa HNSW né cache delle risposte. Un risultato ricevuto prima di una revoca resta nella memoria del client; la garanzia non lo cancella retroattivamente.
Un ritiro è un tombstone, non una cancellazione fisica. In questo modello minimo anche una modifica dell'ACL incrementa la revisione. Fino alla reindicizzazione, la copertura può quindi diminuire perfino per un profilo ancora autorizzato. La ricerca non riusa una versione obsoleta per mascherare questo ritardo.
Dimostrazione e scenari
Una tabella naive_index isolata riceve davvero un vecchio vettore dopo il ritiro; la sua query senza join alla sorgente espone il documento. Gli stessi dati sottoposti al controllo condizionale vengono rifiutati. Questa tabella non è mai usata dal percorso corretto.
Dodici schemi indipendenti ripartono ciascuno dalle 200 sorgenti: due job in pausa poi ritiro; v2 conclusa prima di v1; duplicato; abbandono dopo il calcolo; rollback di effetto/ricevuta; revoca tra embedding e SELECT; risposta precedente conservata poi revoca; vecchia epoch; rebuild e ACL; prenotazione concorrente dello stesso evento; rollback sorgente/outbox; tutti i documenti vietati al profilo. Thread e barriere esplicite riproducono gli ordini, non sleep. I timeout delle barriere rilevano uno scenario bloccato e non ne definiscono l'ordine.
Cosa misurano le uscite
Recall@5 viene mediato solo sulle query con un pertinente autorizzato. Il numero di query senza pertinente e il loro tasso di risposta vuota restano separati. Senza soglia di rifiuto, una query esterna al corpus può restituire documenti autorizzati ma non pertinenti. La copertura iniziale riporta i documenti e i frammenti effettivamente indicizzati.
La sicurezza viene misurata sui dodici scenari, indipendentemente dalle cento query di pertinenza. Il criterio è zero risultati vietati nello snapshot post-commit su questo perimetro. Vengono conservati ricevute APPLIED/STALE, riconsegne DUPLICATE, tempi di query e ritardi dalla creazione della riga outbox alla ricevuta. La riga viene creata prima del commit: questo ritardo non misura quindi esattamente commit→index e include anche questa parte della transazione. I ritardi includono attesa in coda, calcolo e pause imposte; non predicono uno SLA di produzione. Lo scenario1 misura anche i due ritorni al client del commit (v2 poi index) con un orologio monotono; questa durata osservata include codifica e coordinamento, senza pretendere di osservare l'istante interno del WAL. Gli scenari 3e4 misurano la riconsegna, il suo ricalcolo e il commit finale. I costi di recupero vanno letti nella cronologia dello scenario, senza confondere l'assenza di duplicato con l'assenza di ricalcolo.
L'esercizio trattiene v1, modifica il testo, revoca un profilo e poi ricostruisce l'epoch con lo stesso E5 fissato. Versione, hash ed epoch rifiutano il vecchio job; il ritiro non è attivo e questo controllo passa. La ricerca distingue poi il profilo revocato da quello ancora autorizzato. Il rebuild verifica un protocollo di stato; non misura una migrazione verso un altro modello.
Passo 01 — Costruire il catalogo, le versioni e l'outbox
Atomicità SQL e schema sorgente
Errore: l'evento viene scritto dopo il commit del documento
Correzione: spostare le due scritture nella stessa transazione e verificare un 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},
}
Passo 02 — Calcolare embedding reali e una ricerca esatta
Prefissi E5 e sottoinsieme autorizzato
Errore: il vettore contiene ACL copiate che sono diventate errate
Correzione: unire lo stato sorgente a ogni SELECT; non delegare la decisione al vettore.
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,
}
Passo 03 — Svolgere una scrittura di worker obsoleta
Concorrenza e confronto delle revisioni
Errore: il worker mantiene una transazione aperta durante l'inferenza
Correzione: rilasciare prima del calcolo, poi bloccare e verificare in una breve transazione finale.
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)
Passo 04 — Eseguire il trattamento ripetuto senza duplicare gli effetti
Idempotenza, ricevute ed epoch del modello
Errore: un vecchio modello mescola i suoi vettori con quelli nuovi
Correzione: verificare model_epoch e cercare solo nell'epoch attiva, con una commutazione documentata.
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),
}
Passo 05 — Osservare sicurezza e ritardo invece di promettere il tempo reale
Snapshot e cronologia dei commit
Errore: una ricerca iniziata prima del commit viene conteggiata dopo
Correzione: registrare il punto dello snapshot e spiegare il perimetro della garanzia.
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",
],
},
)
Passo 06 — Ritardare un job e ricostruire l'indice
Ricostruire senza ricomparse
Errore: una pulizia globale elimina i volumi del computer
Correzione: pulire solo il progetto Compose di questo TP e i suoi volumi con nome, dopo accordo.
File src/ged_index_transactionnel/exercise.py
from .run import exercise
Fonti
pgvector: ricerca vettoriale esatta e operatori. PostgreSQL18 SELECT e isolamento: materializzazione e snapshot. Psycopg3: installazione binaria, versione 3.3.5 fissata qui. E5 fissato. Outbox Debezium: fonte del pattern; Debezium non è una dipendenza del TP.
Precondizione dello scenario con risposta già ricevuta
Lo scenario 6 sceglie il primo documento realmente restituito dalla ricerca autorizzata, poi ne revoca i diritti. La Top 5 densa non è obbligata a contenere il primo documento del corpus. La risposta e lo stato prima della revoca vengono conservati e verificati separatamente dalla query successiva: una revoca non può cancellare una risposta già consegnata al cliente. Questa fixture verifica la coerenza temporale, non la pertinenza della classifica.
Autorizzazione e classifica nell'esercizio
Dopo la ricostruzione, I-038 deve appartenere ai candidati ammissibili del profilo 1 e restare escluso da quelli del profilo 0. Essere autorizzati non garantisce di comparire nella Top 5: la classifica resta valutata separatamente sulle 100 query. L'esercizio conserva candidati e risultati dei due profili, oltre ai controlli sul vecchio job. Le prove vengono scritte prima di qualunque affermazione finale per rendere un fallimento ispezionabile.
Limiti misurati su questo corpus
Il percorso di sviluppo r003 supera 27 test e 12 scenari, senza risultati vietati in questi scenari. La tabella volutamente ingenua restituisce un documento ritirato. Questo controllo di coerenza non garantisce la pertinenza: su 43 query con una referenza pertinente, il recall medio a 5 è 0,697674. Tutte le 57 query senza referenza pertinente ricevono risultati autorizzati, perché questo TP non definisce una soglia di rifiuto. L'ammissibilità di un documento non garantisce quindi né la sua presenza nella Top 5 né la pertinenza della sua risposta.
Osservazioni sullo sviluppo
Valori ricalcolati dalle uscite grezze di questo percorso. Non sono soglie di guadagno; il percorso dal ZIP li ricalcola sulle stesse popolazioni.
{
"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"
}
