Skip to content

Commit 55c0f05

Browse files
fix: deterministic source-import missing detection and denial-guard supersession (#154)
* fix: deterministic source-import missing detection and denial-guard supersession Two failing offline-gate tests and two latent defects: - mark_source_import_items_missing() gains an explicit source_keys path: a complete scan now marks exactly the planned-missing rows instead of relying on the last_seen_at < run_started heuristic, which a renamed file defeated (rename stamps last_seen_at=now, so the next run's delete never matched). The documents importer shares ObsidianImporter.import_scan and inherits this. - _clear_superseded_denial() now supersedes a billing denial on persisted-state content change, not wall-clock comparison. Equal-timestamp entitlements made the strict '>' stick forever after a valid reconnect, and a naive '>=' would let a pre-denial record saved in the same coarse clock tick resurrect grants. cloud_session.saved_session_digest() fingerprints the session bytes so write order is observed without exposing credentials. - tests: deflake ULID same-millisecond ordering (repair cursor sweep count) and the archived_at == valid_from boundary (half-open temporal exclusion). Verified: full offline suite green, ruff/pyright clean, eval gates unchanged. * fix: review findings — unknown-baseline guard and full-manifest paging P1 (v2_api): _clear_superseded_denial now requires both the captured baseline digest and the current digest to be known before treating a difference as a superseding rewrite. An unreadable state file at denial time left None as the baseline; a later recovered read then differed from it and cleared the guard even though the file still held the pre-denial active entitlement. Unknown baselines now stick (fail-closed) until the next denial cycle or restart. P2 (source imports): list_source_import_items gains offset paging and ObsidianImporter._all_source_items() pages the full manifest (bounded at 200k rows) for both import planning and link reconciliation. A manifest grown past the 10k single-page cap left historical rows beyond the page invisible, so the explicit source_keys missing-marking path skipped them while the run reported itself complete. Verified: unknown-baseline fail-closed probe, reconnect-supersedes probe, 10,050-row paging probe (complete, duplicate-free), full affected test files (obsidian importer/schema, document importer, hosted plan) exit 0, ruff+pyright clean. * fix: round-2 review — generation-guarded missing marks and bounded paging Three follow-up findings on the source-import missing path: - Concurrent imports: the exact-key update is now per-key and conditioned on the row still matching its planned generation (last_seen_at, last_seen_job_id, live state). A newer run that re-upserts a row after an older run planned it missing keeps its fresh state instead of being clobbered back to missing. - SQLite host-parameter limits: the oversized IN clause is gone entirely — per-key updates in one transaction scale to any manifest (200k keys marked in ~6s locally). - Paging cap: _all_source_items() now reports whether the full manifest was read. Beyond the 200k-row memory bound the run is marked partial, missing finalization is skipped, and link reconciliation refuses to retire edges on the incomplete view. Probes: stale-plan-vs-refreshed-row preserved; 200,050-row manifest flags truncation and marks all 200k planned keys; affected suites (obsidian importer/schema, document importer, hosted plan, consolidate) exit 0; ruff+pyright clean. * fix: round-3 review — parse-bound denial digests and keyset manifest paging - Denial guard race: the supersession digest now travels with the parse. _session_entitlement_snapshot() and _read_entitlement_cache_snapshot() return the entitlement plus a sha256 of the exact bytes it was parsed from, and _clear_superseded_denial() compares that observed digest against the denial baseline instead of re-reading a file that may have changed since. The denial persistence write landing between a license read's parse and its check can no longer pose as a superseding reconnect. cloud_session gains saved_entitlement_snapshot(); _session_entitlement()/_read_entitlement_cache() remain as thin wrappers for their other callers. - Manifest paging: OFFSET is replaced by a (relative_path, id) keyset cursor. OFFSET on a live ORDER BY lets a concurrent rename shift an unread row across the page boundary so it is silently skipped while the pager reports a complete read; the keyset cursor returns every row at or after the cursor exactly once, and a row renamed below the read range degrades into content-hash rename detection. - Cap boundary: _all_source_items() probes one row past the 200k-row memory cap, so a manifest of exactly 200,000 rows reads as complete instead of forcing every such import to partial. Probes: mid-read denial write leaves the guard set while a post-write reconnect parse clears it; exactly-at-cap manifest reports complete; beyond-cap reports partial with a duplicate-free plan set; 200k-row keyset read in ~2s. Affected suites (obsidian importer/schema, document importer, hosted plan, consolidate) exit 0; ruff+pyright clean. * fix: round-4 review — finalize only rows the guarded update actually marked mark_source_import_items_missing() returns the source_keys it actually marked instead of a count: the guarded per-key path collects keys whose generation predicates matched, and the heuristic path selects its rows before updating (chunked IN under host-parameter limits). import_scan records job history and the completed report from that reality — rows the generation guard left live because a concurrent import refreshed them are recorded as skipped, never as missing, so the job receipt can no longer claim a live, newer source was removed. Probe: stale plan vs refreshed row -> only the stale key returned and marked; refreshed row stays imported and out of the missing report; heuristic path returns keys. Affected suites exit 0; ruff+pyright clean. * fix: round-5 review — constant-time finalized check in missing finalization The guard-skipped loop tested membership against the finalized list, scanning up to 200k dicts per missing item — quadratic job finalization that can appear hung on a large complete import. Test the item's source_key against the already-built marked_keys set instead. * test(hosted): pin byte-identical denial-supersession invariant - Add the invariant test the review round identified as missing: replaying the exact pre-denial session bytes after _mark_authoritative_denial must NOT clear the process guard — supersession is content-digest-based, never timestamp-based, so coarse-clock ties cannot resurrect grants. - Clarify the finalized-keys check in obsidian_import: hash-set membership (O(1) average) replaced a quadratic list scan; it is a complexity fix, not a timing-sensitive comparison (commit 4813f6a's "constant-time" wording overclaimed). * fix(import): page import previews like execution Preview plans read the vault manifest through the same keyset-paged reader as execution, so manifests larger than one list page no longer make previews silently drop beyond-boundary rows; they are reported as missing like any other unseen source. Regression test seeds a real 10k-row manifest and fails against the unpaged reader (stash-verified). * fix(import): stabilize manifest pagination under concurrent renames * fix(import): fail closed on truncated previews and denials * fix(import): propagate late manifest truncation * fix(entitlement): release lock during denial probes
1 parent 92e70ed commit 55c0f05

10 files changed

Lines changed: 632 additions & 62 deletions

CHANGELOG.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,11 @@ All notable changes to Engraphis are documented here. Format loosely follows
108108

109109
### Fixed
110110

111+
- Import previews now page the source manifest exactly like execution, so vaults whose manifest
112+
outgrew one list page (10k identities) no longer show manifest-only files as silently absent
113+
from the preview plan; beyond-boundary rows are reported as `missing` instead of dropped.
114+
Manifest pages now use one read snapshot and de-duplicate identities that move across a
115+
cursor while a concurrent import updates their path.
111116
- Importing more than 1,000 files through the dashboard no longer fails with "Internal Server
112117
Error": wizard upload routes parse multipart forms under the advertised 1,500-file ceiling
113118
instead of Starlette's hidden 1,000-part parser default, oversized batches return a clear 413,

engraphis/cloud_session.py

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -531,6 +531,66 @@ def saved_entitlement() -> dict:
531531
return {}
532532

533533

534+
def saved_entitlement_snapshot() -> tuple[dict, Optional[str]]:
535+
"""Read the session once; return its entitlement plus a digest of those exact bytes.
536+
537+
Binding the parse to the bytes it came from lets a caller prove where an answer
538+
predates a denial without re-reading: a license read that parsed the pre-denial
539+
session must never mistake the denial-persistence write landing mid-read for a
540+
superseding reconnect. ``None`` means "could not determine" (unreadable state);
541+
``""`` means the file is absent.
542+
"""
543+
544+
try:
545+
raw = read_private_text(
546+
_session_path(), max_bytes=64 * 1024, allow_missing=True
547+
)
548+
except Exception: # noqa: BLE001 — an unreadable session is simply "nothing known"
549+
return {}, None
550+
if not raw:
551+
return {}, ""
552+
digest = hashlib.sha256(raw.encode("utf-8", "surrogatepass")).hexdigest()
553+
try:
554+
value = json.loads(raw)
555+
except (ValueError, RecursionError):
556+
return {}, digest
557+
if not isinstance(value, dict):
558+
return {}, digest
559+
declared = _declared_entitlement(value)
560+
if not declared:
561+
return {}, digest
562+
try:
563+
checked_at = float(value.get("entitlement_checked_at") or 0.0)
564+
except (TypeError, ValueError, OverflowError):
565+
checked_at = 0.0
566+
declared["entitlement_checked_at"] = checked_at
567+
declared["organization_id"] = str(value.get("organization_id") or "")
568+
return declared, digest
569+
570+
571+
def saved_session_digest() -> Optional[str]:
572+
"""Return a ``sha256`` digest over the raw saved session bytes, or ``""`` if absent.
573+
574+
``None`` means "could not determine" (unreadable or unsafe state file), which callers
575+
must treat as "no evidence of change". This lets a caller detect that the session was
576+
*rewritten* — a genuine reconnect always rotates the refresh credential, so the bytes
577+
differ — without parsing the record or exposing any credential material. Wall-clock
578+
timestamps cannot serve this role: two writes inside one coarse clock tick stamp equal
579+
``entitlement_checked_at`` values, so only content distinguishes a post-denial
580+
reconnect from a pre-denial record.
581+
"""
582+
583+
try:
584+
raw = read_private_text(
585+
_session_path(), max_bytes=64 * 1024, allow_missing=True
586+
)
587+
except Exception: # noqa: BLE001 — unreadable state must not crash a digest probe
588+
return None
589+
if raw is None:
590+
return ""
591+
return hashlib.sha256(raw.encode("utf-8", "surrogatepass")).hexdigest()
592+
593+
534594
def record_billing_denial() -> bool:
535595
"""Mark the saved entitlement inactive after an authoritative billing denial.
536596

engraphis/core/store.py

Lines changed: 65 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -3733,17 +3733,28 @@ def get_source_import_item(self, *, vault_id: str, source_key: str) -> Optional[
37333733
return dict(row) if row is not None else None
37343734

37353735
def list_source_import_items(self, *, vault_id: str, states: Optional[list[str]] = None,
3736-
limit: int = 10_000) -> list[dict]:
3736+
limit: int = 10_000, after_path: str = "",
3737+
after_id: str = "") -> list[dict]:
3738+
"""Page the manifest by ``(relative_path, id)`` cursor, not OFFSET.
3739+
3740+
OFFSET is applied to a live ``ORDER BY`` result: a concurrent rename or insert
3741+
shifts unread rows across the page boundary and they are silently skipped while
3742+
the pager believes it saw everything. A keyset cursor is immune — every row at
3743+
or after the cursor is returned exactly once regardless of concurrent writes.
3744+
"""
37373745
if self._source_vault_row(vault_id) is None:
37383746
return []
37393747
params: list[Any] = [vault_id]
37403748
sql = "SELECT * FROM source_imports WHERE vault_id=?"
3749+
if after_path or after_id:
3750+
sql += " AND (relative_path>? OR (relative_path=? AND id>?))"
3751+
params.extend([str(after_path), str(after_path), str(after_id)])
37413752
if states is not None:
37423753
if not states:
37433754
return []
37443755
sql += " AND state IN (" + ",".join("?" for _ in states) + ")"
37453756
params.extend(str(state) for state in states)
3746-
sql += " ORDER BY relative_path LIMIT ?"
3757+
sql += " ORDER BY relative_path, id LIMIT ?"
37473758
params.append(max(1, min(100_000, int(limit))))
37483759
return [dict(row) for row in self.conn.execute(sql, params).fetchall()]
37493760

@@ -3830,22 +3841,66 @@ def rename_source_import_item(self, *, vault_id: str, source_key: str,
38303841
def mark_source_import_items_missing(
38313842
self, *, vault_id: str, seen_before: float,
38323843
preserve_paths: Iterable[str] = (), commit: bool = True,
3833-
) -> int:
3844+
missing_items: Iterable[Any] = (),
3845+
) -> list[str]:
3846+
"""Mark planned rows missing, or fall back to the timestamp heuristic.
3847+
3848+
With ``missing_items`` (the planner's manifest rows), each key is updated
3849+
only while the row still matches its planned generation — ``last_seen_at``,
3850+
``last_seen_job_id``, and a live state. A concurrent import that refreshed
3851+
or re-upserted the row after this run planned it therefore keeps its newer
3852+
state instead of being clobbered back to ``missing``, and per-key updates
3853+
stay clear of SQLite host-parameter limits no matter how many rows died.
3854+
Returns the source_keys actually marked, so callers can finalize job history
3855+
and reports against reality instead of the plan.
3856+
"""
38343857
if self._source_vault_row(vault_id) is None:
3835-
return 0
3858+
return []
3859+
planned = [
3860+
(str(item["source_key"]), item.get("last_seen_at"),
3861+
item.get("last_seen_job_id"))
3862+
for item in missing_items if item.get("source_key")
3863+
]
38363864
with self._write_operation("source_missing", commit=commit):
38373865
for relative_path in {str(path) for path in preserve_paths if str(path)}:
38383866
self.conn.execute(
38393867
"UPDATE source_imports SET last_seen_at=? WHERE vault_id=? "
38403868
"AND relative_path=? AND state NOT IN ('missing','conflict')",
38413869
(float(seen_before), vault_id, relative_path),
38423870
)
3843-
return int(self.conn.execute(
3844-
"UPDATE source_imports SET state='missing', missing_at=? WHERE vault_id=? "
3845-
"AND (last_seen_at IS NULL OR last_seen_at<?) "
3846-
"AND state NOT IN ('missing','conflict')",
3847-
(now_ts(), vault_id, float(seen_before)),
3848-
).rowcount)
3871+
if planned:
3872+
marked: list[str] = []
3873+
stamp = now_ts()
3874+
for key, seen_at, seen_job in planned:
3875+
if self.conn.execute(
3876+
"UPDATE source_imports SET state='missing', missing_at=? "
3877+
"WHERE vault_id=? AND source_key=? "
3878+
"AND (last_seen_at IS NULL OR last_seen_at=?) "
3879+
"AND (last_seen_job_id IS NULL OR last_seen_job_id=?) "
3880+
"AND state NOT IN ('missing','conflict')",
3881+
(stamp, vault_id, key, seen_at, seen_job),
3882+
).rowcount:
3883+
marked.append(key)
3884+
return marked
3885+
marked_rows = [
3886+
str(row["source_key"]) for row in self.conn.execute(
3887+
"SELECT source_key FROM source_imports WHERE vault_id=? "
3888+
"AND (last_seen_at IS NULL OR last_seen_at<?) "
3889+
"AND state NOT IN ('missing','conflict')",
3890+
(vault_id, float(seen_before)),
3891+
).fetchall()
3892+
]
3893+
if not marked_rows:
3894+
return []
3895+
for start in range(0, len(marked_rows), 900):
3896+
chunk = marked_rows[start:start + 900]
3897+
placeholders = ",".join("?" for _ in chunk)
3898+
self.conn.execute(
3899+
f"UPDATE source_imports SET state='missing', missing_at=? "
3900+
f"WHERE vault_id=? AND source_key IN ({placeholders})",
3901+
(now_ts(), vault_id, *chunk),
3902+
)
3903+
return marked_rows
38493904

38503905
def get_source_import(self, import_id: str) -> Optional[dict]:
38513906
row = self.conn.execute("SELECT * FROM source_imports WHERE id=?", (import_id,)).fetchone()

0 commit comments

Comments
 (0)