diff --git a/planfile/api/server.py b/planfile/api/server.py index a241eca..4cc5999 100644 --- a/planfile/api/server.py +++ b/planfile/api/server.py @@ -375,6 +375,8 @@ def _validate_completion_receipt(receipt: dict[str, Any] | None, ticket_id: str) def _ticket_snapshot_signature(pf, sprint: str) -> tuple: + if pf.store.ticket_index_enabled(): + return pf.store._cached_ticket_index_signature() return pf.store.sprint_signature(sprint), pf.store._evidence_revision() @@ -536,6 +538,14 @@ def _bounded_stale_index_response( offset=offset, limit=limit, ) + elif view == "operational": + body, total, count = index.render_operational_payloads( + sprint=sprint, + filters=filters, + offset=offset, + limit=limit, + ) + payload = None else: payload, total = index.list_payloads( sprint=sprint, @@ -543,10 +553,10 @@ def _bounded_stale_index_response( offset=offset, limit=limit, ) - payload = [_ticket_operational_payload(ticket) for ticket in payload] except (json.JSONDecodeError, OSError, sqlite3.DatabaseError): return None - body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + if view != "operational": + body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") return Response( content=body, media_type="application/json", @@ -555,7 +565,7 @@ def _bounded_stale_index_response( "X-Planfile-View": view, "X-Planfile-Index-State": "stale", "X-Total-Count": str(total), - "X-Result-Count": str(len(payload)), + "X-Result-Count": str(count if view == "operational" else len(payload)), }, ) @@ -608,6 +618,7 @@ def _ticket_list_response( offset=offset, limit=limit, repair=False, + signature=signature, ) count = len(payload) elif view == "full": @@ -617,6 +628,7 @@ def _ticket_list_response( offset=offset, limit=limit, repair=False, + signature=signature, ) response_limit = _ticket_list_response_byte_limit() if estimated_bytes > response_limit: @@ -648,6 +660,16 @@ def _ticket_list_response( offset=offset, limit=limit, repair=False, + signature=signature, + ) + elif view == "operational": + body, total, count = pf.store.indexed_ticket_operational_response( + sprint=sprint, + filters=filters, + offset=offset, + limit=limit, + repair=False, + signature=signature, ) else: payload, total = pf.store.indexed_ticket_payloads( @@ -656,9 +678,8 @@ def _ticket_list_response( offset=offset, limit=limit, repair=False, + signature=signature, ) - if view == "operational": - payload = [_ticket_operational_payload(ticket) for ticket in payload] count = len(payload) except TicketIndexContentionError: if ( @@ -709,8 +730,10 @@ def _ticket_list_response( ] if body is None: body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") - # Do not retain a response assembled across a concurrent file change. - if signature == _ticket_snapshot_signature(pf, sprint): + # SQLite reads are transactionally coherent and version-keyed. The + # durable-file fallback still needs a second source check because it + # can span multiple independently replaced YAML files. + if pf.store.ticket_index_enabled() or signature == _ticket_snapshot_signature(pf, sprint): _cache_ticket_list_response( query_key=query_key, versioned_key=key, diff --git a/planfile/core/sqlite_index.py b/planfile/core/sqlite_index.py index f150f60..d14566c 100644 --- a/planfile/core/sqlite_index.py +++ b/planfile/core/sqlite_index.py @@ -406,6 +406,79 @@ def render_payloads( body.extend(b"]") return bytes(body), total, count + def render_operational_payloads( + self, + *, + sprint: str, + filters: dict[str, Any], + offset: int, + limit: int | None, + ) -> tuple[bytes, int, int]: + """Project and render queue payloads inside SQLite's JSON runtime.""" + conditions = [] + parameters: list[Any] = [] + if sprint != "all": + conditions.append("sprint=?") + parameters.append(sprint) + for key in ("status", "priority", "source"): + value = filters.get(key) + if value is None: + continue + conditions.append(f"{key}=?") + parameters.append(str(getattr(value, "value", value))) + where = f" WHERE {' AND '.join(conditions)}" if conditions else "" + page_sql = f"SELECT position, ticket_json FROM tickets{where} ORDER BY position" + page_parameters = list(parameters) + if limit is not None: + page_sql += " LIMIT ? OFFSET ?" + page_parameters.extend((limit, offset)) + elif offset: + page_sql += " LIMIT -1 OFFSET ?" + page_parameters.append(offset) + with self._connect() as connection: + total = int( + connection.execute( + f"SELECT COUNT(*) AS count FROM tickets{where}", + parameters, + ).fetchone()["count"] + ) + cursor = connection.execute( + """ + WITH page AS ( + """ + page_sql + """ + ), stripped AS ( + SELECT position, json_remove( + ticket_json, + '$.history', '$.dsl', '$.file', '$.files', + '$.integration', '$.llm_hints', '$.sync', + '$.source.context', '$.outputs.notes', '$.outputs.artifacts' + ) AS payload + FROM page + ), without_empty_source AS ( + SELECT position, + CASE WHEN json_extract(payload, '$.source') = '{}' + THEN json_remove(payload, '$.source') ELSE payload END AS payload + FROM stripped + ), normalized AS ( + SELECT position, + CASE WHEN json_extract(payload, '$.outputs') = '{}' + THEN json_remove(payload, '$.outputs') ELSE payload END AS payload + FROM without_empty_source + ) + SELECT payload FROM normalized ORDER BY position + """, + page_parameters, + ) + body = bytearray(b"[") + count = 0 + for row in cursor: + if count: + body.extend(b",") + body.extend(row["payload"].encode("utf-8")) + count += 1 + body.extend(b"]") + return bytes(body), total, count + def payload_page_metrics( self, *, diff --git a/planfile/core/store.py b/planfile/core/store.py index dcdd307..e7b4064 100644 --- a/planfile/core/store.py +++ b/planfile/core/store.py @@ -10,6 +10,7 @@ from datetime import UTC, datetime, timedelta from enum import Enum from pathlib import Path +from threading import RLock import yaml from pydantic import BaseModel @@ -67,6 +68,8 @@ def __init__(self, directory: str | Path): self._ticket_index_path = self.base_dir / "index" / "tickets.sqlite3" self._ticket_index_rebuild_lock_path = self.base_dir / "index" / ".rebuild.lock" self._ticket_index_rebuild_deferred_until = 0.0 + self._ticket_index_signature_cache: tuple[float, tuple, tuple] | None = None + self._ticket_index_signature_cache_lock = RLock() self._history_locations_path = self.base_dir / "index" / "history-locations.yaml" def _storage_config(self) -> dict: @@ -494,8 +497,10 @@ def mutation_lock(self): fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) except ImportError: pass + self._invalidate_ticket_index_signature_cache() yield finally: + self._invalidate_ticket_index_signature_cache() try: import fcntl @@ -1181,6 +1186,53 @@ def _sqlite_ticket_index(self): def _ticket_index_signature(self) -> tuple: return self.sprint_signature("all"), self._evidence_revision() + @staticmethod + def _ticket_index_signature_cache_seconds() -> float: + """Bound source-signature reuse during a concurrent API read burst.""" + try: + configured = float( + os.environ.get("PLANFILE_TICKET_INDEX_SIGNATURE_CACHE_SECONDS", "0.25") + ) + except (TypeError, ValueError): + configured = 0.25 + return max(0.0, min(configured, 3.0)) + + def _cached_ticket_index_signature(self) -> tuple: + """Coalesce expensive filesystem scans without hiding local mutations. + + A large store may contain thousands of evidence files. FastAPI executes + synchronous reads in parallel, so scanning that tree independently in + every worker amplifies one dashboard refresh into CPU and I/O + starvation. The mutation lock invalidates this process-local snapshot + before and after every write; the short TTL bounds observation of + changes made by another process. + """ + now = time.monotonic() + probe = self._ticket_index_signature_cache_probe() + with self._ticket_index_signature_cache_lock: + cached = self._ticket_index_signature_cache + if ( + cached is not None + and cached[2] == probe + and now - cached[0] <= self._ticket_index_signature_cache_seconds() + ): + return cached[1] + signature = self._ticket_index_signature() + self._ticket_index_signature_cache = (time.monotonic(), signature, probe) + return signature + + def _ticket_index_signature_cache_probe(self) -> tuple: + """Cheaply detect the active-queue edits that must never wait for TTL.""" + try: + evidence_dir_mtime = self._evidence_dir.stat().st_mtime_ns + except OSError: + evidence_dir_mtime = -1 + return self.sprint_signature("current"), evidence_dir_mtime + + def _invalidate_ticket_index_signature_cache(self) -> None: + with self._ticket_index_signature_cache_lock: + self._ticket_index_signature_cache = None + def _ticket_index_records(self): from planfile.core.fastio import read_yaml_fast @@ -1298,7 +1350,7 @@ def ensure_ticket_index(self, *, force: bool = False) -> dict: return index.status(signature) | {"rebuilt": False} return self._rebuild_ticket_index_unlocked(index, force=force) - def require_current_ticket_index(self) -> dict: + def require_current_ticket_index(self, *, signature: tuple | None = None) -> dict: """Validate the projection without rebuilding it on the caller's thread. Latency-sensitive API reads use this guard. A stale projection is a @@ -1308,7 +1360,8 @@ def require_current_ticket_index(self) -> dict: index = self._sqlite_ticket_index() if not self.ticket_index_enabled(): raise TicketIndexContentionError("ticket_index_disabled") - signature = self._ticket_index_signature() + if signature is None: + signature = self._cached_ticket_index_signature() if not index.is_current(signature): raise TicketIndexContentionError("ticket_index_stale") return index.status(signature) | {"rebuilt": False} @@ -1400,8 +1453,12 @@ def indexed_ticket_summaries( offset: int, limit: int | None, repair: bool = True, + signature: tuple | None = None, ) -> tuple[list[dict], int]: - (self.ensure_ticket_index if repair else self.require_current_ticket_index)() + if repair: + self.ensure_ticket_index() + else: + self.require_current_ticket_index(signature=signature) return self._sqlite_ticket_index().list_summaries( sprint=sprint, filters=filters, @@ -1417,9 +1474,13 @@ def indexed_ticket_payloads( offset: int, limit: int | None, repair: bool = True, + signature: tuple | None = None, ) -> tuple[list[dict], int]: """Read full ticket JSON directly from the disposable SQLite projection.""" - (self.ensure_ticket_index if repair else self.require_current_ticket_index)() + if repair: + self.ensure_ticket_index() + else: + self.require_current_ticket_index(signature=signature) return self._sqlite_ticket_index().list_payloads( sprint=sprint, filters=filters, @@ -1435,9 +1496,13 @@ def indexed_ticket_json_response( offset: int, limit: int | None, repair: bool = True, + signature: tuple | None = None, ) -> tuple[bytes, int, int]: """Render full ticket JSON from SQLite without a Python object graph.""" - (self.ensure_ticket_index if repair else self.require_current_ticket_index)() + if repair: + self.ensure_ticket_index() + else: + self.require_current_ticket_index(signature=signature) return self._sqlite_ticket_index().render_payloads( sprint=sprint, filters=filters, @@ -1453,9 +1518,13 @@ def indexed_ticket_json_metrics( offset: int, limit: int | None, repair: bool = True, + signature: tuple | None = None, ) -> tuple[int, int, int]: """Measure a full-ticket page before materializing its JSON body.""" - (self.ensure_ticket_index if repair else self.require_current_ticket_index)() + if repair: + self.ensure_ticket_index() + else: + self.require_current_ticket_index(signature=signature) return self._sqlite_ticket_index().payload_page_metrics( sprint=sprint, filters=filters, @@ -1463,6 +1532,28 @@ def indexed_ticket_json_metrics( limit=limit, ) + def indexed_ticket_operational_response( + self, + *, + sprint: str, + filters: dict, + offset: int, + limit: int | None, + repair: bool = True, + signature: tuple | None = None, + ) -> tuple[bytes, int, int]: + """Render a bounded operational page without Python JSON object graphs.""" + if repair: + self.ensure_ticket_index() + else: + self.require_current_ticket_index(signature=signature) + return self._sqlite_ticket_index().render_operational_payloads( + sprint=sprint, + filters=filters, + offset=offset, + limit=limit, + ) + def migrate_to_sharded_yaml( self, *, diff --git a/tests/test_sqlite_ticket_index.py b/tests/test_sqlite_ticket_index.py index ae5aff9..ff28889 100644 --- a/tests/test_sqlite_ticket_index.py +++ b/tests/test_sqlite_ticket_index.py @@ -117,6 +117,96 @@ def delayed_records(): assert replacement == [{"id": "PLF-2", "name": "Replacement"}] +def test_concurrent_index_signature_reads_share_one_filesystem_scan( + tmp_path, monkeypatch +): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + pf.create_ticket(name="Signature coalescing") + pf.store.configure_ticket_index(True) + pf.store._invalidate_ticket_index_signature_cache() + original = pf.store._ticket_index_signature + scan_started = threading.Event() + finish_scan = threading.Event() + calls = 0 + + def slow_signature(): + nonlocal calls + calls += 1 + scan_started.set() + assert finish_scan.wait(timeout=5) + return original() + + monkeypatch.setattr(pf.store, "_ticket_index_signature", slow_signature) + with ThreadPoolExecutor(max_workers=16) as pool: + futures = [pool.submit(pf.store._cached_ticket_index_signature) for _ in range(16)] + assert scan_started.wait(timeout=5) + finish_scan.set() + signatures = [future.result(timeout=5) for future in futures] + + assert calls == 1 + assert all(signature == signatures[0] for signature in signatures) + + +def test_local_mutation_invalidates_cached_index_signature(tmp_path): + pf = Planfile(str(tmp_path)) + _disable_archive(pf) + pf.create_ticket(name="Before signature") + pf.store.configure_ticket_index(True) + + before = pf.store._cached_ticket_index_signature() + pf.create_ticket(name="After signature") + after = pf.store._cached_ticket_index_signature() + + assert after != before + assert pf.store.require_current_ticket_index(signature=after)["current"] is True + + +def test_sqlite_renders_operational_projection_without_python_object_graph(tmp_path): + ticket = { + "id": "PLF-1", + "name": "Operational projection", + "status": "open", + "history": [{"large": "journal"}], + "dsl": "large dsl", + "source": {"tool": "test", "context": {"large": "source context"}}, + "outputs": { + "notes": ["large note"], + "artifacts": ["artifact://large"], + "result": {"ready": True}, + }, + } + encoded = json.dumps(ticket, separators=(",", ":")) + index = SQLiteTicketIndex(tmp_path / "tickets.sqlite3") + index.rebuild( + [ + { + "id": ticket["id"], + "sprint": "current", + "status": "open", + "priority": "normal", + "source": "test", + "queue": "default", + "created_at": None, + "updated_at": None, + "position": 0, + "ticket_json": encoded, + "summary_json": encoded, + "blocked_by": [], + } + ], + ("source", 1), + ) + + body, total, count = index.render_operational_payloads( + sprint="all", filters={}, offset=0, limit=100 + ) + + assert total == 1 + assert count == 1 + assert json.loads(body) == [server._ticket_operational_payload(ticket)] + + def test_unrelated_ticket_queries_do_not_share_a_response_build_lock( tmp_path, monkeypatch ):