Skip to content

fix(federation): entity events must not blank a peer's metadata #43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
DarrenZal wants to merge 1 commit into
base: regen-prod
Choose a base branch
from
Open

fix(federation): entity events must not blank a peer's metadata #43

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 95 additions & 13 deletions api/domain_event_handlers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,27 +117,109 @@ async def _handle_forget(conn, domain: str, rid: str, payload: Dict[str, Any]):
logger.info(f"domain.forget domain={domain} rid={rid_val}")


# Entity-domain embedding columns. Entity events do NOT carry a vector today
# (see `_build_entity_federation_payload` in personal_ingest_api for why: entity
# events run ~22k/30d at ~241 bytes, so attaching a 3072-float vector would take
# that domain from ~5 MB to ~1.3 GB/month). This set exists so that an inbound
# event which *does* carry one is honoured rather than silently dropped.
_ENTITY_EMBEDDING_COLS = frozenset({"embedding", "embedding_3072"})


async def _apply_entity(conn, rid: str, event_type: str, payload: Dict[str, Any], source_node: str):
"""UPSERT entity into entity_registry."""
"""UPSERT entity into entity_registry with NON-DESTRUCTIVE merge semantics.

A peer's copy of an entity is frequently *thinner* than ours — until
2026-08-03 both emit sites sent hardcoded ``aliases: []`` / ``metadata: {}``,
so an inbound event did not merely fail to teach us anything, it actively
ERASED locally-computed values via ``SET metadata = EXCLUDED.metadata``.
Measured blast radius at the time of the fix: 644 rows on the MacBook node
and 964 on the NUC had ``first_seen_rid IS NOT NULL`` (proof that
``store_new_entity`` ran locally and wrote a populated metadata JSON) yet
held ``metadata = '{}'``. There is no audit table, so those are unrecoverable
and that count is a lower bound.

Merge rules — every optional field is merge-or-keep, never replace-with-empty:
* ``metadata`` jsonb concat; incoming keys win, absent keys survive,
and an empty ``{}`` is a no-op.
* ``aliases`` set union; the array can never shrink.
* ``description`` keep local unless the incoming value is non-empty.
* ``phonetic_code`` fill-if-missing (drives Tier-1.x phonetic matching).
* embedding fill-if-missing; never overwrite a local vector.

Identity fields (``entity_text``/``entity_type``/``normalized_text``) are
still replaced outright — unchanged from the original behaviour, so that a
legitimate rename or retype still converges.
"""
_assert_embedding_format(payload, rid)

fuseki_uri = payload.get("fuseki_uri", rid)
entity_text = payload.get("entity_text", "")
entity_type = payload.get("entity_type", "")
normalized_text = payload.get("normalized_text", entity_text.lower().strip())
aliases = normalize_alias_list(payload.get("aliases", []))

metadata = payload.get("metadata", {})
if isinstance(metadata, str):
try:
metadata = json.loads(metadata)
except (ValueError, TypeError):
metadata = {}
if not isinstance(metadata, dict):
metadata = {}

description = payload.get("description") or None
phonetic_code = payload.get("phonetic_code") or None

# Optional inbound vector (not emitted today; honoured if present).
emb_col = payload.get("embedding_column")
emb_literal = None
if emb_col in _ENTITY_EMBEDDING_COLS:
emb_literal = _format_vector(payload.get("embedding_value"))
if emb_literal is None:
emb_col = None

cols = [
"fuseki_uri", "entity_text", "entity_type", "normalized_text",
"aliases", "metadata", "description", "phonetic_code",
]
casts = ["", "", "", "", "", "::jsonb", "", ""]
args = [
fuseki_uri, entity_text, entity_type, normalized_text,
aliases or [], json.dumps(metadata), description, phonetic_code,
]
if emb_col:
cols.append(emb_col)
casts.append("::vector")
args.append(emb_literal)

placeholders = ", ".join(f"${i + 1}{casts[i]}" for i in range(len(cols)))

set_clauses = [
"entity_text = EXCLUDED.entity_text",
"entity_type = EXCLUDED.entity_type",
"normalized_text = EXCLUDED.normalized_text",
# Set union — aliases can never shrink across a federation round-trip.
"aliases = (SELECT COALESCE(array_agg(DISTINCT a), '{}'::text[]) "
"FROM unnest(COALESCE(entity_registry.aliases, '{}'::text[]) "
"|| COALESCE(EXCLUDED.aliases, '{}'::text[])) AS a)",
# jsonb concat — an empty incoming object is a no-op, so a thin peer
# can no longer blank a rich local row.
"metadata = COALESCE(entity_registry.metadata, '{}'::jsonb) "
"|| COALESCE(EXCLUDED.metadata, '{}'::jsonb)",
"description = COALESCE(NULLIF(EXCLUDED.description, ''), entity_registry.description)",
"phonetic_code = COALESCE(entity_registry.phonetic_code, EXCLUDED.phonetic_code)",
"updated_at = NOW()",
]
if emb_col:
set_clauses.append(
f"{emb_col} = COALESCE(entity_registry.{emb_col}, EXCLUDED.{emb_col})"
)

await conn.execute("""
INSERT INTO entity_registry (fuseki_uri, entity_text, entity_type, normalized_text, aliases, metadata)
VALUES ($1, $2, $3, $4, $5, $6::jsonb)
ON CONFLICT (fuseki_uri) DO UPDATE SET
entity_text = EXCLUDED.entity_text,
entity_type = EXCLUDED.entity_type,
normalized_text = EXCLUDED.normalized_text,
aliases = EXCLUDED.aliases,
metadata = EXCLUDED.metadata,
updated_at = NOW()
""", fuseki_uri, entity_text, entity_type, normalized_text,
aliases or [], json.dumps(metadata) if isinstance(metadata, dict) else metadata)
await conn.execute(
f"INSERT INTO entity_registry ({', '.join(cols)}) VALUES ({placeholders}) "
f"ON CONFLICT (fuseki_uri) DO UPDATE SET {', '.join(set_clauses)}",
*args,
)

# Upsert relationships if included
relationships = payload.get("relationships", [])
Expand Down
132 changes: 114 additions & 18 deletions api/personal_ingest_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1517,6 +1517,90 @@ async def store_new_entity(
}, rid=canonical.uri, source_rid=document_rid)


# =============================================================================
# Entity federation payload
# =============================================================================

async def _build_entity_federation_payload(
conn,
uri: str,
*,
fallback_name: str = "",
fallback_type: str = "",
) -> Dict[str, Any]:
"""Build an entity federation payload from the authoritative registry row.

Both emit sites used to inline a 6-key literal that hardcoded
``"aliases": []`` and ``"metadata": {}``. That was doubly wrong: the peer
learned nothing, and — because the subscriber's UPSERT did
``SET metadata = EXCLUDED.metadata`` — the empty payload ERASED whatever the
peer had computed locally. See ``_apply_entity`` in domain_event_handlers.py
for the merge side of this fix.

Deliberately NOT included: the embedding vector. Entity events run ~22k per
30 days at ~241 bytes each; a 3072-float vector is ~60 kB, which would take
this domain from roughly 5 MB to 1.3 GB per month. (For contrast the fact
domain does ship vectors, but it is only ~620 events per 30 days.) Receiving
nodes fill ``embedding_3072`` via ``scripts/backfill_entity_embeddings.py``
instead, which also composes a richer text (name + context + description)
than the register path's bare name.

Falls back to the caller-supplied name/type if the row has gone (e.g. the
entity was merged away between write and emit), so the emit never crashes
the request path.
"""
row = None
try:
row = await conn.fetchrow(
"""
SELECT fuseki_uri, entity_text, entity_type, normalized_text,
aliases, metadata, description, phonetic_code
FROM entity_registry
WHERE fuseki_uri = $1
""",
uri,
)
except Exception as e: # never let a federation emit break the write path
logger.warning(f"entity federation payload lookup failed for {uri}: {e}")

if row is None:
return {
"fuseki_uri": uri,
"entity_text": fallback_name,
"entity_type": fallback_type,
"normalized_text": (fallback_name or "").lower().strip(),
"aliases": [],
"metadata": {},
}

metadata = row["metadata"]
if isinstance(metadata, str):
# NB: this module aliases the stdlib json as `json_module_global`
# (line 22) and never binds bare `json` — using `json.loads` here
# raises NameError at runtime, which py_compile does not catch.
try:
metadata = json_module_global.loads(metadata)
except (ValueError, TypeError):
metadata = {}
if not isinstance(metadata, dict):
metadata = {}

payload: Dict[str, Any] = {
"fuseki_uri": row["fuseki_uri"],
"entity_text": row["entity_text"] or fallback_name,
"entity_type": row["entity_type"] or fallback_type,
"normalized_text": row["normalized_text"]
or (row["entity_text"] or fallback_name or "").lower().strip(),
"aliases": list(row["aliases"] or []),
"metadata": metadata,
}
if row["description"]:
payload["description"] = row["description"]
if row["phonetic_code"]:
payload["phonetic_code"] = row["phonetic_code"]
return payload


# =============================================================================
# TerminusDB Outbox Helpers
# =============================================================================
Expand Down Expand Up @@ -2692,16 +2776,22 @@ async def ingest_extraction(request: IngestRequest):
)
logger.info(f"Resolved to existing: {canonical.uri}")

# Emit federation event for entity replication
# Emit federation event for entity replication.
# Payload is read back from entity_registry so the peer
# gets the real aliases/metadata/description rather than
# hardcoded empties that would blank its own copy.
from api.federation_events import emit_domain_event
await emit_domain_event("entity", "NEW" if is_new else "UPDATE", canonical.uri, {
"fuseki_uri": canonical.uri,
"entity_text": canonical.name,
"entity_type": entity.type,
"normalized_text": canonical.name.lower().strip(),
"aliases": [],
"metadata": {},
})
await emit_domain_event(
"entity",
"NEW" if is_new else "UPDATE",
canonical.uri,
await _build_entity_federation_payload(
conn,
canonical.uri,
fallback_name=canonical.name,
fallback_type=entity.type,
),
)

# Link entity to document
await conn.execute("""
Expand Down Expand Up @@ -4082,16 +4172,22 @@ async def register_vault_entity(request: RegisterEntityRequest):
koi_rid=final_koi_rid if request.publication_scope == "federated" else None
)

# Emit federation event for entity replication
# Emit federation event for entity replication. See
# _build_entity_federation_payload — the previous hardcoded
# {"aliases": [], "metadata": {}} literal here is what blanked
# peers' locally-computed metadata on every round-trip.
from api.federation_events import emit_domain_event
await emit_domain_event("entity", "NEW" if is_new else "UPDATE", canonical.uri, {
"fuseki_uri": canonical.uri,
"entity_text": request.name,
"entity_type": request.entity_type,
"normalized_text": canonical.name.lower().strip() if canonical.name else request.name.lower().strip(),
"aliases": [],
"metadata": {},
})
await emit_domain_event(
"entity",
"NEW" if is_new else "UPDATE",
canonical.uri,
await _build_entity_federation_payload(
conn,
canonical.uri,
fallback_name=request.name,
fallback_type=request.entity_type,
),
)

return result

Expand Down
Loading