Một worker chậm có thể hoàn tất việc lập chỉ mục của một phiên bản cũ sau khi tài liệu bị rút lại hoặc có phiên bản mới. Một hàng đợi at-least-once có thể phát lại cùng một sự kiện. Khi đó, một chỉ mục vector không kiểm tra trạng thái nguồn có thể đưa trở lại một tài liệu đã lỗi thời.
Một mini-GED cục bộ bằng PostgreSQL/pgvector duy trì các phiên bản và sự kiện, tính embedding thật và cung cấp các tìm kiếm được cấp quyền. Một kịch bản từng bước cho thấy các job lỗi thời bị từ chối và độ trễ lập chỉ mục không làm hồi sinh các tài liệu đã rút.
Tải dự án hoàn chỉnh, bản sửa đổi r003
Dữ liệu, kho dữ liệu và mô hình
Vesta-Sync có 200 văn bản tiếng Anh gốc, hai mươi hồ sơ và bốn hồ sơ người dùng hư cấu. Một văn bản có từ một đến ba đoạn được giữ làm fragment, tổng cộng 399 fragment ban đầu. Các tham chiếu hồ sơ và thuật ngữ nghiệp vụ là tín hiệu tổng hợp mạnh. Một trăm truy vấn được cố định trước khi chạy có độ liên quan xác định theo chủ đề, hồ sơ và quyền của người dùng; các trường hợp không có tài liệu liên quan vẫn tách riêng trong chỉ số. Không label nào được nạp vào dịch vụ suy luận.
E5-small được cố định tại revision 614241f622f53c4eeff9890bdc4f31cfecc418b3. Truy vấn có tiền tố query: , fragment có tiền tố passage: . Mã tải các tensor đã lấy, kiểm tra khóa và byte, tính trung bình các trạng thái thực theo attention_mask, rồi chuẩn hóa L2 trong 384 chiều. Token đặc biệt thực sự tham gia pooling. CLS pooler không dùng được loại trừ rõ ràng. Một fragment vượt quá 512 token bị từ chối, không bao giờ bị cắt ngầm. Cosine đo độ gần, không phải quyền truy cập.
Một giao dịch trước, một giao dịch sau khi tính toán
Bảng documents_current lưu phiên bản, hash, trạng thái rút lại và ACL. Mỗi thay đổi tạo một revision mới, giữ lịch sử và thêm sự kiện outbox trong cùng commit. Rollback phải hủy cả hai. Chuỗi SQL có thể có khoảng trống sau rollback: đó không phải là mất một sự kiện đã xác thực.
Một worker giữ chỗ ngắn cho một sự kiện bằng khóa và SKIP LOCKED. Nó đóng kết nối trước suy luận E5. Giao dịch cuối lấy khóa theo thứ tự epoch rồi tài liệu, so sánh phiên bản, hash, trạng thái rút lại và epoch đang hoạt động. Nó chỉ chèn fragment và receipt một cách nguyên tử khi mọi kiểm tra đều đạt; nếu không receipt là STALE. Một lần gửi lại sự kiện đã có receipt trả về DUPLICATE mà không có hiệu ứng mới. Việc tính có thể thực hiện lại: không có cam kết thực thi đúng một lần.
Worker trình diễn bị bỏ dở được khôi phục bằng một thao tác gửi lại rõ ràng. Ở đây không có lease tự hết hạn hay điều phối viên phân tán. Bài tập trình bày riêng các guard, không nhầm lẫn việc từ chối khi ghi với việc cấp quyền khi đọc.
Tập được cấp quyền trước khi tính khoảng cách
Tìm kiếm đọc trong một giao dịch read-only repeatable-read. SELECT đầu tiên cố định snapshot; không có truy cập nguồn nào diễn ra trước nó trong giao dịch này. Nó tuân theo các commit được ghi lại trong kịch bản. Một CTE MATERIALIZED nối các fragment với phiên bản/hash hiện tại, epoch đang hoạt động và ACL, đồng thời loại bỏ tài liệu đã rút. Sau đó, phép tính khoảng cách chính xác dùng tập này; khoảng cách fragment tốt nhất xác định thứ hạng tài liệu với các trường hợp bằng nhau có tính xác định.
Bài thực hành giữ lại EXPLAIN ANALYZE VERBOSE JSON và các vector đủ điều kiện để kiểm tra plan và tính lại khoảng cách. Nó không giả định thứ tự văn bản của các mệnh đề SQL áp đặt thứ tự thực thi. Không dùng HNSW hay cache phản hồi. Một kết quả đã gửi trước khi thu hồi quyền vẫn nằm trong bộ nhớ của khách hàng; bảo đảm này không xóa nó hồi tố.
Một lần rút là tombstone, không phải xóa vật lý. Thay đổi ACL cũng tăng revision trong mô hình tối thiểu này. Trước khi lập chỉ mục lại, độ bao phủ có thể giảm ngay cả với một hồ sơ vẫn được cấp quyền. Tìm kiếm không tái sử dụng phiên bản lỗi thời để che giấu độ trễ đó.
Trình diễn và các kịch bản
Một bảng naive_index tách biệt thực sự nhận một vector cũ sau khi tài liệu bị rút; truy vấn không có phép nối nguồn của nó làm lộ tài liệu đó. Dữ liệu tương tự, khi chịu kiểm soát có điều kiện, sẽ bị từ chối. Bảng này không bao giờ được dùng trong luồng đúng.
Mười hai schema độc lập, mỗi schema bắt đầu lại từ 200 nguồn: hai job tạm dừng rồi rút; v2 hoàn tất trước v1; trùng lặp; bỏ dở sau tính toán; rollback hiệu ứng/receipt; thu hồi giữa embedding và SELECT; giữ phản hồi cũ rồi thu hồi; epoch cũ; rebuild và ACL; giữ chỗ đồng thời cùng một sự kiện; rollback nguồn/outbox; mọi tài liệu bị cấm cho hồ sơ. Thread và barrier rõ ràng tái hiện thứ tự, không dùng sleep. Timeout của barrier phát hiện kịch bản bị treo chứ không xác định thứ tự.
Các đầu ra đo lường điều gì
Recall@5 chỉ được lấy trung bình trên các truy vấn có một tài liệu liên quan được cấp quyền. Số truy vấn không có tài liệu liên quan và tỷ lệ phản hồi trống của chúng được tách riêng. Không có ngưỡng từ chối, một truy vấn ngoài kho có thể trả về các tài liệu được cấp quyền nhưng không liên quan. Độ bao phủ ban đầu báo cáo các tài liệu và fragment thực sự được lập chỉ mục.
Độ an toàn được đo trên mười hai kịch bản, độc lập với một trăm truy vấn về độ liên quan. Tiêu chí là không có kết quả bị cấm trong snapshot sau commit trong phạm vi này. Receipt APPLIED/STALE, lần gửi lại DUPLICATE, thời gian truy vấn và độ trễ từ lúc tạo dòng outbox đến receipt được lưu lại. Việc tạo dòng diễn ra trước commit, nên độ trễ này không phải thước đo chính xác từ commit đến index và còn gồm phần giao dịch đó. Độ trễ gồm chờ hàng đợi, tính toán và các lần tạm dừng bắt buộc; chúng không dự báo SLA sản xuất. Kịch bản 1 cũng đo hai lần phản hồi commit cho khách hàng (v2 rồi index) bằng đồng hồ đơn điệu; thời lượng quan sát được gồm mã hóa và điều phối, không tuyên bố quan sát thời điểm WAL nội bộ. Kịch bản 3 và 4 đo việc gửi lại, tính lại và commit cuối. Chi phí khôi phục cần được đọc theo dòng thời gian kịch bản, không nhầm việc không có bản sao với việc không cần tính lại.
Bài tập giữ v1, đổi văn bản, thu hồi một hồ sơ rồi rebuild epoch với cùng E5 đã cố định. Phiên bản, hash và epoch từ chối job cũ; trạng thái rút không hoạt động nên guard này đạt. Sau đó, tìm kiếm phân biệt hồ sơ bị thu hồi với hồ sơ vẫn được cấp quyền. Rebuild kiểm tra một giao thức trạng thái; nó không đo việc di chuyển sang mô hình khác.
Bước 01 — Xây dựng danh mục, các phiên bản và outbox
Tính nguyên tử SQL và schema nguồn
Lỗi: Sự kiện được ghi sau commit của tài liệu
Khắc phục: Đưa hai lần ghi vào cùng giao dịch và kiểm thử rollback.
Tệp 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},
}
Bước 02 — Tính embedding thật và tìm kiếm chính xác
Tiền tố E5 và tập con được cấp quyền
Lỗi: Vector mang ACL đã sao chép nhưng nay không còn đúng
Khắc phục: Nối trạng thái nguồn trong mỗi SELECT; không giao quyết định cho vector.
Tệp 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,
}
Bước 03 — Diễn lại thao tác ghi của một worker lỗi thời
Đồng thời và so sánh revision
Lỗi: Worker giữ một giao dịch mở trong lúc suy luận
Khắc phục: Giải phóng trước khi tính, rồi khóa/kiểm tra trong một giao dịch cuối ngắn.
Tệp 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)
Bước 04 — Thực hiện xử lý lặp lại mà không nhân đôi hiệu ứng
Tính idempotent, receipt và model epoch
Lỗi: Một mô hình cũ trộn vector của nó với mô hình mới
Khắc phục: Kiểm tra model_epoch và chỉ tìm trong epoch đang hoạt động, với việc chuyển đổi được ghi rõ.
Tệp 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),
}
Bước 05 — Quan sát độ an toàn và độ trễ thay vì hứa hẹn thời gian thực
Snapshot và dòng thời gian của các commit
Lỗi: Một tìm kiếm bắt đầu trước commit lại được tính là sau commit
Khắc phục: Ghi lại thời điểm snapshot và giải thích phạm vi của bảo đảm.
Tệp 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",
],
},
)
Bước 06 — Làm chậm một job và rebuild chỉ mục
Rebuild không tái xuất hiện
Lỗi: Một thao tác dọn dẹp toàn cục xóa các volume của máy
Khắc phục: Chỉ dọn dự án Compose của bài thực hành này và các volume được đặt tên của nó sau khi có đồng ý.
Tệp src/ged_index_transactionnel/exercise.py
from .run import exercise
Nguồn
pgvector: tìm kiếm vector chính xác và các toán tử. PostgreSQL18 SELECT và isolation: hiện thực hóa và snapshot. Psycopg3: cài đặt nhị phân, phiên bản 3.3.5 được khóa ở đây. E5 cố định. Outbox Debezium: nguồn của pattern; Debezium không phải là phụ thuộc của bài thực hành.
Điều kiện trước của kịch bản phản hồi đã nhận
Kịch bản 6 chọn tài liệu đầu tiên thực sự được trả về từ tìm kiếm được cấp quyền, rồi thu hồi quyền của nó. Top 5 dense không bắt buộc chứa tài liệu đầu tiên của kho. Phản hồi và trạng thái trước khi thu hồi được giữ lại và kiểm tra tách biệt với truy vấn tiếp theo: việc thu hồi không thể xóa phản hồi đã gửi cho khách hàng. Fixture này kiểm tra tính nhất quán theo thời gian, không phải độ liên quan của thứ hạng.
Cấp quyền và xếp hạng trong bài tập
Sau rebuild, I-038 phải thuộc các ứng viên đủ điều kiện của hồ sơ 1 và vẫn bị loại khỏi hồ sơ 0. Được cấp quyền không bảo đảm xuất hiện trong Top 5: thứ hạng vẫn được đánh giá riêng trên 100 truy vấn. Bài tập giữ các ứng viên và kết quả của hai hồ sơ, cũng như các kiểm tra của job cũ. Bằng chứng được ghi trước mọi assertion cuối để một lỗi có thể được kiểm tra.
Các giới hạn được đo trên kho dữ liệu này
Lượt chạy phát triển r003 thành công 27 kiểm thử và 12 kịch bản, không có kết quả bị cấm trong các kịch bản đó. Bảng cố ý ngây thơ trả về một tài liệu đã rút. Kiểm soát nhất quán này không bảo đảm độ liên quan: trên 43 truy vấn có tham chiếu liên quan, recall trung bình tại 5 là 0,697674. Cả 57 truy vấn không có tham chiếu liên quan đều nhận các kết quả được cấp quyền, vì bài thực hành này không xác định ngưỡng từ chối. Tính đủ điều kiện của tài liệu do đó không bảo đảm nó xuất hiện trong Top 5 hay phản hồi của nó liên quan.
Quan sát trong quá trình phát triển
Các giá trị được tính lại từ đầu ra thô của lượt chạy này. Chúng không phải ngưỡng cải thiện; lượt chạy từ ZIP sẽ tính lại chúng trên cùng các tập dữ liệu.
{
"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"
}
