diff --git a/README.md b/README.md index 49c23a9..396c0cf 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ in the Go binary (no database port at all) — all metadata I/O is mediated by t | **DB-file mutex** | Bash `flock` loop in `docker/leader-election.sh` only `exec`s supervisord on the winner, so two containers sharing one `/meta-core` volume can never run two writers on the database file. It is a mutex, not discovery — nothing reads the lock. | | **Embedded SQLite store** | One schema-free `kv(k, v)` table in `/meta-core/db/meta-core.db` (pure-Go `modernc.org/sqlite`, WAL, `synchronous=NORMAL`). One writer goroutine with group commit; a record write (fields + root index + CID aliases + FTS row + events) is atomic. Replaced Redis in 2026-10 (Redis + the in-memory search copy needed ~6.6 GB on watch.nsl.sh's 169k records; the SQLite file is ~2.5 GB on disk with a bounded page cache). | | **HTTP API** | Typed REST surface (gorilla/mux) for metadata, KV browsing, snapshots, schema inference, mounts, watchers, neighbours, identity, UDL, and file access by CID. | -| **Indexed search** | `POST /api/metadata/search` builds a superset of candidates in SQL — a partial expression index `kv_field` over (field name, value) of every record key, plus an FTS5 trigram table — then verifies each with the unchanged Go matchers. Fresh immediately, no in-memory copy (`internal/storage/search.go`, `internal/api/search_index.go`). | +| **Indexed search** | `POST /api/metadata/search` builds a superset of candidates in SQL — a partial expression index `kv_field_v` over (field name, value) of every record key, plus an FTS5 trigram table; the smallest part drives, the other filters are probed per candidate — then verifies each with the unchanged Go matchers. Fresh immediately, no in-memory copy (`internal/storage/search.go`, `internal/api/search_index.go`). | | **SSE event streams** | `/api/events/files` and `/api/events/meta` mirror the store's `file:events` / `meta:events` streams so external services never touch the store. `meta:events` entries are written in the same transaction as the change. | | **WebDAV server** | `/webdav/*` exposes the `/files` volume (read+write) for cross-service file access. nginx in front of meta-core handles caching. | | **Service discovery** | beacon v2: UDP multicast advertise/listen on `239.255.99.1:9099`, neighbours served at `/neighbors`. The file-based registry under `/meta-core/services/*.json` it replaced is gone. | @@ -425,7 +425,7 @@ the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md). | `internal/leader` | `LeaderInfoProvider` — builds the `/urls` payload from hostname/IP/config; no election logic. | | `internal/meshdisco` | beacon v2 node (advertise, probe, neighbour map) — the Go port of the spec. | | `internal/kv` | Storage engine: SQLite `kv` table, single writer + group commit, reader pool, in-transaction change events (`kv.go`), event streams as rows (`stream.go`). | -| `internal/storage` | Flat-key records (`client.go`), UUIDv7 minting, CID reverse index (`cid_resolution.go`), schema sentinel, search over `kv_field` + FTS (`search.go`), UDL store (`udl.go`, `udl_admin.go`), stats. | +| `internal/storage` | Flat-key records (`client.go`), UUIDv7 minting, CID reverse index (`cid_resolution.go`), schema sentinel, search over `kv_field_v` + FTS (`search.go`), derived counters (`onChange`), UDL store (`udl.go`, `udl_admin.go`), stats. | | `internal/cid` | CID parsing + rank ladder for canonical-CID selection (`cid-rank-vectors.json`, guarded by the meta-root `scripts/check-cid-vectors.sh`). | | `internal/identity` | Multi-account secp256k1 keystore, challenge/proof-of-possession, signature verification. | | `internal/api` | HTTP server + all handlers + SSE event endpoints + search matchers + `/api/stats`. | diff --git a/docs/metadata-storage-structure.md b/docs/metadata-storage-structure.md index 7ddaf31..96a1e21 100644 --- a/docs/metadata-storage-structure.md +++ b/docs/metadata-storage-structure.md @@ -169,10 +169,12 @@ Two derived structures are maintained on top of `kv`, in the same transaction as every record write. Both are rebuildable from the kv rows and never authoritative (`internal/storage/search.go`): -- **`kv_field`** — a partial expression index - `ON kv(lower(substr(k,instr(k,'/')+1)), lower(v)) WHERE k>='file:' AND k<'file;' AND instr(k,'/')>0`. +- **`kv_field_v`** — a partial expression index + `ON kv(lower(substr(k,instr(k,'/')+1)), lower(v), v) WHERE k>='file:' AND k<'file;' AND instr(k,'/')>0`. It answers "which records have field F with a value like V" for **any** field, including ones invented tomorrow — the index is over the key itself. + It carries `v` so the substring test is answered from the index alone + (covering; 13 ms vs 1.48 s without it, for tmdbid on watch). - **`fts`** — an FTS5 trigram table `fts(rid UNINDEXED, hay)` holding one row per record: its search haystack plus its rid (`idx:fts/` points at the row). Trigrams make every 3+ character substring indexable. @@ -180,6 +182,10 @@ never authoritative (`internal/storage/search.go`): `POST /api/metadata/search` uses SQL to build a superset of candidates, then verifies each with the unchanged Go matchers (`recordMatchesFilter`, `matchAllTokens`), so semantics are identical to the old in-memory index. +The smallest part (by a capped count) is collected once and sorted; every +other filter is probed per page as an EXISTS over the candidate's own key +range. The plan, its measurements and the rejected shapes are in +[`docs/project-architecture/metacore-storage-engine.md`](../../../docs/project-architecture/metacore-storage-engine.md). Results still come in hashID order and are fresh immediately (no refresh lag). This replaced the API's in-memory `searchIndex` (60 s refresh, ~30 KB RAM per record). diff --git a/tools/search-parity/README.md b/tools/search-parity/README.md new file mode 100644 index 0000000..700023d --- /dev/null +++ b/tools/search-parity/README.md @@ -0,0 +1,29 @@ +# search-parity + +Regression check for `POST /api/metadata/search` against a running meta-core. + +- `parity.py ` builds ~230 queries from real records (free text, + short tokens, key-set and flat filters, exact/contains, multi-filter). It + computes each query's expected result by brute force: one streaming pass + over every record, using a port of the Go matchers. Then it diffs that + against the live API. Any `lost` result means the SQL candidate superset + dropped a real match. +- `timing.py [meta-core.db]` times the same query set at limit 100 and 2000. + +Run them inside the container. It has python3 and reads the database +read-only: + +```bash +docker cp tools/search-parity metacore-app:/tmp/sp +docker exec metacore-app python3 -u /tmp/sp/parity.py /meta-core/db/meta-core.db +docker exec metacore-app python3 -u /tmp/sp/timing.py +``` + +The reference pass takes ~10 min at 169k records. Records written while it +runs can show up as `extra` at a limit-2000 boundary; check their UUIDv7 time +before treating a diff as real. + +History: these tools gated the 2026-10 Redis → SQLite cut-over. Results: +0/217 diffs on v1.1.0, and 232/234 on v1.2.0, where the 2 diffs were writes +made during the run. See +`docs/project-architecture/metacore-storage-engine.md` in the meta-root. diff --git a/tools/search-parity/parity.py b/tools/search-parity/parity.py new file mode 100644 index 0000000..d57d97c --- /dev/null +++ b/tools/search-parity/parity.py @@ -0,0 +1,178 @@ +"""Search parity check for the meta-core Redis -> SQLite cut-over. + + parity.py + +Builds a query set from real records, computes the EXPECTED result of each +query by brute force — one streaming pass over every record in the database, +evaluated with a line-by-line port of the Go matchers (api.recordMatchesFilter, +api.matchAllTokens, storage.Haystack) — and diffs it against what the live +POST /api/metadata/search returns. A miss means the SQL candidate superset +dropped a real match; that is the failure this exists to catch. +""" +import json, random, sqlite3, sys, time, urllib.request + +BASE = "http://127.0.0.1:9000" +LIMIT = 2000 +DB = sys.argv[1] + + +def post(body): + req = urllib.request.Request(BASE + "/api/metadata/search", data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}) + t0 = time.time() + with urllib.request.urlopen(req, timeout=300) as resp: + data = json.load(resp) + return [r["hashId"] for r in data["results"]], time.time() - t0 + + +SEARCH_FIELDS = ["title", "fileName", "originaltitle", "showtitle", "filePath"] + + +def haystack(f): + parts = [f[k].lower() for k in SEARCH_FIELDS if f.get(k)] + names = [] + for k, v in f.items(): + if k.startswith("titles/") and v not in ("", "false"): + rest = k[len("titles/"):] + if "/" in rest: + name = rest.split("/", 1)[1] + if name: + names.append(name.replace("∕", "/")) + parts += [n.lower() for n in sorted(names)] + return "".join(p + "\n" for p in parts) + + +def truthy(v): + v = (v or "").strip().lower() + return v not in ("", "false", "0") + + +def match_filter(f, flt): + values = [v.strip() for v in flt.get("values", []) if v.strip()] + prop = flt["property"] + if not prop or not values: + return True + for v in values: + if truthy(f.get(prop + "/" + v)): + return True + wanted = {v.lower() for v in values} + prefix = prop.lower() + "/" + for k, val in f.items(): + if len(k) > len(prefix) and k[:len(prefix)].lower() == prefix: + if k[len(prefix):].lower() in wanted and truthy(val): + return True + if prop not in f: + return False + for piece in f[prop].split(","): + piece = piece.strip() + if not piece: + continue + for v in values: + if flt.get("exact"): + if piece.lower() == v.lower(): + return True + elif v.lower() in piece.lower(): + return True + return False + + +def match(rid, f, q): + query = q.get("query", "") + if query: + tokens = query.lower().split() + hay = haystack(f) + if not all(t in hay for t in tokens) and query.lower() not in rid.lower(): + return False + return all(match_filter(f, flt) for flt in q.get("filters", [])) + + +def records(db): + cur = db.execute("SELECT k, v FROM kv WHERE k >= 'file:' AND k < 'file;' ORDER BY k") + rid, fields = None, {} + for k, v in cur: + r, sep, field = k[5:].partition("/") + if not sep or not field: + continue + if r != rid: + if rid is not None: + yield rid, fields + rid, fields = r, {} + fields[field] = v + if rid is not None: + yield rid, fields + + +def build_queries(db): + random.seed(7) + roots = [r for (r,) in db.execute("SELECT substr(k, 11) FROM kv WHERE k >= 'idx:roots/' AND k < 'idx:roots0'")] + sample = [] + for rid in random.sample(roots, 80): + f = {k[len(rid) + 6:]: v for k, v in db.execute( + "SELECT k, v FROM kv WHERE k >= ? AND k < ?", (f"file:{rid}/", f"file:{rid}0"))} + sample.append(f) + qs = [{}] + for fld in ("fileType", "contentKind", "domain", "workForm", "videoType", "provenance", "quality", "indexer"): + for v in sorted({m[fld] for m in sample if m.get(fld)})[:3]: + qs.append({"filters": [{"property": fld, "values": [v], "exact": True}]}) + qs.append({"filters": [{"property": fld, "values": [v.upper()[:3]]}]}) + for m in sample[:30]: + for k in sorted(m): + if k.startswith(("genres/", "categories/", "source/", "titles/")) and k.count("/") == 1: + p, v = k.split("/", 1) + qs.append({"filters": [{"property": p, "values": [v.lower()]}]}) + break + for idf in ("tmdbid", "imdbid", "season"): + if m.get(idf): + qs.append({"filters": [{"property": idf, "values": [m[idf]], "exact": True}]}) + words = (m.get("title") or "").split() + if words: + w = max(words, key=len) + qs.append({"query": w}) + qs.append({"query": w[:2]}) + qs.append({"query": " ".join(words[:2])}) + if m.get("fileType"): + qs.append({"query": w, "filters": [{"property": "fileType", "values": [m["fileType"]]}]}) + for m in sample[30:45]: + words = (m.get("title") or "").split() + if words and m.get("contentKind"): + qs.append({"query": max(words, key=len), "filters": [{"property": "contentKind", "values": [m["contentKind"]], "exact": True}]}) + qs.append({"query": max(words, key=len), "filters": [{"property": "fileType", "values": ["video"]}, {"property": "contentKind", "values": [m["contentKind"]]}]}) + qs += [{"query": "s01e01"}, {"query": "1080p", "filters": [{"property": "fileType", "values": ["video"], "exact": True}]}, + {"filters": [{"property": "genres", "values": ["action", "comedy"]}, {"property": "fileType", "values": ["video"], "exact": True}]}] + for q in qs: + q["limit"] = LIMIT + return qs + + +def main(): + db = sqlite3.connect(f"file:{DB}?mode=ro", uri=True) + qs = build_queries(db) + print(f"{len(qs)} queries; computing reference over every record...", flush=True) + t0 = time.time() + expected = [[] for _ in qs] + n = 0 + for rid, f in records(db): + n += 1 + for i, q in enumerate(qs): + if match(rid, f, q): + expected[i].append(rid) + print(f"reference: {n} records in {time.time() - t0:.0f}s", flush=True) + + bad = 0 + for q, exp in zip(qs, expected): + exp = sorted(exp)[:LIMIT] + got, secs = post(q) + lost, extra = set(exp) - set(got), set(got) - set(exp) + order_ok = got == sorted(got) + status = "OK " if not lost and not extra and order_ok else "DIFF" + bad += status != "OK " + print(f"{status} want={len(exp):5d} got={len(got):5d} lost={len(lost):4d} extra={len(extra):4d} " + f"{secs*1000:7.1f}ms {json.dumps({k: v for k, v in q.items() if k != 'limit'})[:100]}", flush=True) + if lost: + print(" lost e.g.", sorted(lost)[:4]) + if extra: + print(" extra e.g.", sorted(extra)[:4]) + print(f"\n{bad} of {len(qs)} queries differ") + + +main() diff --git a/tools/search-parity/timing.py b/tools/search-parity/timing.py new file mode 100644 index 0000000..2250e49 --- /dev/null +++ b/tools/search-parity/timing.py @@ -0,0 +1,24 @@ +import importlib.util, json, sqlite3, sys, time +spec = importlib.util.spec_from_file_location("parity", "/tmp/parity.py") +sys.argv = ["parity", "/meta-core/db/meta-core.db"] +src = open("/tmp/parity.py").read().replace("\nmain()\n", "\n") +ns = {} +exec(compile(src, "parity", "exec"), ns) +db = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) +qs = ns["build_queries"](db) +for limit in (100, 2000): + times = [] + worst = [] + for q in qs: + q = dict(q, limit=limit) + try: + _, secs = ns["post"](q) + except Exception as e: + print(" ERROR", e, json.dumps(q)[:150], flush=True) + secs = 99 + times.append(secs * 1000) + worst.append((secs * 1000, json.dumps({k: v for k, v in q.items() if k != "limit"})[:90])) + times.sort() + print(f"limit={limit}: n={len(times)} p50={times[len(times)//2]:.0f}ms p90={times[int(len(times)*.9)]:.0f}ms max={times[-1]:.0f}ms") + for t, q in sorted(worst, reverse=True)[:4]: + print(f" {t:7.0f}ms {q}")