diff --git a/README.md b/README.md index 5f28a1c..5434791 100644 --- a/README.md +++ b/README.md @@ -265,6 +265,26 @@ repository's local runtime. No external database server or generator is needed to run the retained generated example. Use only the example's test database; reset/repeated-run behavior is documented in its README. +## Customize internal ID allocation + +By default, the SQL provider uses its own durable per-entity ID space. To use +an application allocator, install an asynchronous `InternalIdAllocator` on the +Context before saving entities: + +```python +context.with_internal_id_allocator(my_allocator) +``` + +The allocator implements `async next_id(entity: str) -> int` and +`async ensure_id_floor(entity: str, floor: int) -> None`. The latter reserves +explicitly supplied IDs so future automatic IDs cannot collide. Direct INSERT, +nested batch and graph Save use this one Context capability. An installed +allocator failure stops the write; it never silently falls back to the SQL +provider. Implementations shared across processes must provide durable, +atomic allocation and floor advancement. See the runnable +[Internal ID example](examples/internal-id/README.md). The old synchronous +`generate_id`/`next_id` context calls are unsupported; use the async capability. + ## Optional standard OpenTelemetry module The local standard-module changes are not yet released. Install the optional diff --git a/examples/internal-id/README.md b/examples/internal-id/README.md new file mode 100644 index 0000000..c3fdb25 --- /dev/null +++ b/examples/internal-id/README.md @@ -0,0 +1,15 @@ +# Context-owned Internal ID example + +This SQLite example installs a typed, asynchronous `InternalIdAllocator` on +`UserContext`. Direct, nested-batch, and graph saves use the same allocator; +explicit IDs advance its floor. An allocator failure rejects the write rather +than silently falling back to the provider's ID space. + +```bash +PYTHONPATH=src python examples/internal-id/main.py +``` + +The in-memory allocator here is for one-process demonstration only. Production +allocators shared by multiple processes must reserve IDs and explicit-ID floors +durably and atomically. Gaps after failed operations are acceptable; reuse is +not. diff --git a/examples/internal-id/main.py b/examples/internal-id/main.py new file mode 100644 index 0000000..a733335 --- /dev/null +++ b/examples/internal-id/main.py @@ -0,0 +1,83 @@ +"""Run a small end-to-end Internal ID capability example.""" + +import asyncio +from pathlib import Path +from tempfile import TemporaryDirectory + +import aiosqlite + +from teaql.core.entity import EntityKey +from teaql.core.meta import EntityDescriptor, PropertyDescriptor +from teaql.core.mutation import InsertCommand, MutationRequest +from teaql.core.value import DataType +from teaql.provider.sqlite import SimpleSchemaProvider, create_sqlite_service +from teaql.runtime import RuntimeModule + + +class ExampleAllocator: + def __init__(self): + self._level = 99 + self._lock = asyncio.Lock() + + async def next_id(self, entity: str) -> int: + assert entity == "School" + async with self._lock: + self._level += 1 + return self._level + + async def ensure_id_floor(self, entity: str, floor: int) -> None: + assert entity == "School" + async with self._lock: + self._level = max(self._level, floor) + + +def school_insert(name: str, identity: int | None = None) -> MutationRequest: + command = InsertCommand.new("School").value("name", name) + if identity is not None: + command.value("id", identity) + return MutationRequest(command, comment="create school with custom internal ID") + + +async def run() -> None: + with TemporaryDirectory() as directory: + path = Path(directory) / "internal-id.db" + school = (EntityDescriptor("School").table_name("school_data") + .property(PropertyDescriptor("id", DataType.I64).is_id()) + .property(PropertyDescriptor("version", DataType.I64).is_version()) + .property(PropertyDescriptor("name", DataType.Text))) + provider = SimpleSchemaProvider() + provider.register_entity(school) + service = create_sqlite_service(str(path), provider) + context = (RuntimeModule.new().entity(school).into_context() + .with_schema_provider(service) + .insert_resource("dataService", service) + .with_internal_id_allocator(ExampleAllocator())) + context.disable_sql_log() + await context.ensure_schema() + + await service.mutate(context, school_insert("Direct")) + await service.mutate(context, MutationRequest.Batch([ + school_insert("Batch"), + MutationRequest.Batch([school_insert("Nested")], comment="nested schools"), + ], comment="batch schools")) + await service.mutate(context, school_insert("Explicit", 500)) + + async def save_graph(graph): + scope = graph.scope(EntityKey("School", 501)) + return await graph.transaction.mutate( + graph.context, + graph.request(InsertCommand.new("School").value("name", "Graph"), scope), + ) + + await context.execute_graph_save(save_graph, comment="save school graph") + async with aiosqlite.connect(path) as db: + async with db.execute("SELECT id FROM school_data ORDER BY id") as rows: + assert [row[0] for row in await rows.fetchall()] == [100, 101, 102, 500, 501] + async with db.execute("SELECT COUNT(*) FROM teaql_id_space") as rows: + assert (await rows.fetchone())[0] == 0 + await service.close() + print("PASS Python context-owned Internal ID example") + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/scripts/verify-examples.sh b/scripts/verify-examples.sh index 7762c16..d47868c 100755 --- a/scripts/verify-examples.sh +++ b/scripts/verify-examples.sh @@ -2,7 +2,7 @@ set -euo pipefail repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -expected=(business-clock business-id conformance facet-trace mutation-policy object-storage opaque-entity-reference order-management query-policy school-management task_board trace-chain) +expected=(business-clock business-id conformance facet-trace internal-id mutation-policy object-storage opaque-entity-reference order-management query-policy school-management task_board trace-chain) mapfile -t actual < <(find "$repo/examples" -mindepth 1 -maxdepth 1 -type d -printf '%f\n' | sort) if [[ "${actual[*]}" != "${expected[*]}" ]]; then echo "example inventory changed; update scripts/verify-examples.sh: ${actual[*]}" >&2 @@ -29,6 +29,7 @@ PYTHONPATH="$repo/src" python -m unittest discover -s "$repo/examples/school-man PYTHONPATH="$repo/src" python "$repo/examples/mutation-policy/main.py" PYTHONPATH="$repo/src" python "$repo/examples/business-clock/main.py" PYTHONPATH="$repo/src" python "$repo/examples/business-id/main.py" +PYTHONPATH="$repo/src" python "$repo/examples/internal-id/main.py" PYTHONPATH="$repo/src" python "$repo/examples/query-policy/main.py" PYTHONPATH="$repo/src" python "$repo/examples/opaque-entity-reference/main.py" order_management_tmp="$(mktemp -d)" diff --git a/src/teaql/runtime/__init__.py b/src/teaql/runtime/__init__.py index c6d127e..bb731bb 100644 --- a/src/teaql/runtime/__init__.py +++ b/src/teaql/runtime/__init__.py @@ -12,6 +12,7 @@ from .store import DataStore from .audit import RawAuditEvent, SafeAuditEvent, MutationAuditKind from .business_clock import BusinessClock, FixedBusinessClock, SystemBusinessClock +from .internal_id import InternalIdAllocator from .entity_reference import ( AeadEntityReferenceCodec, EntityReferenceClaims, @@ -57,7 +58,7 @@ from .wire_fields import NormalizedWireInput, WireEntityMetadata, WireFieldMetadata, WireInputError, create_wire_entity_metadata, encode_wire_output, normalize_wire_input, retain_submitted_paths from teaql.core.entity import EntityKey, EntityChangeSet, EntityRoot -__all__ = ["WireFieldMetadata", "WireEntityMetadata", "NormalizedWireInput", "WireInputError", "create_wire_entity_metadata", "normalize_wire_input", "encode_wire_output", "retain_submitted_paths", "EntityKey", "EntityChangeSet", "EntityRoot", "ContextEntityRef", "ContextRootError", "CheckException", "CheckResult", "I18nCatalog", "JsonFieldNamingProfile", "Locale", "ObjectLocation", "UnsupportedLocaleError", "UserContext", "TeaqlRuntime", "SqlLogEntry", "SqlLogOperation", "DiagnosticSqlLogSink", "TextDiagnosticSqlLogSink", "ServiceRuntimeFromEnv", "RuntimeModule", "DataStore", "RawAuditEvent", "SafeAuditEvent", "MutationAuditKind", "BusinessClock", "FixedBusinessClock", "SystemBusinessClock", "AeadEntityReferenceCodec", "EntityReferenceClaims", "EntityReferenceCodec", "EntityReferenceTokenError", "ENTITY_REFERENCE_AAD", "UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT", "UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT", "BUSINESS_ID_PERMUTATION_V1_ALPHABET", "BUSINESS_ID_PERMUTATION_V1_DOMAIN_SIZE", "BUSINESS_ID_PERMUTATION_V1_MAX_SEQUENCE", "BUSINESS_ID_PERMUTATION_V1_WIDTH", "DefaultBusinessIdProfileFactory", "DefaultBusinessIdService", "InMemoryBusinessIdAllocator", "PermutedDailyBusinessIdProfile", "StaticBusinessIdKeyProvider", "encode_business_id_permutation_v1", "ContextTools", "ExecutableHttpTool", "HTTP_TOOL", "HttpIntentPhase", "HttpTool", "HttpToolProvider", "ToolDeniedError", "ToolError", "ToolPolicy", "ToolRisk", "Tools", "ToolToken", "ToolUnavailableError"] +__all__ = ["WireFieldMetadata", "WireEntityMetadata", "NormalizedWireInput", "WireInputError", "create_wire_entity_metadata", "normalize_wire_input", "encode_wire_output", "retain_submitted_paths", "EntityKey", "EntityChangeSet", "EntityRoot", "ContextEntityRef", "ContextRootError", "CheckException", "CheckResult", "I18nCatalog", "JsonFieldNamingProfile", "Locale", "ObjectLocation", "UnsupportedLocaleError", "UserContext", "TeaqlRuntime", "SqlLogEntry", "SqlLogOperation", "DiagnosticSqlLogSink", "TextDiagnosticSqlLogSink", "ServiceRuntimeFromEnv", "RuntimeModule", "DataStore", "RawAuditEvent", "SafeAuditEvent", "MutationAuditKind", "BusinessClock", "FixedBusinessClock", "SystemBusinessClock", "InternalIdAllocator", "AeadEntityReferenceCodec", "EntityReferenceClaims", "EntityReferenceCodec", "EntityReferenceTokenError", "ENTITY_REFERENCE_AAD", "UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT", "UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT", "BUSINESS_ID_PERMUTATION_V1_ALPHABET", "BUSINESS_ID_PERMUTATION_V1_DOMAIN_SIZE", "BUSINESS_ID_PERMUTATION_V1_MAX_SEQUENCE", "BUSINESS_ID_PERMUTATION_V1_WIDTH", "DefaultBusinessIdProfileFactory", "DefaultBusinessIdService", "InMemoryBusinessIdAllocator", "PermutedDailyBusinessIdProfile", "StaticBusinessIdKeyProvider", "encode_business_id_permutation_v1", "ContextTools", "ExecutableHttpTool", "HTTP_TOOL", "HttpIntentPhase", "HttpTool", "HttpToolProvider", "ToolDeniedError", "ToolError", "ToolPolicy", "ToolRisk", "Tools", "ToolToken", "ToolUnavailableError"] __all__ += [ "GraphMutationSession", "GraphCommittedError", diff --git a/src/teaql/runtime/context.py b/src/teaql/runtime/context.py index d636270..73a43ec 100644 --- a/src/teaql/runtime/context.py +++ b/src/teaql/runtime/context.py @@ -8,6 +8,7 @@ from datetime import date, datetime, timedelta, timezone from .business_clock import BusinessClock, SystemBusinessClock +from .internal_id import InternalIdAllocator, validate_internal_id_allocator from .entity_reference import ( EntityReferenceClaims, EntityReferenceCodec, @@ -123,6 +124,7 @@ def __init__(self): self._checked_mutations = set() self._business_clock: BusinessClock = SystemBusinessClock() self._entity_reference_codec: Optional[EntityReferenceCodec] = None + self._internal_id_allocator: Optional[InternalIdAllocator] = None from .mutation_policy import MutationPolicyRuntimeState self._mutation_policy = MutationPolicyRuntimeState() @@ -548,12 +550,20 @@ def configure_audit_policy(self, entity: str, mask_fields: List[str], descriptor.audit_value_max_len(value_max_len) return self - def with_internal_id_generator(self, gen: Any) -> 'UserContext': - self.insert_resource("internal_id_generator", gen) + def with_internal_id_allocator(self, allocator: InternalIdAllocator) -> 'UserContext': + """Install ID allocation independently of the query/mutation provider.""" + self._internal_id_allocator = validate_internal_id_allocator(allocator) return self - def set_internal_id_generator(self, gen: Any): - self.insert_resource("internal_id_generator", gen) + def internal_id_allocator(self) -> Optional[InternalIdAllocator]: + return self._internal_id_allocator + + def with_internal_id_generator(self, gen: InternalIdAllocator) -> 'UserContext': + """Compatibility spelling; synchronous generate_id providers are unsupported.""" + return self.with_internal_id_allocator(gen) + + def set_internal_id_generator(self, gen: InternalIdAllocator): + self.with_internal_id_allocator(gen) def with_business_id_profile_factory(self, factory: Any) -> 'UserContext': if factory is None or not callable(getattr(factory, "create", None)): @@ -621,17 +631,10 @@ def i18n_catalog(self): return self.get_resource("i18n_catalog") or I18nCatalog.builtin() def generate_id(self, entity: str) -> Optional[int]: - gen = self.get_resource("internal_id_generator") - if gen and hasattr(gen, "generate_id"): - return gen.generate_id(entity) - return None + raise RuntimeError("synchronous generate_id is unsupported; use an async InternalIdAllocator") def next_id(self, entity: str) -> int: - gen_id = self.generate_id(entity) - if gen_id is not None: - return gen_id - # Simple fallback - return int(datetime.now().timestamp() * 1000) + raise RuntimeError("synchronous next_id is unsupported; use an async InternalIdAllocator") def entity(self, name: str) -> Optional[Any]: if self._metadata and hasattr(self._metadata, "entity"): diff --git a/src/teaql/runtime/internal_id.py b/src/teaql/runtime/internal_id.py new file mode 100644 index 0000000..d58984f --- /dev/null +++ b/src/teaql/runtime/internal_id.py @@ -0,0 +1,26 @@ +"""Trusted, context-owned internal ID allocation capability.""" + +from inspect import iscoroutinefunction +from typing import Protocol, runtime_checkable + + +@runtime_checkable +class InternalIdAllocator(Protocol): + """Allocate unique per-entity IDs and reserve explicit-ID floors. + + Implementations used across processes must persist the floor. A failed + transaction may leave gaps, but an allocated ID must never be reused. + """ + + async def next_id(self, entity: str) -> int: ... + + async def ensure_id_floor(self, entity: str, floor: int) -> None: ... + + +def validate_internal_id_allocator(allocator: InternalIdAllocator) -> InternalIdAllocator: + """Reject incomplete/synchronous capabilities before provider execution.""" + if not isinstance(allocator, InternalIdAllocator): + raise TypeError("internal ID allocator must provide next_id() and ensure_id_floor()") + if not iscoroutinefunction(allocator.next_id) or not iscoroutinefunction(allocator.ensure_id_floor): + raise TypeError("internal ID allocator methods must be async") + return allocator diff --git a/src/teaql/sql/executor.py b/src/teaql/sql/executor.py index b5bbf21..1a449f2 100644 --- a/src/teaql/sql/executor.py +++ b/src/teaql/sql/executor.py @@ -1030,12 +1030,14 @@ async def _mutate(self, context: 'UserContext', request: MutationRequest, privac None, ) if isinstance(req_data, InsertCommand) and id_prop is not None: + allocator = context.internal_id_allocator() if context is not None else None + allocator = allocator if allocator is not None else self if id_prop.name not in req_data.values: req_data.values[id_prop.name] = Value.from_any( - await self.next_id(req_data.entity) + await allocator.next_id(req_data.entity) ) else: - await self.ensure_id_floor( + await allocator.ensure_id_floor( req_data.entity, int(req_data.values[id_prop.name].val)) version_prop = next(( prop for prop in getattr(entity_desc, "properties", []) diff --git a/tests/provider/test_internal_id_allocator.py b/tests/provider/test_internal_id_allocator.py new file mode 100644 index 0000000..ba712aa --- /dev/null +++ b/tests/provider/test_internal_id_allocator.py @@ -0,0 +1,160 @@ +"""Context-owned Internal ID allocation must reach every SQLite write path.""" + +import asyncio + +import aiosqlite +import pytest + +from teaql.core.entity import EntityKey +from teaql.core.meta import EntityDescriptor, PropertyDescriptor +from teaql.core.mutation import InsertCommand, MutationRequest +from teaql.core.value import DataType +from teaql.provider.sqlite import SimpleSchemaProvider, create_sqlite_service +from teaql.runtime import InternalIdAllocator, RuntimeModule +from teaql.sql.executor import SqlDataServiceExecutor + + +class RecordingAllocator: + def __init__(self): + self._lock = asyncio.Lock() + self._levels = {} + self.floors = [] + self.fail_next = False + self.fail_floor = False + + async def next_id(self, entity: str) -> int: + async with self._lock: + if self.fail_next: + raise RuntimeError("custom allocation rejected") + next_value = self._levels.get(entity, 499) + 1 + self._levels[entity] = next_value + return next_value + + async def ensure_id_floor(self, entity: str, floor: int) -> None: + async with self._lock: + if self.fail_floor: + raise RuntimeError("custom floor rejected") + self.floors.append((entity, floor)) + self._levels[entity] = max(self._levels.get(entity, 499), floor) + + +async def fixture(path, allocator=None): + school = (EntityDescriptor("School").table_name("school_data") + .property(PropertyDescriptor("id", DataType.I64).is_id()) + .property(PropertyDescriptor("version", DataType.I64).is_version()) + .property(PropertyDescriptor("name", DataType.Text))) + provider = SimpleSchemaProvider() + provider.register_entity(school) + service = create_sqlite_service(str(path), provider) + context = RuntimeModule.new().entity(school).into_context() + context.with_schema_provider(service).insert_resource("dataService", service) + context.disable_sql_log() + if allocator is not None: + context.with_internal_id_allocator(allocator) + await context.ensure_schema() + return service, context + + +def insert(name, identity=None): + command = InsertCommand.new("School").value("name", name) + if identity is not None: + command.value("id", identity) + return MutationRequest(command, comment="save school with internal ID allocator") + + +async def ids(path): + async with aiosqlite.connect(path) as db: + async with db.execute("SELECT id FROM school_data ORDER BY id") as cursor: + return [row[0] for row in await cursor.fetchall()] + + +async def assert_no_default_id_space_rows(path): + async with aiosqlite.connect(path) as db: + async with db.execute("SELECT COUNT(*) FROM teaql_id_space") as cursor: + assert (await cursor.fetchone())[0] == 0 + + +@pytest.mark.asyncio +async def test_context_allocator_reaches_direct_nested_batch_explicit_and_graph(tmp_path, monkeypatch): + async def forbidden_provider_id(*args, **kwargs): + raise AssertionError("default provider ID space must not be called") + + monkeypatch.setattr(SqlDataServiceExecutor, "next_id", forbidden_provider_id) + monkeypatch.setattr(SqlDataServiceExecutor, "ensure_id_floor", forbidden_provider_id) + path = tmp_path / "school.db" + allocator = RecordingAllocator() + assert isinstance(allocator, InternalIdAllocator) + service, context = await fixture(path, allocator) + await service.mutate(context, insert("direct")) + await service.mutate(context, MutationRequest.Batch([ + insert("batch one"), + MutationRequest.Batch([insert("batch nested")], comment="nested batch"), + ], comment="batch of schools")) + await service.mutate(context, insert("explicit", 900)) + + async def graph_save(graph): + scope = graph.scope(EntityKey("School", 901)) + return await graph.transaction.mutate(graph.context, + graph.request(InsertCommand.new("School").value("name", "graph"), scope)) + + await context.execute_graph_save(graph_save, comment="save school graph") + assert await ids(path) == [500, 501, 502, 900, 901] + assert ("School", 900) in allocator.floors + await assert_no_default_id_space_rows(path) + await service.close() + + +@pytest.mark.asyncio +async def test_allocator_concurrent_allocations_are_unique_per_entity(): + allocator = RecordingAllocator() + ids = await asyncio.gather(*(allocator.next_id("School") for _ in range(24))) + assert sorted(ids) == list(range(500, 524)) + assert await allocator.next_id("Vendor") == 500 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", ["next", "floor"]) +async def test_custom_failure_does_not_fall_back_or_write(tmp_path, failure, monkeypatch): + async def forbidden_provider_id(*args, **kwargs): + raise AssertionError("default provider ID space must not be called") + + monkeypatch.setattr(SqlDataServiceExecutor, "next_id", forbidden_provider_id) + monkeypatch.setattr(SqlDataServiceExecutor, "ensure_id_floor", forbidden_provider_id) + path = tmp_path / "failure.db" + allocator = RecordingAllocator() + service, context = await fixture(path, allocator) + allocator.fail_next = failure == "next" + allocator.fail_floor = failure == "floor" + request = insert("rejected", 900 if failure == "floor" else None) + with pytest.raises(RuntimeError, match="custom (allocation|floor) rejected"): + await service.mutate(context, request) + assert await ids(path) == [] + await assert_no_default_id_space_rows(path) + await service.close() + + +@pytest.mark.asyncio +async def test_without_override_retains_provider_id_space(tmp_path): + path = tmp_path / "default.db" + service, context = await fixture(path) + await service.mutate(context, insert("default")) + assert await ids(path) == [1] + await service.close() + + +def test_malformed_or_old_synchronous_generator_is_rejected_before_use(): + from teaql.runtime import UserContext + class Incomplete: + async def next_id(self, entity): + return 5 + class Synchronous: + def next_id(self, entity): + return 5 + def ensure_id_floor(self, entity, floor): + return None + with pytest.raises(TypeError, match="next_id.*ensure_id_floor"): + UserContext().with_internal_id_allocator(Incomplete()) + with pytest.raises(TypeError, match="must be async"): + UserContext().with_internal_id_allocator(Synchronous()) + with pytest.raises(TypeError, match="next_id.*ensure_id_floor"): + UserContext().with_internal_id_generator(object())