feat(engine): add central engine lookup + provisioning machinery

Foundation for the central svc-db-*/svc-dns-* engine roles: the engine lookup and engines descriptor, per-consumer engine credentials in the inventory manager, shared-net overlay rendering, container_depends_on, the sys-svc-engine provisioning dispatcher, and the compose_volumes NFS-placement split. base.yml.j2 gates the embedded redis sidecar on lookup('engine','redis',application_id,'local').

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.ansible-lint-ignore

@@ -26,6 +26,13 @@ roles/web-app-minio/tasks/roles/sys-svc-compose/tasks/utils/network/create.yml l
 roles/web-app-seaweedfs/tasks/roles/sys-svc-compose/tasks/utils/network/create.yml load-failure[filenotfounderror]
 roles/svc-db-mariadb/tasks/roles/sys-svc-compose/tasks/utils/swarm/resolve_container.yml load-failure[filenotfounderror]
 roles/web-app-mediawiki/tasks/roles/sys-svc-compose/tasks/utils/finalize.yml load-failure[filenotfounderror]
+roles/svc-db-elasticsearch/tasks/roles/sys-svc-compose/tasks/utils/network/routine.yml load-failure[filenotfounderror]
+roles/svc-db-memcached/tasks/roles/sys-svc-compose/tasks/utils/network/routine.yml load-failure[filenotfounderror]
+roles/svc-db-qdrant/tasks/roles/sys-svc-compose/tasks/utils/network/routine.yml load-failure[filenotfounderror]
+roles/svc-db-rabbitmq/tasks/roles/sys-svc-compose/tasks/utils/network/routine.yml load-failure[filenotfounderror]
+roles/svc-db-redis/tasks/roles/sys-svc-compose/tasks/utils/network/routine.yml load-failure[filenotfounderror]
+roles/svc-db-typesense/tasks/roles/sys-svc-compose/tasks/utils/network/routine.yml load-failure[filenotfounderror]
+roles/svc-dns-unbound/tasks/roles/sys-svc-compose/tasks/utils/network/routine.yml load-failure[filenotfounderror]
 
 # syntax-check ----------------------------------------------------------
 # `keycloak_kcadm_update` is a role-local module shipped at

group_vars/all/08_networks.yml

@@ -84,9 +84,30 @@ PORT_BANDS: # nocheck: unused-var  # consumed by the port-allocation plugin outs
     postgres:
       start: 5432
       end: 5432
+    redis:
+      start: 6379
+      end: 6379
+    memcached:
+      start: 11211
+      end: 11211
+    rabbitmq:
+      start: 5672
+      end: 5672
+    typesense:
+      start: 8108
+      end: 8108
+    qdrant:
+      start: 6333
+      end: 6333
+    elasticsearch:
+      start: 9200
+      end: 9200
     ldap:
       start: 389
       end: 389
+    unbound:
+      start: 53
+      end: 53
     registry:
       start: 5000
       end: 5001

plugins/filter/compose_volumes.py

@@ -23,6 +23,7 @@ try:
         resolve_database_service_key,
     )
     from utils.roles.applications.services.sso import get_sso_config
+    from utils.roles.meta_lookup import get_role_default_placement
 except ModuleNotFoundError:
     from docker_service_enabled import FilterModule as _DockerServiceEnabledFilter
     from get_entity_name import get_entity_name
@@ -38,6 +39,7 @@ except ModuleNotFoundError:
         resolve_database_service_key,
     )
     from utils.roles.applications.services.sso import get_sso_config
+    from utils.roles.meta_lookup import get_role_default_placement
 
 
 def _to_plain(obj: Any) -> Any:
@@ -235,16 +237,6 @@ def compose_volumes(
     )
     canonical_entries = normalize_volumes_meta(raw_meta_volumes)
 
-    # Legacy nfs opt-in: only present when the meta is dict-shape.
-    legacy_dict_volumes = role_data.get("volumes")
-    nfs_keys_legacy = {
-        k
-        for k, v in (
-            legacy_dict_volumes.items() if isinstance(legacy_dict_volumes, dict) else []
-        )
-        if isinstance(v, dict) and v.get("nfs") is not None
-    }
-
     for semantic_name, entry in canonical_entries.items():
         vtype = entry.get("type", "volume")
 
@@ -278,19 +270,21 @@ def compose_volumes(
     swarm_nfs_enabled = (
         deployment_mode == "swarm" and str(storage_backend).lower() == "nfs"
     )
-
-    nfs_canonical = {
-        semantic_name
-        for semantic_name, entry in canonical_entries.items()
-        if bool(entry.get("nfs"))
-    }
-    nfs_keys = nfs_keys_legacy | nfs_canonical
+    # NFS placement is derived from the role's default_placement, never a per-volume
+    # flag: a pinned (default_placement: manager) role stays on one node so its volumes
+    # are node-local; an unpinned role can reschedule, so its volumes must live on shared
+    # NFS to survive and stay consistent across nodes. State that cannot live on NFS
+    # (DB / queue / sqlite engines) is kept node-local by pinning its role instead.
+    role_pinned = (
+        str(get_role_default_placement(application_id) or "").strip().lower()
+        == "manager"
+    )
 
     for vol_name, vol_spec in list(volumes.items()):
         if not isinstance(vol_spec, dict):
             continue
-        wants_nfs = bool(vol_spec.pop("nfs", False)) or (vol_name in nfs_keys)
-        if wants_nfs and swarm_nfs_enabled:
+        vol_spec.pop("nfs", None)  # legacy per-volume flag: ignored, placement decides
+        if swarm_nfs_enabled and not role_pinned:
             named = vol_spec.get("name", vol_name)
             vol_spec.update(_swarm_nfs_driver_opts(dir_var_lib, str(named)))
 

plugins/lookup/container_depends_on.py

@@ -75,10 +75,12 @@ class LookupModule(LookupBase):
         if db_host:
             entries[db_host] = {"condition": "service_healthy"}
 
+        redis_cfg = applications[application_id].get("services", {}).get("redis", {})
+        redis_local = _DockerServiceEnabledFilter.is_docker_service_enabled(
+            applications, application_id, "redis"
+        ) and not bool(redis_cfg.get("shared", False))
         redis_enabled = (
-            _DockerServiceEnabledFilter.is_docker_service_enabled(
-                applications, application_id, "redis"
-            )
+            redis_local
             or get_sso_config(applications, application_id)["is_proxy_gated"]
         )
         if redis_enabled:

plugins/lookup/engine.py

@@ -0,0 +1,171 @@
+"""Lookup ``engine``: resolve a central/embedded engine connection for a consumer.
+
+    {{ lookup('engine', 'redis', application_id) }}            # full dict
+    {{ lookup('engine', 'redis', application_id, 'url') }}     # one field
+
+Mirrors the ``database`` lookup but for the non-RDBMS engines registered in
+:data:`utils.roles.applications.services.engines.ENGINES` (redis, memcached,
+rabbitmq, elasticsearch, typesense, qdrant, unbound). When the consumer declares
+``services.<engine>.shared: true`` the connection points at the central
+``svc-db-*`` / ``svc-dns-*`` stack with the consumer's own provisioned credentials;
+otherwise it points at the embedded sidecar inside the consumer's own stack.
+
+See ``docs/architecture/central-engines.md``.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ansible.errors import AnsibleError
+from ansible.plugins.lookup import LookupBase
+
+from utils.cache.applications import get_merged_applications
+from utils.roles.applications.config import get
+from utils.roles.applications.services.engines import (
+    ENGINES,
+    engine_service,
+    is_engine_enabled,
+    is_engine_shared,
+)
+from utils.roles.entity_name import get_entity_name
+
+# Fields returned by the lookup (``want`` selects one, default ``all``).
+_FIELDS = (
+    "engine",
+    "enabled",
+    "shared",
+    "local",
+    "host",
+    "port",
+    "username",
+    "password",
+    "instance",
+    "prefix",
+    "url",
+)
+
+
+class LookupModule(LookupBase):
+    def run(
+        self,
+        terms: list[Any],
+        variables: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> list[Any]:
+        terms = terms or []
+        if len(terms) not in (2, 3):
+            raise AnsibleError(
+                "engine: requires engine_type, consumer_id [, want_path]"
+            )
+        engine = str(terms[0]).strip()
+        consumer_id = str(terms[1]).strip()
+        want = str(terms[2]).strip() if len(terms) == 3 else "all"
+        if engine not in ENGINES:
+            raise AnsibleError(
+                f"engine: unknown engine '{engine}'; known: {', '.join(ENGINES)}"
+            )
+        if not consumer_id:
+            raise AnsibleError("engine: consumer_id must not be empty")
+
+        vars_ = variables or self._templar.available_variables
+        applications = get_merged_applications(
+            variables=vars_,
+            roles_dir=kwargs.get("roles_dir"),
+            templar=getattr(self, "_templar", None),
+        )
+
+        descriptor = ENGINES[engine]
+        svc_id = descriptor["svc"]
+        port = int(descriptor["port"])
+        entity = get_entity_name(consumer_id)
+
+        enabled = is_engine_enabled(applications, consumer_id, engine)
+        shared = is_engine_shared(applications, consumer_id, engine)
+        svc_cfg = engine_service(applications, consumer_id, engine)
+
+        # Central name = the central engine's service name; embedded host = the
+        # engine's bare service key inside the consumer stack (e.g. "redis").
+        central_name = str(
+            get(
+                applications,
+                svc_id,
+                f"services.{engine}.name",
+                strict=False,
+                default=engine,
+                skip_missing_app=True,
+            )
+            or engine
+        ).strip()
+
+        username = entity
+        prefix = entity
+        password = str(
+            svc_cfg.get("password")
+            or get(
+                applications,
+                consumer_id,
+                f"credentials.{engine}_password",
+                strict=False,
+                default="",
+            )
+            or ""
+        ).strip()
+        host = central_name if shared else engine
+        instance = central_name if shared else entity
+
+        # Redis ACL users live only in memory and are lost whenever the central
+        # redis container is recreated (a deploy can recreate it after the
+        # consumer was provisioned), so per-consumer ACL users are not durable.
+        # Shared consumers therefore authenticate as the central `default` user
+        # with the engine-wide password (security-equivalent: the per-consumer
+        # ACL was already +@all ~* allkeys).
+        if engine == "redis" and shared:
+            username = "default"
+            password = str(
+                get(
+                    applications,
+                    svc_id,
+                    "credentials.REDIS_PASSWORD",
+                    strict=False,
+                    default="",
+                    skip_missing_app=True,
+                )
+                or ""
+            ).strip()
+
+        if engine == "rabbitmq":
+            url = (
+                f"amqp://{username}:{password}@{host}:{port}/{entity}"
+                if shared
+                else f"amqp://{host}:{port}/"
+            )
+        elif engine in ("redis", "memcached"):
+            scheme = "redis" if engine == "redis" else "memcached"
+            auth = f"{username}:{password}@" if (shared and password) else ""
+            url = f"{scheme}://{auth}{host}:{port}"
+        elif engine in ("elasticsearch", "typesense", "qdrant"):
+            url = f"http://{host}:{port}"
+        else:  # unbound
+            url = host
+
+        resolved = {
+            "engine": engine,
+            "enabled": enabled,
+            "shared": shared,
+            "local": bool(enabled and not shared),
+            "host": host,
+            "port": port,
+            "username": username if shared else "",
+            "password": password if shared else "",
+            "instance": instance,
+            "prefix": prefix,
+            "url": url,
+        }
+        if want == "all":
+            return [resolved]
+        if want not in _FIELDS:
+            raise AnsibleError(
+                f"engine: unknown want '{want}'; known: {', '.join(_FIELDS)}"
+            )
+        return [resolved.get(want, "")]

roles/categories.yml

@@ -112,6 +112,11 @@ roles:
       description: "Database applications (DBMS, LDAP, Redis, etc.)"
       icon: "fas fa-database"
       invokable: true
+    dns:
+      title: "DNS Resolvers"
+      description: "Central recursive/forwarding DNS resolver engines (Unbound, etc.)"
+      icon: "fas fa-globe"
+      invokable: true
     opt:
       title: "Service Optimation"
       description: "Optimation Services to improve your system"

roles/sys-svc-compose/tasks/utils/network/create.yml

@@ -8,6 +8,8 @@
     scope: swarm
     driver_options:
       encrypted: "{{ 'true' if swarm.network.encryption else 'false' }}"
+    ipam_config:
+      - subnet: "{{ lookup('config', network_role_id, 'server.networks.local.subnet') }}"
     labels:
       com.docker.compose.network: "default"
       com.docker.compose.project: "{{ network_role_id | get_entity_name }}"

roles/sys-svc-compose/templates/base.yml.j2

@@ -4,12 +4,12 @@ services:
 {% if lookup('database', application_id, 'enabled') %}
 {% include 'roles/sys-svc-rdbms/templates/services/main.yml.j2' %}
 {% endif %}
-{# Load embedded object store #}
 {% if lookup('objstore', application_id, 'local') %}
 {% include 'roles/' ~ lookup('objstore', application_id, 'id') ~ '/templates/service.yml.j2' %}
 {% endif %}
-{# Load Redis (oauth2-proxy session store for SSO-proxied roles) #}
-{% if lookup('applications') | is_docker_service_enabled(application_id, 'redis') or lookup('sso', application_id, 'is_proxy_gated') %}
+{# Embedded Redis sidecar: only when redis is enabled and NOT shared (central),
+   or for the oauth2-proxy session store of SSO-proxied roles. #}
+{% if lookup('engine', 'redis', application_id, 'local') or lookup('sso', application_id, 'is_proxy_gated') %}
 {% include 'roles/svc-db-redis/templates/service.yml.j2' %}
 {% endif %}
 {# Load SSO-proxy sidecar (oauth2-proxy flavor only) #}

roles/sys-svc-engine/meta/main.yml

@@ -0,0 +1,34 @@
+---
+galaxy_info:
+  author: Kevin Veen-Birkenbach
+  description: Engine-agnostic consumer wiring for the central cache / queue / search / vector engines (redis, memcached, rabbitmq, elasticsearch, typesense,
+    qdrant). Delegates per-consumer provisioning (ACL user / vhost / index / scoped key) to the resolved svc-db-<engine> provider role's init entrypoint when
+    the engine is shared.
+  license: Infinito.Nexus Community License (Non-Commercial)
+  company: |
+    Kevin Veen-Birkenbach
+    https://www.veen.world
+  min_ansible_version: "2.9"
+  platforms:
+    - name: ArchLinux
+      versions:
+        - all
+    - name: Debian
+      versions:
+        - all
+    - name: EL
+      versions:
+        - all
+    - name: Fedora
+      versions:
+        - all
+    - name: Ubuntu
+      versions:
+        - all
+  galaxy_tags:
+    - redis
+    - rabbitmq
+    - elasticsearch
+    - cache
+    - docker
+dependencies: []

roles/sys-svc-engine/meta/services.yml

@@ -0,0 +1,4 @@
+---
+# nocheck: required-by-coverage
+svc-engine:
+  lifecycle: alpha

roles/sys-svc-engine/tasks/main.yml

@@ -0,0 +1,13 @@
+---
+# nocheck: run-once
+- name: "For '{{ application_id }}': Freeze engine consumer id"
+  ansible.builtin.set_fact:
+    _engine_consumer_id: "{{ engine_consumer_id | default(application_id) }}"
+  changed_when: false
+
+- name: "For '{{ _engine_consumer_id }}': Provision shared engines on their central providers"
+  include_tasks: provision.yml
+  loop: "{{ ['rabbitmq', 'elasticsearch', 'typesense'] }}"
+  loop_control:
+    loop_var: _engine_type
+  when: lookup('engine', _engine_type, _engine_consumer_id, 'shared') | bool

roles/sys-svc-engine/tasks/provision.yml

@@ -0,0 +1,25 @@
+---
+- name: "Provision '{{ _engine_consumer_id }}' on engine provider '{{ _engine_provider_role }}'"
+  vars:
+    _engine_provider_role: "svc-db-{{ _engine_type }}"
+  block:
+    - name: "Load engine provider '{{ _engine_provider_role }}' (lazy)"
+      ansible.builtin.include_role:
+        name: "{{ _engine_provider_role }}"
+      vars:
+        application_id: "{{ _engine_provider_role }}"
+      when: >-
+        lookup(
+          'vars',
+          'run_once_' ~ (_engine_provider_role | replace('-', '_')),
+          default='__missing__'
+        ) == '__missing__'
+
+    - name: "Initialize '{{ _engine_consumer_id }}' on provider '{{ _engine_provider_role }}'"
+      ansible.builtin.include_role:
+        name: "{{ _engine_provider_role }}"
+      vars:
+        application_id: "{{ _engine_provider_role }}"
+        database_init: true
+        database_consumer_id: "{{ _engine_consumer_id }}"
+        compose_repository_pull: false

tests/lint/ansible/roles/meta/test_unpinned_no_engine_volume.py

@@ -0,0 +1,102 @@
+"""A role that is NOT placement-pinned must not own a database / queue / index
+engine's on-disk data volume.
+
+NFS placement is derived from the role's ``default_placement`` (see
+``plugins/filter/compose_volumes.py``): an unpinned role's volumes are bound to
+the shared NFS mount so they survive a node reschedule, while a pinned
+(``default_placement: manager``) role keeps its volumes node-local. A database,
+message-queue or search-index engine's on-disk state CANNOT live on NFS (fsync +
+locking corruption). Such state must therefore stay node-local -- which means its
+role must be pinned (so the service never leaves the node holding the data), or
+the engine must be externalised to a central pinned service. This lint catches an
+unpinned role that would otherwise have its engine data silently placed on NFS.
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from utils.cache.yaml import load_yaml_any
+from utils.roles.mapping import ROLE_FILE_META_VOLUMES
+from utils.roles.meta_lookup import get_role_default_placement
+
+from . import PROJECT_ROOT
+
+# Mount-target prefixes that are an engine's on-disk data directory (NFS-hostile).
+_ENGINE_TARGET_PREFIXES = (
+    "/var/lib/postgresql",
+    "/var/lib/mysql",
+    "/var/lib/mariadb",
+    "/var/lib/ldap",
+    "/etc/ldap/slapd.d",
+    "/var/lib/redis",
+    "/var/lib/rabbitmq",
+    "/var/lib/mongodb",
+    "/data/db",
+    "/qdrant",
+    "/prometheus",
+)
+# Engine keywords matched anywhere in the target path.
+_ENGINE_TARGET_KEYWORDS = (
+    "elasticsearch",
+    "opensearch",
+    "typesense",
+    "clickhouse",
+    "/solr",
+)
+
+
+def _is_engine_target(target: str) -> bool:
+    t = (target or "").strip().lower()
+    if not t:
+        return False
+    if any(t.startswith(p) for p in _ENGINE_TARGET_PREFIXES):
+        return True
+    return any(k in t for k in _ENGINE_TARGET_KEYWORDS)
+
+
+class TestUnpinnedNoEngineVolume(unittest.TestCase):
+    def test_unpinned_roles_declare_no_engine_data_volume(self) -> None:
+        roles_root = PROJECT_ROOT / "roles"
+        if not roles_root.is_dir():
+            self.skipTest(f"roles/ not present at {roles_root}")
+
+        violations: list[str] = []
+        for volumes_yml in sorted(roles_root.glob(f"*/{ROLE_FILE_META_VOLUMES}")):
+            role_id = volumes_yml.parent.parent.name
+            if (
+                str(get_role_default_placement(role_id) or "").strip().lower()
+                == "manager"
+            ):
+                continue  # pinned: volumes are node-local, engine data is safe
+            try:
+                content = load_yaml_any(str(volumes_yml), default_if_missing={})
+            except Exception as exc:
+                violations.append(f"{role_id}: YAML parse error: {exc}")
+                continue
+            if not isinstance(content, dict):
+                continue
+            for key, entry in content.items():
+                if not isinstance(entry, dict) or entry.get("type") != "volume":
+                    continue
+                for mount in entry.get("mounts") or []:
+                    if not isinstance(mount, dict):
+                        continue
+                    if _is_engine_target(mount.get("target", "")):
+                        violations.append(
+                            f"{role_id}: volume '{key}' mounts engine data at "
+                            f"'{mount.get('target')}' but the role is not pinned. "
+                            "Engine state cannot live on NFS; pin the role "
+                            "(default_placement: manager) or externalise the engine."
+                        )
+                        break
+
+        if violations:
+            self.fail(
+                "unpinned roles with NFS-hostile engine data volumes:\n"
+                + "\n".join(f"  - {v}" for v in violations)
+            )
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

tests/unit/plugins/filter/test_compose_volumes.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import sys
 import unittest
 from typing import Any
+from unittest import mock
 
 from ansible.errors import AnsibleFilterError
 
@@ -264,24 +265,31 @@ class TestComposeVolumes(unittest.TestCase):
         self.assertEqual(vol["driver_opts"]["device"], "/var/lib/infinito/app_images")
         self.assertNotIn("nfs", vol)
 
-    def test_swarm_nfs_disabled_volume_stays_local(self):
+    def test_swarm_pinned_role_stays_node_local(self):
+        # NFS is derived from the role's placement: a pinned role
+        # (default_placement: manager) keeps its volumes node-local even on a
+        # swarm/nfs backend, so engine state stays put instead of being shared.
         apps = self._base_apps()
-        rendered = _call(
-            apps,
-            "app",
-            extra_volumes={
-                "images": {"name": "app_images", "nfs": True},
-                "data": {"name": "app_data"},
-            },
-            deployment_mode="swarm",
-            storage={
-                "backend": "nfs",
-                "nfs": {"server": "10.0.0.20", "export_base": "/srv/nfs"},
-            },
-        )
+        with mock.patch(
+            "plugins.filter.compose_volumes.get_role_default_placement",
+            return_value="manager",
+        ):
+            rendered = _call(
+                apps,
+                "app",
+                extra_volumes={
+                    "images": {"name": "app_images"},
+                    "data": {"name": "app_data"},
+                },
+                deployment_mode="swarm",
+                storage={
+                    "backend": "nfs",
+                    "nfs": {"server": "10.0.0.20", "export_base": "/srv/nfs"},
+                },
+            )
         data = self._parse_yaml(rendered)
 
-        self.assertIn("driver_opts", data["volumes"]["images"])
+        self.assertNotIn("driver_opts", data["volumes"]["images"])
         self.assertNotIn("driver_opts", data["volumes"]["data"])
 
     def test_compose_mode_ignores_nfs_flag(self):

utils/manager/inventory.py

@@ -212,6 +212,19 @@ class InventoryManager:
                 self.value_generator.generate_value("alphanumeric")
             )
 
+        for engine_name in ("rabbitmq", "elasticsearch", "typesense"):
+            engine_cfg = (
+                services.get(engine_name) if isinstance(services, dict) else None
+            )
+            if isinstance(engine_cfg, dict) and is_explicit_truth(
+                engine_cfg.get("enabled")
+            ):
+                apps = self.inventory.setdefault("applications", {})
+                target = apps.setdefault(app_id, {})
+                target.setdefault("credentials", {})[f"{engine_name}_password"] = (
+                    self.value_generator.generate_value("alphanumeric")
+                )
+
     def apply_schema(self) -> dict:
         """
         Apply schema into inventory for:

utils/networks/render.py

@@ -197,6 +197,15 @@ def render_compose_networks(
             lines.append("    attachable: true")
             lines.append("    driver_opts:")
             lines.append(f'      encrypted: "{"true" if swarm_encrypted else "false"}"')
+            if not is_own_shared_net_provider:
+                subnet = lookup_config(
+                    application_id, "server.networks.local.subnet", ""
+                )
+                if subnet:
+                    lines.append("    ipam:")
+                    lines.append("      driver: default")
+                    lines.append("      config:")
+                    lines.append(f"        - subnet: {subnet}")
         elif is_own_shared_net_provider:
             lines.append("    driver: bridge")
         else:

utils/roles/applications/services/engines.py

@@ -0,0 +1,98 @@
+"""Central "engine" dependencies (cache / queue / search / vector / DNS resolver).
+
+A role may consume an engine either *embedded* (a sidecar inside its own stack) or
+*central* (a single pinned ``svc-db-*`` / ``svc-dns-*`` stack shared by many roles),
+selected per consumer with the ``shared`` flag exactly like the RDBMS
+(postgres / mariadb) pattern in :mod:`utils.roles.applications.services.database`.
+
+See ``docs/architecture/central-engines.md`` for the rationale (engine on-disk state
+cannot live on NFS, so centralising it keeps the application roles unpinned and
+NFS-shareable in swarm).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+# Engine service-key -> descriptor.
+#   svc:       central role id that owns the engine when ``shared: true``.
+#   port:      default container port.
+#   isolation: how a consumer gets its own logical slice on the central instance.
+#              acl  -> dedicated user + key-prefix (redis)
+#              vhost-> dedicated virtual host + user (rabbitmq)
+#              index-> dedicated index/alias namespace (elasticsearch)
+#              apikey -> scoped api key (typesense)
+#              collection -> per-consumer collection prefix (qdrant)
+#              prefix -> key-prefix only, no enforcement (memcached)
+#              none -> shared, no per-consumer split (unbound resolver)
+ENGINES: dict[str, dict[str, Any]] = {
+    "redis": {"svc": "svc-db-redis", "port": 6379, "isolation": "acl"},
+    "memcached": {"svc": "svc-db-memcached", "port": 11211, "isolation": "prefix"},
+    "rabbitmq": {"svc": "svc-db-rabbitmq", "port": 5672, "isolation": "vhost"},
+    "elasticsearch": {
+        "svc": "svc-db-elasticsearch",
+        "port": 9200,
+        "isolation": "index",
+    },
+    "typesense": {"svc": "svc-db-typesense", "port": 8108, "isolation": "apikey"},
+    "qdrant": {"svc": "svc-db-qdrant", "port": 6333, "isolation": "collection"},
+    "unbound": {"svc": "svc-dns-unbound", "port": 53, "isolation": "none"},
+}
+
+ENGINE_KEYS = tuple(ENGINES)
+
+
+def _as_mapping(value: Any) -> dict[str, Any]:
+    return value if isinstance(value, dict) else {}
+
+
+def get_services(
+    applications: Mapping[str, Any], application_id: str
+) -> dict[str, Any]:
+    return _as_mapping(_as_mapping(applications.get(application_id)).get("services"))
+
+
+def engine_service(
+    applications: Mapping[str, Any], application_id: str, engine: str
+) -> dict[str, Any]:
+    """The consumer's ``services.<engine>`` block (or empty)."""
+    return _as_mapping(get_services(applications, application_id).get(engine))
+
+
+def is_engine_enabled(
+    applications: Mapping[str, Any], application_id: str, engine: str
+) -> bool:
+    return bool(engine_service(applications, application_id, engine).get("enabled"))
+
+
+def is_engine_shared(
+    applications: Mapping[str, Any], application_id: str, engine: str
+) -> bool:
+    """Consumer wants the central instance (``enabled`` and ``shared``)."""
+    svc = engine_service(applications, application_id, engine)
+    return bool(svc.get("enabled")) and bool(svc.get("shared"))
+
+
+def is_engine_embedded(
+    applications: Mapping[str, Any], application_id: str, engine: str
+) -> bool:
+    """Consumer wants an embedded sidecar (``enabled`` and not ``shared``)."""
+    svc = engine_service(applications, application_id, engine)
+    return bool(svc.get("enabled")) and not bool(svc.get("shared"))
+
+
+def central_consumers(applications: Mapping[str, Any], engine: str) -> list[str]:
+    """Roles consuming ``engine`` as ``shared: true`` -- drives on-demand enable of
+    the central engine stack (deploy it only when at least one role needs it)."""
+    return sorted(
+        application_id
+        for application_id in applications
+        if is_engine_shared(applications, application_id, engine)
+    )
+
+
+def central_engine_needed(applications: Mapping[str, Any], engine: str) -> bool:
+    return bool(central_consumers(applications, engine))