feat(sync): tombstone, pull dall'export, ritocchi flush e fallback su 404
Prepara il client al gateway 2.12.0 (export + soft delete), mantenendo la compatibilità con la 2.11.0 in produzione fino al redeploy. - tombstone: colonna `deleted_at` (+ migrazione), monotona nel merge, esclusa dalle ricerche locali di default; `--deleted` in CLI e /qmem:local find; conteggio nel report/status; marker 🗑 nei risultati di qmem_get - `pull` usa `include_deleted=true` e mappa l'intero payload dell'export (incluso deleted_at), così il mirror impara le cancellazioni - rate limit: pace del flush 600 ms (100 req/min < 120/min del gateway) e messaggio dedicato su 429 (prima 300-400 ms → possibile 429 con code grandi) - `qmem_get`: fallback sull'indice locale anche sul 404 (un id in coda non è ancora sul gateway) con etichetta "non presente sul gateway" - test: 27 controlli (nuovi: pull con tombstone, esclusione/visibilità tombstone, ricerca del record esportato) Verificato con la suite locale completa: 27/27.
This commit is contained in:
Matteo Benedetto
+49
-13
@@ -92,7 +92,9 @@ CREATE TABLE IF NOT EXISTS records(
|
||||
parent_id TEXT, topic TEXT, level TEXT, private INTEGER NOT NULL DEFAULT 0,
|
||||
text_hash TEXT, sources TEXT, synced_at TEXT, updated_at TEXT,
|
||||
/** 1 = record creato offline, non ancora presente sul gateway */
|
||||
pending INTEGER NOT NULL DEFAULT 0
|
||||
pending INTEGER NOT NULL DEFAULT 0,
|
||||
/** Tombstone: istante di cancellazione sul gateway (soft delete) */
|
||||
deleted_at TEXT
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS records_project ON records(project_id);
|
||||
CREATE INDEX IF NOT EXISTS records_kind ON records(kind);
|
||||
@@ -132,6 +134,7 @@ CREATE INDEX IF NOT EXISTS pending_status ON pending(status);
|
||||
.all()
|
||||
.map((c: any) => c.name);
|
||||
if (!cols.includes("pending")) db.exec("ALTER TABLE records ADD COLUMN pending INTEGER NOT NULL DEFAULT 0");
|
||||
if (!cols.includes("deleted_at")) db.exec("ALTER TABLE records ADD COLUMN deleted_at TEXT");
|
||||
return db;
|
||||
}
|
||||
|
||||
@@ -156,6 +159,8 @@ export interface LocalRecordInput {
|
||||
private?: boolean | null;
|
||||
/** true = creato offline (in coda), false = presente sul gateway */
|
||||
pending?: boolean | null;
|
||||
/** Tombstone dal gateway (soft delete): se valorizzato il record è cancellato */
|
||||
deleted_at?: string | null;
|
||||
/** Fonte dell'osservazione: store | correct | get | search | export | enrich | outbox | sync */
|
||||
source: string;
|
||||
}
|
||||
@@ -178,11 +183,11 @@ export function upsertRecords(db: Db, records: LocalRecordInput[]): number {
|
||||
const find = db.prepare("SELECT * FROM records WHERE memory_id = ?");
|
||||
const insert = db.prepare(`INSERT INTO records
|
||||
(memory_id, text, kind, project_id, scope, agent_id, confidence, importance, created_at,
|
||||
supersedes_id, superseded_by, parent_id, topic, level, private, text_hash, sources, synced_at, updated_at, pending)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`);
|
||||
supersedes_id, superseded_by, parent_id, topic, level, private, text_hash, sources, synced_at, updated_at, pending, deleted_at)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`);
|
||||
const update = db.prepare(`UPDATE records SET
|
||||
text=?, kind=?, project_id=?, scope=?, agent_id=?, confidence=?, importance=?, created_at=?,
|
||||
supersedes_id=?, superseded_by=?, parent_id=?, topic=?, level=?, private=?, text_hash=?, sources=?, synced_at=?, updated_at=?, pending=?
|
||||
supersedes_id=?, superseded_by=?, parent_id=?, topic=?, level=?, private=?, text_hash=?, sources=?, synced_at=?, updated_at=?, pending=?, deleted_at=?
|
||||
WHERE rowid_=?`);
|
||||
const now = new Date().toISOString();
|
||||
let written = 0;
|
||||
@@ -217,18 +222,20 @@ export function upsertRecords(db: Db, records: LocalRecordInput[]): number {
|
||||
now,
|
||||
now,
|
||||
r.pending === true ? 1 : r.pending === false ? 0 : (existing?.pending ?? 0),
|
||||
r.deleted_at ?? existing?.deleted_at ?? null,
|
||||
];
|
||||
if (existing) {
|
||||
// I campi già noti non vengono mai azzerati da osservazioni più povere
|
||||
// (es. un risultato di ricerca che non riporta project_id).
|
||||
const keys = ["text", "kind", "project_id", "scope", "agent_id", "confidence", "importance", "created_at",
|
||||
"supersedes_id", "superseded_by", "parent_id", "topic", "level", "private", "text_hash", "sources", "synced_at", "updated_at", "pending"];
|
||||
"supersedes_id", "superseded_by", "parent_id", "topic", "level", "private", "text_hash", "sources", "synced_at", "updated_at", "pending", "deleted_at"];
|
||||
const merged = vals.map((v, i) => {
|
||||
const key = keys[i];
|
||||
const cur = (existing as any)[key];
|
||||
const curEmpty = cur === null || cur === undefined || cur === "" || cur === 0;
|
||||
if (key === "private") return cur === 1 ? 1 : v; // monotono: una volta privato resta privato
|
||||
if (key === "pending") return v; // 0 = sincronizzato: prevale sullo stato locale
|
||||
if (key === "deleted_at") return cur ?? v; // tombstone monotono: una volta cancellato resta cancellato
|
||||
if (key === "sources" || key === "synced_at" || key === "updated_at") return v;
|
||||
if (v === null || v === "" ) return curEmpty ? v : cur; // non azzerare
|
||||
return v;
|
||||
@@ -432,6 +439,8 @@ export interface LocalSearchParams {
|
||||
topic?: string;
|
||||
include_superseded?: boolean;
|
||||
include_private?: boolean;
|
||||
/** Includi i tombstone (record cancellati sul gateway) */
|
||||
include_deleted?: boolean;
|
||||
top_k?: number;
|
||||
/** true = match esatto della frase, senza espansione prefisso */
|
||||
exact?: boolean;
|
||||
@@ -449,6 +458,7 @@ export interface LocalSearchHit {
|
||||
sources: string | null;
|
||||
/** 1 = creato offline, non ancora sul gateway */
|
||||
pending?: number;
|
||||
deleted_at?: string | null;
|
||||
snippet: string;
|
||||
rank: number;
|
||||
/** "and" = tutti i termini presenti; "or" = match parziale (AND senza risultati) */
|
||||
@@ -490,10 +500,11 @@ async function runSearch(db: Db, params: LocalSearchParams, match: string): Prom
|
||||
args.push(params.topic);
|
||||
}
|
||||
if (!params.include_superseded) where.push("r.superseded_by IS NULL");
|
||||
if (!params.include_deleted) where.push("r.deleted_at IS NULL");
|
||||
if (!params.include_private) where.push("r.private = 0");
|
||||
args.push(Math.min(Math.max(params.top_k ?? 5, 1), 50));
|
||||
const sql = `SELECT r.memory_id, r.text, r.kind, r.project_id, r.scope, r.agent_id, r.created_at,
|
||||
r.superseded_by, r.sources, r.pending, bm25(records_fts) AS rank,
|
||||
r.superseded_by, r.sources, r.pending, r.deleted_at, bm25(records_fts) AS rank,
|
||||
snippet(records_fts, 0, '«', '»', '…', 14) AS snippet
|
||||
FROM records_fts JOIN records r ON r.rowid_ = records_fts.rowid
|
||||
WHERE ${where.join(" AND ")} ORDER BY rank LIMIT ?`;
|
||||
@@ -527,7 +538,7 @@ export async function localGet(memoryId: string, opts?: { dbFile?: string }): Pr
|
||||
try {
|
||||
const row = db
|
||||
.prepare(
|
||||
`SELECT memory_id, text, kind, project_id, scope, agent_id, created_at, superseded_by, sources, pending, 0 AS rank,
|
||||
`SELECT memory_id, text, kind, project_id, scope, agent_id, created_at, superseded_by, sources, pending, deleted_at, 0 AS rank,
|
||||
COALESCE(substr(text,1,200),'') AS snippet
|
||||
FROM records WHERE memory_id = ?`,
|
||||
)
|
||||
@@ -662,7 +673,7 @@ export async function pullFromGatewayExport(
|
||||
let pages = 0;
|
||||
let fetched = 0;
|
||||
for (; pages < maxPages; pages++) {
|
||||
const route = `/v1/memories:export?limit=${limit}${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ""}`;
|
||||
const route = `/v1/memories:export?limit=${limit}&include_deleted=true${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ""}`;
|
||||
const { ok, status, data } = await gatewayRequest(
|
||||
cfg,
|
||||
"GET",
|
||||
@@ -676,7 +687,29 @@ export async function pullFromGatewayExport(
|
||||
if (!ok) return { supported: false, pages, fetched, message: `export fallito: HTTP ${status}` };
|
||||
const items = data?.results ?? data?.records ?? [];
|
||||
if (Array.isArray(items) && items.length) {
|
||||
upsertRecords(db, items.map((r: any) => ({ ...r, source: "export" })));
|
||||
upsertRecords(
|
||||
db,
|
||||
items.map((r: any) => ({
|
||||
memory_id: r.memory_id ?? r.id,
|
||||
text: r.text,
|
||||
kind: r.kind,
|
||||
project_id: r.project_id,
|
||||
scope: r.scope,
|
||||
agent_id: r.agent_id,
|
||||
confidence: r.confidence,
|
||||
importance: typeof r.importance === "number" ? r.importance : null,
|
||||
created_at: r.created_at,
|
||||
supersedes_id: r.supersedes_id,
|
||||
superseded_by: r.superseded_by,
|
||||
parent_id: r.parent_id,
|
||||
topic: r.topic,
|
||||
level: r.level,
|
||||
private: r.private,
|
||||
deleted_at: r.deleted_at ?? null,
|
||||
pending: false,
|
||||
source: "export",
|
||||
})),
|
||||
);
|
||||
fetched += items.length;
|
||||
}
|
||||
cursor = data?.next_cursor ?? "";
|
||||
@@ -864,7 +897,7 @@ export async function flushQueue(
|
||||
const items = db
|
||||
.prepare("SELECT local_id, payload FROM pending WHERE status='queued' ORDER BY id LIMIT ?")
|
||||
.all(opts?.limit ?? 100) as Array<{ local_id: string; payload: string }>;
|
||||
const pace = opts?.paceMs ?? 300; // < rate limit del gateway (120/min)
|
||||
const pace = opts?.paceMs ?? 600; // 100 richieste/min (< rate limit 120/min)
|
||||
const perRequestMs = Math.min(cfg.timeoutMs ?? 30_000, 12_000);
|
||||
for (const item of items) {
|
||||
stats.processed++;
|
||||
@@ -962,7 +995,7 @@ export async function flushQueue(
|
||||
}
|
||||
} else {
|
||||
// 0/429/5xx: gateway non utilizzabile, resta in coda
|
||||
const msg = `HTTP ${res.status}`;
|
||||
const msg = res.status === 429 ? "rate limit del gateway (HTTP 429): ripreso al prossimo flush" : `HTTP ${res.status}`;
|
||||
db.prepare("UPDATE pending SET attempts=attempts+1, last_attempt=?, last_error=? WHERE local_id=?").run(nowIso(), msg, item.local_id);
|
||||
stats.errors.push(`${item.local_id.slice(0, 8)}: ${msg}`);
|
||||
stats.stopped = msg;
|
||||
@@ -1035,7 +1068,7 @@ let flushInFlight: Promise<unknown> | null = null;
|
||||
/** Flush in background (single-flight): usato dopo operazioni riuscite e su session_start. */
|
||||
export function maybeBackgroundFlush(cfg: MemoryConfig, dbFile?: string): void {
|
||||
if (flushInFlight) return;
|
||||
flushInFlight = flushQueue(cfg, { dbFile, limit: 20, paceMs: 400 })
|
||||
flushInFlight = flushQueue(cfg, { dbFile, limit: 20, paceMs: 600 })
|
||||
.catch(() => undefined)
|
||||
.finally(() => {
|
||||
flushInFlight = null;
|
||||
@@ -1045,7 +1078,7 @@ export function maybeBackgroundFlush(cfg: MemoryConfig, dbFile?: string): void {
|
||||
export async function flushQueueIfPending(cfg: MemoryConfig, dbFile?: string): Promise<FlushStats | null> {
|
||||
const stats = await queueStats({ dbFile });
|
||||
if (!stats.queued) return null;
|
||||
return flushQueue(cfg, { dbFile, limit: 50, paceMs: 300 });
|
||||
return flushQueue(cfg, { dbFile, limit: 50, paceMs: 600 });
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -1070,6 +1103,7 @@ export interface LocalDbReport {
|
||||
duplicateQueue: number;
|
||||
failedQueue: number;
|
||||
pendingInIndex: number;
|
||||
deleted: number;
|
||||
oldestQueued?: string | null;
|
||||
queueLastError?: string | null;
|
||||
topProjects: Array<{ project_id: string | null; n: number }>;
|
||||
@@ -1094,6 +1128,7 @@ export async function localDbReport(opts?: { dbFile?: string }): Promise<LocalDb
|
||||
duplicateQueue: 0,
|
||||
failedQueue: 0,
|
||||
pendingInIndex: 0,
|
||||
deleted: 0,
|
||||
topProjects: [],
|
||||
duplicates: [],
|
||||
};
|
||||
@@ -1123,6 +1158,7 @@ export async function localDbReport(opts?: { dbFile?: string }): Promise<LocalDb
|
||||
report.duplicateQueue = one("SELECT COUNT(*) n FROM pending WHERE status='duplicate'");
|
||||
report.failedQueue = one("SELECT COUNT(*) n FROM pending WHERE status='failed'");
|
||||
report.pendingInIndex = one("SELECT COUNT(*) n FROM records WHERE pending = 1");
|
||||
report.deleted = one("SELECT COUNT(*) n FROM records WHERE deleted_at IS NOT NULL");
|
||||
report.oldestQueued = db.prepare("SELECT created_at FROM pending WHERE status='queued' ORDER BY id LIMIT 1").get()?.created_at ?? null;
|
||||
report.queueLastError = db.prepare("SELECT last_error FROM pending WHERE last_error IS NOT NULL ORDER BY id DESC LIMIT 1").get()?.last_error ?? null;
|
||||
report.topProjects = db.prepare("SELECT project_id, COUNT(*) n FROM records GROUP BY project_id ORDER BY n DESC LIMIT 8").all() as any;
|
||||
|
||||
Reference in New Issue
Block a user