feat(store): stage validate=true registration behind a committed overlay

Add a staging overlay to GtsStore: staged entities are visible to internal
validation and the $ref reference registry but invisible to public reads
(get_committed) until commit. A validate=true registration stages, validates,
then commits on success or discards on failure, so a concurrent reader never
observes an unvalidated entity and no lock is held across validation.

Batch add_schemas now runs two-phase - stage every entry, validate each
against the fully-staged set, then commit the survivors and discard the rest -
making it order-independent while never publishing an entry that fails. Add a
threaded concurrency test.

Signed-off-by: Artifizer <artifizer@gmail.com>

Author: Artifizer

gts/src/gts/ops.py

@@ -444,37 +444,40 @@ def add_entity(
         else:
             assert entity.raw_id is not None
             store_key = entity.raw_id
-        with self.store.transaction():
-            previous = self.store.get(store_key)
-            if (
-                previous
-                and not self.allow_entity_updates
-                and previous.content != entity.content
-            ):
-                return GtsAddEntityResult(
-                    ok=False,
-                    error=f"Entity '{store_key}' is already registered with different content",
-                    is_type_schema=entity.is_schema,
-                    conflict=True,
-                )
-            self.store.register(entity)
+        previous = self.store.get_committed(store_key)
+        if (
+            previous
+            and not self.allow_entity_updates
+            and previous.content != entity.content
+        ):
+            return GtsAddEntityResult(
+                ok=False,
+                error=f"Entity '{store_key}' is already registered with different content",
+                is_type_schema=entity.is_schema,
+                conflict=True,
+            )
 
-            try:
-                if entity.is_schema:
-                    self.store.validate_schema_basic(store_key)
-                    if validate:
-                        self.store.validate_schema(store_key, gts_ref_validation)
-                elif validate:
-                    self.store.validate_instance(store_key, gts_ref_validation)
-            except Exception as e:  # noqa: BLE001 - converted to a result object at API boundary
-                self.store.unregister(store_key)
-                if previous:
-                    self.store.register(previous)
-                return GtsAddEntityResult(
-                    ok=False,
-                    error=f"Validation failed: {e!s}",
-                    is_type_schema=entity.is_schema,
-                )
+        # Stage the entity (invisible to public reads) and validate it before
+        # publishing. A failure discards the staged copy, so a reader never
+        # observes an entity that has not passed validation, and the committed
+        # state (any prior version under this id) is never touched. No lock is
+        # held across validation, so concurrent reads are not blocked.
+        self.store.stage(entity)
+        try:
+            if entity.is_schema:
+                self.store.validate_schema_basic(store_key)
+                if validate:
+                    self.store.validate_schema(store_key, gts_ref_validation)
+            elif validate:
+                self.store.validate_instance(store_key, gts_ref_validation)
+        except Exception as e:  # noqa: BLE001 - converted to a result object at API boundary
+            self.store.discard(store_key)
+            return GtsAddEntityResult(
+                ok=False,
+                error=f"Validation failed: {e!s}",
+                is_type_schema=entity.is_schema,
+            )
+        self.store.commit(store_key)
 
         # Return gts_id if available, otherwise raw_id
         entity_id = entity.gts_id.id if entity.gts_id else (entity.raw_id or "")
@@ -494,39 +497,33 @@ def add_entities(
         ok = all(r.ok for r in results)
         return GtsAddEntitiesResult(ok=ok, results=results)
 
-    def add_schemas(
+    def _validate_staged(
         self,
-        schemas: builtins.list[dict[str, Any]],
-        validate: bool = False,
-        gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID,
-    ) -> GtsAddSchemasResult:
-        """Register a batch of GTS Type Schemas.
-
-        Each entry's GTS Type Identifier is derived from its embedded ``$id``;
-        the aggregate ``ok`` is ``True`` only when every entry registered.
-        ``validate`` / ``gts_ref_validation`` apply to every entry exactly as
-        they do on ``POST /entities``.
-        """
-        results = [
-            self.add_schema(schema, validate=validate, gts_ref_validation=gts_ref_validation)
-            for schema in schemas
-        ]
-        ok = all(r.ok for r in results)
-        return GtsAddSchemasResult(ok=ok, results=results)
-
-    def add_schema(
-        self,
-        schema: dict[str, Any],
-        validate: bool = False,
-        gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID,
-    ) -> GtsAddSchemaResult:
-        """Register a single GTS Type Schema, deriving its type_id from ``$id``.
-
-        The embedded ``$schema`` / ``$id`` presence checks are batch-specific;
-        the actual registration and (when requested) validation reuse the
-        single-entity :meth:`add_entity` path, so each entry honors ``validate``
-        / ``gts_ref_validation`` exactly like a ``POST /entities`` call.
-        """
+        entity: GtsEntity,
+        store_key: str,
+        validate: bool,
+        gts_ref_validation: GtsRefValidationMode,
+    ) -> str | None:
+        """Validate a staged entity against the current (staged + committed)
+        set, returning an error message on failure or ``None`` on success. The
+        caller stages before and commits/discards after."""
+        try:
+            if entity.is_schema:
+                self.store.validate_schema_basic(store_key)
+                if validate:
+                    self.store.validate_schema(store_key, gts_ref_validation)
+            elif validate:
+                self.store.validate_instance(store_key, gts_ref_validation)
+        except Exception as e:  # noqa: BLE001 - converted to a result object at API boundary
+            return f"Validation failed: {e!s}"
+        return None
+
+    def _prepare_type_schema(
+        self, schema: dict[str, Any]
+    ) -> tuple[str, GtsEntity, str] | GtsAddSchemaResult:
+        """Run the batch-specific $schema/$id checks and build the entity. On
+        success returns ``(type_id, entity, store_key)``; on failure returns the
+        per-item error result. Does not stage or register anything."""
         if not isinstance(schema, dict):
             return GtsAddSchemaResult(
                 ok=False,
@@ -558,18 +555,112 @@ def add_schema(
                 type_id=type_id,
                 error=f"Invalid GTS Type Schema $id: {embedded_id}",
             )
+        entity = GtsEntity(content=schema, cfg=self.cfg)
+        if not entity.is_schema or not entity.gts_id:
+            return GtsAddSchemaResult(
+                ok=False, type_id=type_id, error="Unable to detect GTS ID in schema"
+            )
+        store_key = entity.gts_id.id
+        previous = self.store.get_committed(store_key)
+        if (
+            previous
+            and not self.allow_entity_updates
+            and previous.content != entity.content
+        ):
+            return GtsAddSchemaResult(
+                ok=False,
+                type_id=type_id,
+                error=f"Entity '{store_key}' is already registered with different content",
+                conflict=True,
+            )
+        return (type_id, entity, store_key)
 
-        result = self.add_entity(
-            schema, validate=validate, gts_ref_validation=gts_ref_validation
-        )
-        if result.ok:
-            return GtsAddSchemaResult(ok=True, type_id=type_id)
-        return GtsAddSchemaResult(
-            ok=False,
-            type_id=type_id,
-            error=result.error,
-            conflict=result.conflict,
-        )
+    def add_schemas(
+        self,
+        schemas: builtins.list[dict[str, Any]],
+        validate: bool = False,
+        gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID,
+    ) -> GtsAddSchemasResult:
+        """Register a batch of GTS Type Schemas.
+
+        Each entry's GTS Type Identifier is derived from its embedded ``$id``;
+        the aggregate ``ok`` is ``True`` only when every entry registered.
+        ``validate`` / ``gts_ref_validation`` apply to every entry exactly as
+        they do on ``POST /entities``.
+
+        With ``validate`` the batch runs in two phases so the outcome is
+        order-independent and nothing invalid is ever published: every
+        structurally-valid entry is staged first (invisible to public reads),
+        then each is validated against the fully-staged set - so an entry can
+        resolve intra-batch references/ancestors regardless of position - and
+        finally the entries that passed are committed while the rest are
+        discarded.
+        """
+        gts_ref_validation = _normalize_gts_ref_validation(gts_ref_validation)
+        if not validate:
+            results = [
+                self.add_schema(schema, validate=False, gts_ref_validation=gts_ref_validation)
+                for schema in schemas
+            ]
+            return GtsAddSchemasResult(ok=all(r.ok for r in results), results=results)
+
+        results: list[GtsAddSchemaResult | None] = [None] * len(schemas)
+        # Phase 1: stage every structurally-valid entry.
+        staged: list[tuple[int, str, GtsEntity, str]] = []
+        for index, schema in enumerate(schemas):
+            prepared = self._prepare_type_schema(schema)
+            if isinstance(prepared, GtsAddSchemaResult):
+                results[index] = prepared
+                continue
+            type_id, entity, store_key = prepared
+            self.store.stage(entity)
+            staged.append((index, type_id, entity, store_key))
+
+        # Phase 2: validate every staged entry against the fully-staged set.
+        verdicts: list[tuple[int, str, str, str | None]] = []
+        for index, type_id, entity, store_key in staged:
+            error = self._validate_staged(entity, store_key, True, gts_ref_validation)
+            verdicts.append((index, type_id, store_key, error))
+
+        # Phase 3: publish the entries that passed, discard the ones that failed.
+        for index, type_id, store_key, error in verdicts:
+            if error is None:
+                self.store.commit(store_key)
+                results[index] = GtsAddSchemaResult(ok=True, type_id=type_id)
+            else:
+                self.store.discard(store_key)
+                results[index] = GtsAddSchemaResult(
+                    ok=False, type_id=type_id, error=error
+                )
+
+        final = [r for r in results if r is not None]
+        return GtsAddSchemasResult(ok=all(r.ok for r in final), results=final)
+
+    def add_schema(
+        self,
+        schema: dict[str, Any],
+        validate: bool = False,
+        gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID,
+    ) -> GtsAddSchemaResult:
+        """Register a single GTS Type Schema, deriving its type_id from ``$id``.
+
+        Stages the entity (invisible to public reads), validates it, then
+        commits on success or discards on failure, so each entry honors
+        ``validate`` / ``gts_ref_validation`` exactly like a ``POST /entities``
+        call and an invalid schema is never observable.
+        """
+        gts_ref_validation = _normalize_gts_ref_validation(gts_ref_validation)
+        prepared = self._prepare_type_schema(schema)
+        if isinstance(prepared, GtsAddSchemaResult):
+            return prepared
+        type_id, entity, store_key = prepared
+        self.store.stage(entity)
+        error = self._validate_staged(entity, store_key, validate, gts_ref_validation)
+        if error is not None:
+            self.store.discard(store_key)
+            return GtsAddSchemaResult(ok=False, type_id=type_id, error=error)
+        self.store.commit(store_key)
+        return GtsAddSchemaResult(ok=True, type_id=type_id)
 
     def validate_id(self, gts_id: str) -> GtsIdValidationResult:
         # Check if it's a wildcard pattern (contains *)
@@ -855,7 +946,8 @@ def get_entity(self, gts_id: str) -> GtsGetEntityResult:
             GtsGetEntityResult with entity details or error
         """
         try:
-            entity = self.store.get(gts_id)
+            # Public read: never expose a staged (not-yet-committed) entity.
+            entity = self.store.get_committed(gts_id)
             if not entity:
                 return GtsGetEntityResult(
                     ok=False, error=f"Entity '{gts_id}' not found"

gts/src/gts/store.py

@@ -139,6 +139,14 @@ def __init__(self, reader: GtsReader | None = None) -> None:
             reader: GtsReader instance to populate entities from
         """
         self._by_id: dict[str, GtsEntity] = {}
+        # Entities registered but not yet published. A staged entity is visible
+        # to internal validation/resolution (via `get`, the reference registry
+        # and x-gts-ref existence checks) so a batch can resolve intra-batch
+        # references regardless of entry order, but it is invisible to public
+        # reads (`get_committed`, `values`, `keys`, `items`, `query`) until
+        # `commit`. This is how a validate=true registration avoids ever
+        # exposing an entity that has not yet passed validation.
+        self._staged: dict[str, GtsEntity] = {}
         self._reader = reader
         self._lock = threading.RLock()
         self._reference_registry: Registry | None = None
@@ -228,15 +236,31 @@ def register_schema(self, type_id: str, schema: dict[str, Any]) -> None:
 
     def get(self, entity_id: str) -> GtsEntity | None:
         """
-        Get a JsonEntity by its ID.
-        If not found in cache, try to fetch from reader.
-        Returns None if not found.
+        Get a JsonEntity by its ID for INTERNAL use (validation, $ref/existence
+        resolution). The staging overlay is consulted first so an entity being
+        validated as part of a batch can resolve its not-yet-committed siblings
+        regardless of order. Public/API reads MUST use :meth:`get_committed` so
+        uncommitted entities are never exposed.
 
         Lookups are normalized to the canonical bare form here, so callers may
         pass either a bare ``gts.`` id or a ``gts://`` URI without stripping the
         scheme themselves.
         """
         entity_id = strip_scheme(entity_id)
+        with self._lock:
+            staged = self._staged.get(entity_id)
+            if staged is not None:
+                return copy.deepcopy(staged)
+        return self.get_committed(entity_id)
+
+    def get_committed(self, entity_id: str) -> GtsEntity | None:
+        """
+        Committed-only lookup: the staging overlay is never consulted, so a
+        staged-but-not-yet-committed entity is invisible here. This is the read
+        path for public/API consumers. Falls back to the reader for ids not yet
+        cached. Returns None if not found.
+        """
+        entity_id = strip_scheme(entity_id)
         with self._lock:
             entity = self._by_id.get(entity_id)
             if entity is not None:
@@ -253,6 +277,47 @@ def get(self, entity_id: str) -> GtsEntity | None:
 
             return None
 
+    @staticmethod
+    def _entity_key(entity: GtsEntity) -> str:
+        """The registry key for an entity: the raw id for instances that carry
+        one (anonymous/UUID instances), otherwise the canonical GTS id."""
+        if not entity.is_schema and entity.raw_id:
+            return entity.raw_id
+        if entity.gts_id and entity.gts_id.id:
+            return entity.gts_id.id
+        if entity.raw_id:
+            return entity.raw_id
+        raise ValueError("Entity must have a valid gts_id or raw_id")
+
+    def stage(self, entity: GtsEntity) -> str:
+        """Place an entity into the staging overlay WITHOUT publishing it, and
+        return its registry key. A staged entity is visible to internal
+        validation but invisible to public reads until :meth:`commit`. Raises
+        ``EntityConflictError`` when a committed entity with different content
+        already holds the key and updates are not allowed. Callers MUST
+        eventually :meth:`commit` or :meth:`discard` the staged key."""
+        stored = copy.deepcopy(entity)
+        key = self._entity_key(stored)
+        with self._lock:
+            self._staged[key] = stored
+            self._invalidate_reference_registry()
+        return key
+
+    def commit(self, entity_id: str) -> None:
+        """Publish a previously staged entity, making it visible to public reads."""
+        with self._lock:
+            staged = self._staged.pop(entity_id, None)
+            if staged is not None:
+                self._by_id[entity_id] = staged
+                self._invalidate_reference_registry()
+
+    def discard(self, entity_id: str) -> None:
+        """Drop a staged entity that failed validation. The committed state is
+        untouched, so a client never observes the discarded (invalid) entity."""
+        with self._lock:
+            if self._staged.pop(entity_id, None) is not None:
+                self._invalidate_reference_registry()
+
     def get_schema_content(self, type_id: str) -> dict[str, Any]:
         """Get schema content as dict (legacy method for backward compatibility)."""
         entity = self.get(type_id)
@@ -265,7 +330,11 @@ def _create_reference_registry(self) -> Registry:
             if self._reference_registry is not None:
                 return self._reference_registry
             registry = Registry()
-            for entity_id, entity in self._by_id.items():
+            # Committed entities plus the staging overlay (staged overrides a
+            # committed entry of the same id), so a schema being validated as
+            # part of a batch resolves `$ref`s to its not-yet-committed siblings.
+            merged = {**self._by_id, **self._staged}
+            for entity_id, entity in merged.items():
                 if entity.is_schema and isinstance(entity.content, dict):
                     resource = Resource.from_contents(
                         _without_x_gts_ref(entity.content),