Repository: geldata/gel
Pull request: Add TOML config file support
Sample TOML config file:

```toml
["cfg::Config"]
session_idle_timeout = "1 minute"
shared_buffers = "128MiB"
_pg_prepared_statement_cache_size = 10
listen_port = 5858

[[magic_smtp_config]]
_tname = "cfg::SMTPProviderConfig"
name = "gmail"
sender = "alice@example.com"
port = 587
security = "STARTTLS"

[[magic_smtp_config]]
_tname = "cfg::SMTPProviderConfig"
name = "outlook"
sender = "bob@example.com"
timeout_per_email = "PT1M48S"
```

The single-tenant server takes a new --config-file argument that points to a TOML config file like above. At the same time, a multi-tenant server accepts a new key "config-file" in its per-tenant config JSON. Each tenant could define its own TOML config file.

The TOML config file itself is reloadable (defined by --reload-config-files), and the config-file path in a multitenant config JSON file is also reloadable. The reloaded values do not affect running queries or active transactions; they are only effective with new queries, transactions, or backend connections.

The config values loaded from the TOML config file are placed in a new config tier, "configuration file," which overrides "environment variable" (and lower tiers) and is overridden by (not-yet-implemented) "command line" (and higher tiers). The root section in the TOML must be cfg at this moment.

Implementation-wise, the TOML values are JSON-ified and stored in the _edgecon_state temp table, so that different servers on the same backend Postgres may define different values in their TOML config files while not affecting each other (multi-tenancy for the backend). On reload, all alive pgcons shall run a SQL to re-insert reloaded TOML values as soon as they become idle; new pgcons automatically get to insert the reloaded TOML values.

- [x] Deal with backend_setting after reload
- [x] Add tests

Replaces #8059, Fixes #1325, Fixes #7990

Problem statement:
Allow reading configuration from a local file
Need an option to read server configuration from a specified JSON file in the local filesystem. 

Per-tenant configuration in multitenant
I think there are two components here:

- [ ] Per tenant "environmental" configuration, where we can do the equivalent of passing config in environment variables. I think we decided that a toml file was the plan, but I'm not sure we agreed on any specifics.
- [ ] `CONFIGURE INSTANCE`, which I think is currently disabled in multi-tenant

Hints from issue discussion:
Please use Hjson or Yaml. JSON configs are still ugly in 2020.
Truth be told, using YAML is also controversial. I've seen numerous comments on HN/GH with people complaining that whitespace-only nature of YAML makes it hard for some people to edit it. 🤷‍♂️ This is a bit painful to admit for me since both Elvis and I are huge YAML fans — it was in fact the first DSL for our schema (now we have ESDL).

While verbose and more machine than human friendly, JSON has some advantage: built-in support in every language and code editor there is + literally everybody is familiar with it. So if our config files aren't going to be huge I'd stick with JSON.

I also like TOML quite a bit, we can give it a try if we need to handle relatively complex configs.
> Truth be told, using YAML is also controversial.

That's why I've listed Hjson as the first option.

> While verbose and more machine than human friendly, JSON has some advantage:

It's not "more machine friendly" it's "less human-friendly".

> So if our config files aren't going to be huge I'd stick with JSON.

If config is larger than two lines, it needs comments. Also trailing comma is a common issue both with writing configs and especially with versioning them.

> I also like TOML quite a bit, we can give it a try if we need to handle relatively complex configs.

1. Toml is fine for simple configs having just few "sections". But structured data is much better even in JSON:

    a. This is because Toml keys can be scattered around across `dotted.keys` and sections in different parts of file. That is a feature, but can quickly make config hardly readable. Scattered list of dicts are a special nightmare.
    b. "Inline tables" (dicts denoted by curly braces) are single-line only. And you end up switching from inline mode to standard `[table]` after adding another key (which usually means also moving this entry to the end of the current table). Which often means you often can't group things together in a meaningful way.

2. Toml's scalars are only int/float/date/string. Which means `size=100` is fine but `size="100Mb"` needs to be quoted. This makes values with units like `100Mb` and `5sec` harder to explain and handle and thus less common in configs (which is a bad thing). Multi-line strings are also more useful in Hjson (indentation is correct)

So Hjson is basically a right superset of JSON fixing comments and trailing comma (and unquoted strings), but without introducing indentation-sensitive syntax (which is mostly what Yaml is hated for). This is currently my favorite configuration format (for simple configs that don't need Yaml's anchors and tags) although I haven't used it in production too much at this time.
> If config is larger than two lines, it needs comments.

This is a very good point against using JSON.
Having started using poetry recently, I've warmed up to TOML a whole lot. It's nearly as easy to edit in something like `nano` as an .ini file but is unambiguous and better standardized. 

@tailhook I don't fully understand the use case for "highly structured data" you're imagining here as it relates to a database configuration.
@agritheory there are a bunch of things. Few examples:

1. Inline mappings are **single-line only**: `key = {a='b', c='d'}` is fine, but then if you need add another key, it must be converted to:
    ```
    [key]
    a = 'b'
    c = 'd'
    ```
    and then, confusingly enough, this works:
    ```
    key = {a='b', c=[
      '1',
      '2',
    ]}
    ```
2. Sections can be messed up especially lists, let's see if you have a thing like this:
    ```toml
    [[sections]]
    x = 1
    [sections.subkey]
    y = 2
    ```
    This is already hard to read, but it's basically:
    ```json
    {"sections": [
      {"x": 1,
       "subkey": {"y": 2}},
    ]}
    ```
   And then, you want to add a new section. Intuition is to do this:
    ```toml
    [[section]]
    x = 1
    [[section]]
    x = 2
    [section.subkey]
    y = 2
    ```
    And you effectively moved `subkey` to a next section:
    ```json
    {"sections": [
      {"x": 1},
      {"x": 2,
       "subkey": {"y": 2}},
    ]}
    ```
   If you think that this is a lame mistake and contrived example, just note that between the `[[section]]` title and `[[section.items]]` can be screenful of lines (e.g. documentation comments). And someone may already put `[another-key]` in-between half year ago (and that, obviously, didn't break anything technically).

Commit message: Add TOML config file support

Patch to review:
diff --git a/edb/common/asyncutil.py b/edb/common/asyncutil.py
index 7d439a6172c..461859076ae 100644
--- a/edb/common/asyncutil.py
+++ b/edb/common/asyncutil.py
@@ -18,9 +18,20 @@
 
 
 from __future__ import annotations
-from typing import Callable, TypeVar, Awaitable
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    cast,
+    overload,
+    Self,
+    TypeVar,
+    Type,
+)
 
 import asyncio
+import inspect
+import warnings
 
 
 _T = TypeVar('_T')
@@ -140,3 +151,186 @@ async def debounce(
         batch = []
         last_signal = t
         target_time = None
+
+
+_Owner = TypeVar("_Owner")
+HandlerFunction = Callable[[], Awaitable[None]]
+HandlerMethod = Callable[[Any], Awaitable[None]]
+
+
+class ExclusiveTask:
+    """Manages to run a repeatable task once at a time."""
+
+    _handler: HandlerFunction
+    _task: asyncio.Task | None
+    _scheduled: bool
+    _stop_requested: bool
+
+    def __init__(self, handler: HandlerFunction) -> None:
+        self._handler = handler
+        self._task = None
+        self._scheduled = False
+        self._stop_requested = False
+
+    @property
+    def scheduled(self) -> bool:
+        return self._scheduled
+
+    async def _run(self) -> None:
+        if self._scheduled and not self._stop_requested:
+            self._scheduled = False
+        else:
+            return
+        try:
+            await self._handler()
+        finally:
+            if self._scheduled and not self._stop_requested:
+                self._task = asyncio.create_task(self._run())
+            else:
+                self._task = None
+
+    def schedule(self) -> None:
+        """Schedule to run the task as soon as possible.
+
+        If already scheduled, nothing happens; it won't queue up.
+
+        If the task is already running, it will be scheduled to run again as
+        soon as the running task is done.
+        """
+        if not self._stop_requested:
+            self._scheduled = True
+            if self._task is None:
+                self._task = asyncio.create_task(self._run())
+
+    async def stop(self) -> None:
+        """Cancel scheduled task and wait for the running one to finish.
+
+        After an ExclusiveTask is stopped, no more new schedules are allowed.
+        """
+        self._scheduled = False
+        self._stop_requested = True
+        if self._task is not None:
+            await self._task
+
+
+class ExclusiveTaskProperty:
+    _method: HandlerMethod
+    _name: str | None
+
+    def __init__(
+        self, method: HandlerMethod, *, slot: str | None = None
+    ) -> None:
+        self._method = method
+        self._name = slot
+
+    def __set_name__(self, owner: Type[_Owner], name: str) -> None:
+        if (slots := getattr(owner, "__slots__", None)) is not None:
+            if self._name is None:
+                raise TypeError("missing slot in @exclusive_task()")
+            if self._name not in slots:
+                raise TypeError(
+                    f"slot {self._name!r} must be defined in __slots__"
+                )
+
+        if self._name is None:
+            self._name = name
+
+    @overload
+    def __get__(self, instance: None, owner: Type[_Owner]) -> Self: ...
+
+    @overload
+    def __get__(
+        self, instance: _Owner, owner: Type[_Owner]
+    ) -> ExclusiveTask: ...
+
+    def __get__(
+        self, instance: _Owner | None, owner: Type[_Owner]
+    ) -> ExclusiveTask | Self:
+        # getattr on the class
+        if instance is None:
+            return self
+
+        assert self._name is not None
+
+        # getattr on an object with __dict__
+        if (d := getattr(instance, "__dict__", None)) is not None:
+            if rv := d.get(self._name, None):
+                return rv
+            rv = ExclusiveTask(self._method.__get__(instance, owner))
+            d[self._name] = rv
+            return rv
+
+        # getattr on an object with __slots__
+        else:
+            if rv := getattr(instance, self._name, None):
+                return rv
+            rv = ExclusiveTask(self._method.__get__(instance, owner))
+            setattr(instance, self._name, rv)
+            return rv
+
+
+ExclusiveTaskDecorator = Callable[
+    [HandlerFunction | HandlerMethod], ExclusiveTask | ExclusiveTaskProperty
+]
+
+
+def _exclusive_task(
+    handler: HandlerFunction | HandlerMethod, *, slot: str | None
+) -> ExclusiveTask | ExclusiveTaskProperty:
+    sig = inspect.signature(handler)
+    params = list(sig.parameters.values())
+    if len(params) == 0:
+        handler = cast(HandlerFunction, handler)
+        if slot is not None:
+            warnings.warn(
+                "slot is specified but unused in @exclusive_task()",
+                stacklevel=2,
+            )
+        return ExclusiveTask(handler)
+    elif len(params) == 1 and params[0].kind in (
+        inspect.Parameter.POSITIONAL_ONLY,
+        inspect.Parameter.POSITIONAL_OR_KEYWORD,
+    ):
+        handler = cast(HandlerMethod, handler)
+        return ExclusiveTaskProperty(handler, slot=slot)
+    else:
+        raise TypeError("bad signature")
+
+
+@overload
+def exclusive_task(handler: HandlerFunction) -> ExclusiveTask: ...
+
+
+@overload
+def exclusive_task(
+    handler: HandlerMethod, *, slot: str | None = None
+) -> ExclusiveTaskProperty: ...
+
+
+@overload
+def exclusive_task(*, slot: str | None = None) -> ExclusiveTaskDecorator: ...
+
+
+def exclusive_task(
+    handler: HandlerFunction | HandlerMethod | None = None,
+    *,
+    slot: str | None = None,
+) -> ExclusiveTask | ExclusiveTaskProperty | ExclusiveTaskDecorator:
+    """Convert an async function into an ExclusiveTask.
+
+    This decorator can be applied to either top-level functions or methods
+    in a class. In the latter case, the exclusiveness is bound to each object
+    of the owning class. If the owning class defines __slots__, you must also
+    define an extra slot to store the exclusive state and tell exclusive_task()
+    by providing the `slot` argument.
+    """
+    if handler is None:
+
+        def decorator(
+            handler: HandlerFunction | HandlerMethod,
+        ) -> ExclusiveTask | ExclusiveTaskProperty:
+            return _exclusive_task(handler, slot=slot)
+
+        return decorator
+
+    return _exclusive_task(handler, slot=slot)
diff --git a/edb/pgsql/metaschema.py b/edb/pgsql/metaschema.py
index e9eeec84551..b7976444658 100644
--- a/edb/pgsql/metaschema.py
+++ b/edb/pgsql/metaschema.py
@@ -3116,6 +3116,7 @@ def __init__(self) -> None:
                 'postgres environment variable',
                 'postgres configuration file',
                 'environment variable',
+                'configuration file',
                 'command line',
                 'postgres command line',
                 'postgres global',
@@ -3612,6 +3613,7 @@ class SysConfigFullFunction(trampoline.VersionedFunction):
                 (CASE
                     WHEN s.type = 'A' THEN 'command line'
                     WHEN s.type = 'E' THEN 'environment variable'
+                    WHEN s.type = 'F' THEN 'configuration file'
                     ELSE 'session'
                 END) AS source,
                 FALSE AS from_backend  -- only 'B' is for backend settings
diff --git a/edb/server/args.py b/edb/server/args.py
index f76505df3f3..820a261fdf3 100644
--- a/edb/server/args.py
+++ b/edb/server/args.py
@@ -150,6 +150,7 @@ class ReloadTrigger(enum.StrEnum):
     3. Multi-tenant config file (server config)
     4. Readiness state (server or tenant config)
     5. JWT sub allowlist and revocation list (server or tenant config)
+    6. The TOML config file (server or tenant config)
     """
 
     Default = "default"
@@ -265,6 +266,7 @@ class ServerConfig(NamedTuple):
     disable_dynamic_system_config: bool
     reload_config_files: ReloadTrigger
     net_worker_mode: NetWorkerMode
+    config_file: Optional[pathlib.Path]
 
     startup_script: Optional[StartupScript]
     status_sinks: List[Callable[[str], None]]
@@ -1106,6 +1108,13 @@ def resolve_envvar_value(self, ctx: click.Context):
         default='default',
         help='Controls how the std::net workers work.',
     ),
+    click.option(
+        "--config-file", type=PathPath(), metavar="PATH",
+        envvar="GEL_SERVER_CONFIG_FILE",
+        cls=EnvvarResolver,
+        help='Path to a TOML file to configure the server.',
+        hidden=True,
+    ),
 ])
 
 
@@ -1534,6 +1543,7 @@ def parse_args(**kwargs: Any):
             "readiness_state_file",
             "jwt_sub_allowlist_file",
             "jwt_revocation_list_file",
+            "config_file",
         ):
             if kwargs.get(name):
                 opt = "--" + name.replace("_", "-")
diff --git a/edb/server/config/__init__.py b/edb/server/config/__init__.py
index 4cf98290979..c3420ae0272 100644
--- a/edb/server/config/__init__.py
+++ b/edb/server/config/__init__.py
@@ -18,7 +18,9 @@
 
 
 from __future__ import annotations
-from typing import Any, Mapping
+from typing import Any, Mapping, TypedDict
+
+import enum
 
 import immutables
 
@@ -50,9 +52,25 @@
     'load_ext_settings_from_schema',
     'get_compilation_config',
     'QueryCacheMode',
+    'ConState', 'ConStateType',
 )
 
 
+# See edb/server/pgcon/connect.py for documentation of the types
+class ConStateType(enum.StrEnum):
+    session_config = "C"
+    backend_session_config = "B"
+    command_line_argument = "A"
+    environment_variable = "E"
+    config_file = "F"
+
+
+class ConState(TypedDict):
+    name: str
+    value: Any
+    type: ConStateType
+
+
 def lookup(
     name: str,
     *configs: Mapping[str, SettingValue],
diff --git a/edb/server/main.py b/edb/server/main.py
index de1de206150..a4c76f5ca53 100644
--- a/edb/server/main.py
+++ b/edb/server/main.py
@@ -36,7 +36,6 @@
 
 import asyncio
 import contextlib
-import enum
 import json
 import logging
 import os
@@ -199,6 +198,7 @@ async def _run_server(
     do_setproctitle: bool,
     new_instance: bool,
     compiler: edbcompiler.Compiler,
+    init_con_data: list[config.ConState],
 ):
 
     sockets = service_manager.get_activation_listen_sockets()
@@ -218,10 +218,12 @@ async def _run_server(
             backend_adaptive_ha=args.backend_adaptive_ha,
             extensions_dir=args.extensions_dir,
         )
+        tenant.set_init_con_data(init_con_data)
         tenant.set_reloadable_files(
             readiness_state_file=args.readiness_state_file,
             jwt_sub_allowlist_file=args.jwt_sub_allowlist_file,
             jwt_revocation_list_file=args.jwt_revocation_list_file,
+            config_file=args.config_file,
         )
         ss = server.Server(
             runstate_dir=runstate_dir,
@@ -258,6 +260,8 @@ async def _run_server(
             await tenant.load_sidechannel_configs(
                 json.loads(magic_smtp), compiler=compiler
             )
+        if args.config_file:
+            await tenant.load_config_file(compiler)
         # This coroutine runs as long as the server,
         # and compiler(.state) is *heavy*, so make sure we don't
         # keep a reference to it.
@@ -303,6 +307,7 @@ def load_configuration(_signum):
                     args.tls_client_ca_file,
                 )
                 ss.load_jwcrypto(args.jws_key_file)
+                tenant.reload_config_file.schedule()
             except Exception:
                 logger.critical(
                     "Unexpected error occurred during reload configuration; "
@@ -531,10 +536,12 @@ async def run_server(
                 compiler_state.config_spec,
             )
 
-            sys_config, backend_settings = initialize_static_cfg(
-                args,
-                is_remote_cluster=True,
-                compiler=compiler,
+            sys_config, backend_settings, init_con_data = (
+                initialize_static_cfg(
+                    args,
+                    is_remote_cluster=True,
+                    compiler=compiler,
+                )
             )
             del compiler
             with _internal_state_dir(runstate_dir, args) as (
@@ -557,6 +564,7 @@ async def run_server(
                     internal_runstate_dir=int_runstate_dir,
                     do_setproctitle=do_setproctitle,
                     compiler_state=compiler_state,
+                    init_con_data=init_con_data,
                 )
         except server.StartupError as e:
             abort(str(e))
@@ -616,7 +624,7 @@ async def run_server(
 
         new_instance, compiler = await _init_cluster(cluster, args)
 
-        _, backend_settings = initialize_static_cfg(
+        _, backend_settings, init_con_data = initialize_static_cfg(
             args,
             is_remote_cluster=not is_local_cluster,
             compiler=compiler,
@@ -673,6 +681,7 @@ async def run_server(
                     do_setproctitle=do_setproctitle,
                     new_instance=new_instance,
                     compiler=compiler,
+                    init_con_data=init_con_data,
                 )
 
     except server.StartupError as e:
@@ -807,28 +816,23 @@ def main_dev():
     main()
 
 
-class Source(enum.StrEnum):
-    command_line_argument = "A"
-    environment_variable = "E"
-
-
-sources = {
-    Source.command_line_argument: "command line argument",
-    Source.environment_variable: "environment variable",
-}
-
-
 def initialize_static_cfg(
     args: srvargs.ServerConfig,
     is_remote_cluster: bool,
     compiler: edbcompiler.Compiler,
-) -> Tuple[Mapping[str, config.SettingValue], Dict[str, str]]:
+) -> Tuple[
+    Mapping[str, config.SettingValue], Dict[str, str], list[config.ConState]
+]:
     result = {}
-    init_con_script_data = []
+    init_con_script_data: list[config.ConState] = []
     backend_settings = {}
     config_spec = compiler.state.config_spec
+    sources = {
+        config.ConStateType.command_line_argument: "command line argument",
+        config.ConStateType.environment_variable: "environment variable",
+    }
 
-    def add_config_values(obj: dict[str, Any], source: Source):
+    def add_config_values(obj: dict[str, Any], source: config.ConStateType):
         settings = compiler.compile_structured_config(
             {"cfg::Config": obj}, source=sources[source]
         )["cfg::Config"]
@@ -837,7 +841,7 @@ def add_config_values(obj: dict[str, Any], source: Source):
 
             if is_remote_cluster:
                 if setting.backend_setting and setting.requires_restart:
-                    if source == Source.command_line_argument:
+                    if source == config.ConStateType.command_line_argument:
                         where = "on command line"
                     else:
                         where = "as an environment variable"
@@ -876,7 +880,7 @@ def add_config_values(obj: dict[str, Any], source: Source):
                 if cfg != name:
                     values[cfg] = value
     if values:
-        add_config_values(values, Source.environment_variable)
+        add_config_values(values, config.ConStateType.environment_variable)
 
     values = {}
     if args.bind_addresses:
@@ -884,13 +888,9 @@ def add_config_values(obj: dict[str, Any], source: Source):
     if args.port:
         values["listen_port"] = args.port
     if values:
-        add_config_values(values, Source.command_line_argument)
-
-    if init_con_script_data:
-        from . import pgcon
-        pgcon.set_init_con_script_data(init_con_script_data)
+        add_config_values(values, config.ConStateType.command_line_argument)
 
-    return immutables.Map(result), backend_settings
+    return immutables.Map(result), backend_settings, init_con_script_data
 
 
 if __name__ == '__main__':
diff --git a/edb/server/multitenant.py b/edb/server/multitenant.py
index c4fdbfc33c1..513ff16eaf5 100644
--- a/edb/server/multitenant.py
+++ b/edb/server/multitenant.py
@@ -63,6 +63,7 @@
         "jwt-revocation-list-file": str,
         "readiness-state-file": str,
         "admin": bool,
+        "config-file": str,
     },
 )
 
@@ -70,6 +71,7 @@
 class MultiTenantServer(server.BaseServer):
     _config_file: pathlib.Path
     _sys_config: Mapping[str, config.SettingValue]
+    _init_con_data: list[config.ConState]
     _backend_settings: Mapping[str, str]
 
     _tenants_by_sslobj: MutableMapping
@@ -89,6 +91,7 @@ def __init__(
         *,
         compiler_pool_tenant_cache_size: int,
         sys_config: Mapping[str, config.SettingValue],
+        init_con_data: list[config.ConState],
         backend_settings: Mapping[str, str],
         sys_queries: Mapping[str, bytes],
         report_config_typedesc: dict[defines.ProtocolVersion, bytes],
@@ -97,6 +100,7 @@ def __init__(
         super().__init__(**kwargs)
         self._config_file = config_file
         self._sys_config = sys_config
+        self._init_con_data = init_con_data
         self._backend_settings = backend_settings
         self._compiler_pool_tenant_cache_size = compiler_pool_tenant_cache_size
 
@@ -243,13 +247,18 @@ async def _create_tenant(self, conf: TenantConfig) -> edbtenant.Tenant:
             max_backend_connections=max_conns,
             backend_adaptive_ha=conf.get("backend-adaptive-ha", False),
         )
+        tenant.set_init_con_data(self._init_con_data)
+        config_file = conf.get("config-file")
         tenant.set_reloadable_files(
             readiness_state_file=conf.get("readiness-state-file"),
             jwt_sub_allowlist_file=conf.get("jwt-sub-allowlist-file"),
             jwt_revocation_list_file=conf.get("jwt-revocation-list-file"),
+            config_file=config_file,
         )
         tenant.set_server(self)
         tenant.load_jwcrypto()
+        if config_file:
+            await tenant.load_config_file(self.get_compiler_pool())
         try:
             await tenant.init_sys_pgcon()
             await tenant.init()
@@ -379,6 +388,7 @@ async def _reload_tenant(self, serial: int, sni: str, conf: TenantConfig):
                             "readiness-state-file",
                             "jwt-sub-allowlist-file",
                             "jwt-revocation-list-file",
+                            "config-file",
                         }
                         if diff:
                             logger.warning(
@@ -395,6 +405,7 @@ async def _reload_tenant(self, serial: int, sni: str, conf: TenantConfig):
                                 "jwt-sub-allowlist-file"),
                             jwt_revocation_list_file=conf.get(
                                 "jwt-revocation-list-file"),
+                            config_file=conf.get("config-file"),
                         ):
                             # none of the reloadable values was modified
                             return
@@ -429,6 +440,7 @@ async def run_server(
     args: srvargs.ServerConfig,
     *,
     sys_config: Mapping[str, config.SettingValue],
+    init_con_data: list[config.ConState],
     backend_settings: Mapping[str, str],
     sys_queries: Mapping[str, bytes],
     report_config_typedesc: dict[defines.ProtocolVersion, bytes],
@@ -444,6 +456,7 @@ async def run_server(
         ss = MultiTenantServer(
             multitenant_config_file,
             sys_config=sys_config,
+            init_con_data=init_con_data,
             backend_settings=backend_settings,
             sys_queries=sys_queries,
             report_config_typedesc=report_config_typedesc,
diff --git a/edb/server/pgcon/__init__.py b/edb/server/pgcon/__init__.py
index c6bc771fb67..2edf5755519 100644
--- a/edb/server/pgcon/__init__.py
+++ b/edb/server/pgcon/__init__.py
@@ -31,14 +31,12 @@
 )
 from .connect import (
     pg_connect,
-    set_init_con_script_data,
     SETUP_TEMP_TABLE_SCRIPT,
     SETUP_CONFIG_CACHE_SCRIPT,
 )
 
 __all__ = (
     'pg_connect',
-    'set_init_con_script_data',
     'PGConnection',
     'BackendError',
     'BackendConnectionError',
diff --git a/edb/server/pgcon/connect.py b/edb/server/pgcon/connect.py
index 7d912b76c70..899318b5459 100644
--- a/edb/server/pgcon/connect.py
+++ b/edb/server/pgcon/connect.py
@@ -18,12 +18,10 @@
 
 from __future__ import annotations
 
-import json
 import logging
 import textwrap
 
 from edb.pgsql.common import quote_ident as pg_qi
-from edb.pgsql.common import quote_literal as pg_ql
 from edb.pgsql import params as pg_params
 from edb.server import pgcon
 
@@ -33,7 +31,6 @@
 logger = logging.getLogger('edb.server')
 
 INIT_CON_SCRIPT: bytes | None = None
-INIT_CON_SCRIPT_DATA = ''
 
 # The '_edgecon_state table' is used to store information about
 # the current session. The `type` column is one character, with one
@@ -45,12 +42,16 @@
 #   a corresponding Postgres config setting.
 # * 'A': an instance-level config setting from command-line arguments
 # * 'E': an instance-level config setting from environment variable
+# * 'F': an instance/tenant-level config setting from the TOML config file
+#
+# Please also update ConStateType in edb/server/config/__init__.py if changed.
 SETUP_TEMP_TABLE_SCRIPT = '''
         CREATE TEMPORARY TABLE _edgecon_state (
             name text NOT NULL,
             value jsonb NOT NULL,
             type text NOT NULL CHECK(
-                type = 'C' OR type = 'B' OR type = 'A' OR type = 'E'),
+                type = 'C' OR type = 'B' OR type = 'A' OR type = 'E'
+                OR type = 'F'),
             UNIQUE(name, type)
         );
 '''.strip()
@@ -82,8 +83,6 @@ def _build_init_con_script(*, check_pg_is_in_recovery: bool) -> bytes:
         {SETUP_TEMP_TABLE_SCRIPT}
         {SETUP_CONFIG_CACHE_SCRIPT}
 
-        {INIT_CON_SCRIPT_DATA}
-
         PREPARE _clear_state AS
             WITH x1 AS (
                 DELETE FROM _config_cache
@@ -193,15 +192,3 @@ async def pg_connect(
             raise
 
     return pgconn
-
-
-def set_init_con_script_data(cfg):
-    global INIT_CON_SCRIPT, INIT_CON_SCRIPT_DATA
-    INIT_CON_SCRIPT = None
-    INIT_CON_SCRIPT_DATA = (
-        f'''
-        INSERT INTO _edgecon_state
-        SELECT * FROM jsonb_to_recordset({pg_ql(json.dumps(cfg))}::jsonb)
-        AS cfg(name text, value jsonb, type text);
-    '''
-    ).strip()
diff --git a/edb/server/pgcon/pgcon.pxd b/edb/server/pgcon/pgcon.pxd
index 93da530be39..f1260f07bc2 100644
--- a/edb/server/pgcon/pgcon.pxd
+++ b/edb/server/pgcon/pgcon.pxd
@@ -138,6 +138,7 @@ cdef class PGConnection:
         public object pinned_by
 
         object last_state
+        public object last_init_con_data
 
         str last_indirect_return
 
diff --git a/edb/server/pgcon/pgcon.pyi b/edb/server/pgcon/pgcon.pyi
index 7c6821bf028..8e961dfdc62 100644
--- a/edb/server/pgcon/pgcon.pyi
+++ b/edb/server/pgcon/pgcon.pyi
@@ -36,8 +36,6 @@ class BackendConnectionError(BackendError): ...
 class BackendPrivilegeError(BackendError): ...
 class BackendCatalogNameError(BackendError): ...
 
-def set_init_con_script_data(cfg: list[dict[str, Any]]): ...
-
 class PGConnection(asyncio.Protocol):
 
     idle: bool
@@ -47,6 +45,7 @@ class PGConnection(asyncio.Protocol):
     parameter_status: dict[str, str]
     backend_secret: int
     is_ssl: bool
+    last_init_con_data: object
 
     def __init__(self, dbname): ...
     async def close(self): ...
diff --git a/edb/server/tenant.py b/edb/server/tenant.py
index b1c1656e975..f296511919b 100644
--- a/edb/server/tenant.py
+++ b/edb/server/tenant.py
@@ -42,6 +42,7 @@
 import pickle
 import struct
 import sys
+import textwrap
 import time
 import tomllib
 import uuid
@@ -51,6 +52,7 @@
 
 from edb import buildmeta
 from edb import errors
+from edb.common import asyncutil
 from edb.common import retryloop
 from edb.common import verutils
 from edb.common.log import current_tenant
@@ -76,6 +78,7 @@
 
     from . import pgcluster
     from . import server as edbserver
+    from . import compiler_pool as edbcompiler_pool
 
 
 logger = logging.getLogger("edb.server")
@@ -116,12 +119,16 @@ class Tenant(ha_base.ClusterProtocol):
     _suggested_client_pool_size: int
     _pg_pool: connpool.Pool
     _pg_unavailable_msg: str | None
+    _init_con_data: list[config.ConState]
+    _init_con_sql: bytes | None
+    _update_init_con_sql: bytes | None
 
     _ha_master_serial: int
     _backend_adaptive_ha: adaptive_ha.AdaptiveHASupport | None
     _readiness_state_file: pathlib.Path | None
     _readiness: srvargs.ReadinessState
     _readiness_reason: str
+    _config_file: pathlib.Path | None
 
     _extensions_dirs: tuple[pathlib.Path, ...]
 
@@ -182,6 +189,7 @@ def __init__(
         self._readiness_state_file = None
         self._readiness = srvargs.ReadinessState.Default
         self._readiness_reason = ""
+        self._config_file = None
 
         self._max_backend_connections = max_backend_connections
         self._suggested_client_pool_size = max(
@@ -199,6 +207,8 @@ def __init__(
         self._pg_unavailable_msg = None
         self._block_new_connections = set()
         self._report_config_data = {}
+        self._init_con_data = []
+        self._init_con_sql = self._update_init_con_sql = None
 
         # DB state will be initialized in init().
         self._dbindex = None
@@ -222,6 +232,7 @@ def set_reloadable_files(
         readiness_state_file: str | pathlib.Path | None = None,
         jwt_sub_allowlist_file: str | pathlib.Path | None = None,
         jwt_revocation_list_file: str | pathlib.Path | None = None,
+        config_file: str | pathlib.Path | None = None,
     ) -> bool:
         rv = False
 
@@ -243,6 +254,12 @@ def set_reloadable_files(
             self._jwt_revocation_list_file = jwt_revocation_list_file
             rv = True
 
+        if isinstance(config_file, str):
+            config_file = pathlib.Path(config_file)
+        if self._config_file != config_file:
+            self._config_file = config_file
+            rv = True
+
         return rv
 
     def set_server(self, server: edbserver.BaseServer) -> None:
@@ -250,7 +267,12 @@ def set_server(self, server: edbserver.BaseServer) -> None:
         self.__loop = server.get_loop()
 
     async def load_sidechannel_configs(
-        self, value: Any, *, compiler: edbcompiler.Compiler | None = None
+        self,
+        value: Any,
+        *,
+        compiler: (
+            edbcompiler.Compiler | edbcompiler_pool.AbstractPool | None
+        ) = None,
     ) -> None:
         if compiler is None:
             compiler = self._server.get_compiler_pool()
@@ -258,6 +280,8 @@ async def load_sidechannel_configs(
             {"cfg::Config": {"email_providers": value}}, source="magic",
             allow_nested=True,
         )
+        if asyncio.iscoroutine(result):
+            result = await result
         email_providers = result["cfg::Config"]["email_providers"]
         self._sidechannel_email_configs = list(email_providers.value)
 
@@ -598,6 +622,15 @@ def reload_jwt_revocation_list_file():
                 )
             )
 
+        if self._config_file is not None:
+
+            def reload_config_file():
+                self.reload_config_file.schedule()
+
+            self._file_watch_finalizers.append(
+                self.server.monitor_fs(self._config_file, reload_config_file)
+            )
+
     async def start_accepting_new_tasks(self) -> None:
         assert self._task_group is None
         self._task_group = asyncio.TaskGroup()
@@ -695,6 +728,28 @@ def terminate_sys_pgcon(self) -> None:
             self.__sys_pgcon = None
         del self._sys_pgcon_waiter
 
+    def set_init_con_data(self, data: list[config.ConState]) -> None:
+        self._init_con_data = data
+        self._init_con_sql = None
+        self._update_init_con_sql = None
+        if data:
+            self._init_con_sql = self._make_init_con_sql(data)
+
+    def _make_init_con_sql(self, data: list[config.ConState]) -> bytes:
+        if not data:
+            return b""
+
+        from edb.pgsql import common
+
+        quoted_json = common.quote_literal(json.dumps(data))
+        return textwrap.dedent(
+            f'''
+                INSERT INTO _edgecon_state
+                    SELECT * FROM jsonb_to_recordset({quoted_json}::jsonb)
+                        AS cfg(name text, value jsonb, type text);
+            '''
+        ).strip().encode()
+
     async def _pg_connect(
         self,
         dbname: str,
@@ -714,6 +769,11 @@ async def _pg_connect(
             )
             if self._server.stmt_cache_size is not None:
                 rv.set_stmt_cache_size(self._server.stmt_cache_size)
+
+            if self._init_con_sql:
+                await rv.sql_execute(self._init_con_sql)
+            rv.last_init_con_data = self._init_con_data
+
         except Exception:
             metrics.backend_connection_establishment_errors.inc(
                 1.0, self._instance_name
@@ -965,11 +1025,26 @@ async def acquire_pgcon(self, dbname: str) -> pgcon.PGConnection:
 
         for _ in range(self._pg_pool.max_capacity):
             conn = await self._pg_pool.acquire(dbname)
-            if conn.is_healthy():
-                return conn
+            if not conn.is_healthy():
+                logger.warning("acquired an unhealthy pgcon; discard now")
+            elif conn.last_init_con_data is not self._init_con_data:
+                if self._update_init_con_sql:
+                    try:
+                        await conn.sql_execute(self._update_init_con_sql)
+                    except Exception as e:
+                        logger.warning(
+                            "failed to update pgcon; discard now: %s", e
+                        )
+                    else:
+                        conn.last_init_con_data = self._init_con_data
+                        return conn
+                else:
+                    logger.warning(
+                        "don't know how to update pgcon; discard now"
+                    )
             else:
-                logger.warning("Acquired an unhealthy pgcon; discard now.")
-                self._pg_pool.release(dbname, conn, discard=True)
+                return conn
+            self._pg_pool.release(dbname, conn, discard=True)
         else:
             # This is unlikely to happen, but we defer to the caller to retry
             # when it does happen
@@ -1349,9 +1424,13 @@ async def _load_reported_config(self) -> None:
     async def _load_sys_config(
         self,
         query_name: str = "sysconfig",
+        syscon: pgcon.PGConnection | None = None,
     ) -> Mapping[str, config.SettingValue]:
-        async with self.use_sys_pgcon() as syscon:
-            query = self._server.get_sys_query(query_name)
+        query = self._server.get_sys_query(query_name)
+        if syscon is None:
+            async with self.use_sys_pgcon() as syscon:
+                sys_config_json = await syscon.sql_fetch_val(query)
+        else:
             sys_config_json = await syscon.sql_fetch_val(query)
 
         return config.from_json(self._server.config_settings, sys_config_json)
@@ -1563,6 +1642,83 @@ def set_readiness_state(self, state: srvargs.ReadinessState, reason: str):
         self._readiness = state
         self._readiness_reason = reason
 
+    @asyncutil.exclusive_task
+    async def reload_config_file(self):
+        if self._config_file is None:
+            return
+
+        try:
+            await self._reload_config_file()
+        except Exception:
+            logger.error("failed to reload config file", exc_info=True)
+            metrics.background_errors.inc(
+                1.0, self._instance_name, "reload_config_file"
+            )
+
+    async def load_config_file(self, compiler):
+        logger.info("loading config file")
+
+        # Read the TOML file
+        with self._config_file.open('rb') as f:
+            toml_data = tomllib.load(f)
+
+        # Handle special case for `magic_smtp_config`
+        magic_smtp_config = toml_data.pop("magic_smtp_config", None)
+        if magic_smtp_config:
+            await self.load_sidechannel_configs(
+                magic_smtp_config, compiler=compiler
+            )
+
+        # Parse TOML config file content into JSON
+        if toml_data and toml_data.get("cfg::Config"):
+            result = compiler.compile_structured_config(
+                toml_data, source="configuration file"
+            )
+            if asyncio.iscoroutine(result):
+                result = await result
+            json_obj = config.to_json_obj(
+                self._server.config_settings,
+                result["cfg::Config"],
+                include_source=False,
+            )
+            config_file_data = [
+                {
+                    "name": name,
+                    "value": value,
+                    "type": config.ConStateType.config_file,
+                }
+                for name, value in json_obj.items()
+            ]
+        else:
+            config_file_data = []
+
+        # Update init_con_data and SQL
+        self.set_init_con_data(
+            [
+                cs
+                for cs in self._init_con_data
+                if cs["type"] != config.ConStateType.config_file
+            ]
+            + config_file_data
+        )
+        return config_file_data
+
+    async def _reload_config_file(self):
+        # Load TOML config file
+        compiler = self._server.get_compiler_pool()
+        config_file_data = await self.load_config_file(compiler)
+        self._update_init_con_sql = b"""
+            DELETE FROM _edgecon_state WHERE "type" = 'F';
+        """.strip() + self._make_init_con_sql(config_file_data)
+
+        # Update sys pgcon and reload system config
+        async with self.use_sys_pgcon() as syscon:
+            if syscon.last_init_con_data is not self._init_con_data:
+                await syscon.sql_execute(self._update_init_con_sql)
+            sys_config = await self._load_sys_config(syscon=syscon)
+            # GOTCHA: don't notify the change of sysconfig because it's local
+        self._dbindex.update_sys_config(sys_config)
+
     def reload(self):
         # In multi-tenant mode, the file paths for the following states may be
         # unset in a reload, while it's impossible in a regular server.
@@ -1577,6 +1733,7 @@ def reload(self):
 
         self.reload_readiness_state()
         self.load_jwcrypto()
+        self.reload_config_file.schedule()
 
         self.start_watching_files()
 
diff --git a/tests/common/test_asyncutil.py b/tests/common/test_asyncutil.py
index dfd99f37b2a..2f9f31afcc4 100644
--- a/tests/common/test_asyncutil.py
+++ b/tests/common/test_asyncutil.py
@@ -101,3 +101,222 @@ async def sleep_until(t):
                 (2020, [15]),
             ],
         )
+
+
+class TestExclusiveTask(unittest.TestCase):
+    async def _test(self, task: asyncutil.ExclusiveTask, get_counter):
+        # double-schedule is effective only once
+        task.schedule()
+        self.assertTrue(task.scheduled)
+        task.schedule()
+        self.assertTrue(task.scheduled)
+        self.assertEqual(get_counter(), 0)
+
+        # an exclusive task is running, schedule another one with a double shot
+        await asyncio.sleep(4)
+        self.assertFalse(task.scheduled)
+        await asyncio.sleep(1)
+        task.schedule()
+        self.assertTrue(task.scheduled)
+        task.schedule()
+        self.assertTrue(task.scheduled)
+        self.assertEqual(get_counter(), 1)
+
+        # first task done, second follows immediately
+        await asyncio.sleep(5)
+        self.assertFalse(task.scheduled)
+        self.assertEqual(get_counter(), 3)
+
+        # all done
+        await asyncio.sleep(9)
+        self.assertFalse(task.scheduled)
+        self.assertEqual(get_counter(), 4)
+
+        # works repeatedly
+        await asyncio.sleep(1)
+        task.schedule()
+        self.assertTrue(task.scheduled)
+        await asyncio.sleep(3)
+        self.assertFalse(task.scheduled)
+        await asyncio.sleep(1)
+        task.schedule()
+        self.assertTrue(task.scheduled)
+        self.assertEqual(get_counter(), 5)
+
+        # now stop the scheduled task and wait for the running one to finish
+        await asyncio.sleep(1)
+        await task.stop()
+        self.assertFalse(task.scheduled)
+        self.assertEqual(get_counter(), 6)
+
+        # no further schedule allowed
+        task.schedule()
+        self.assertFalse(task.scheduled)
+        await asyncio.sleep(10)
+        self.assertEqual(get_counter(), 6)
+
+    @with_fake_event_loop
+    async def test_exclusive_task_01(self):
+        counter = 0
+
+        @asyncutil.exclusive_task
+        async def task():
+            nonlocal counter
+            counter += 1
+            await asyncio.sleep(8)
+            counter += 1
+
+        await self._test(task, lambda: counter)
+
+    @with_fake_event_loop
+    async def test_exclusive_task_02(self):
+        counter = 0
+
+        @asyncutil.exclusive_task()
+        async def task():
+            nonlocal counter
+            counter += 1
+            await asyncio.sleep(8)
+            counter += 1
+
+        await self._test(task, lambda: counter)
+
+    @with_fake_event_loop
+    async def test_exclusive_task_03(self):
+        class MyClass:
+            def __init__(self):
+                self.counter = 0
+
+            @asyncutil.exclusive_task
+            async def task(self):
+                self.counter += 1
+                await asyncio.sleep(8)
+                self.counter += 1
+
+        obj = MyClass()
+        await self._test(obj.task, lambda: obj.counter)
+
+    @with_fake_event_loop
+    async def test_exclusive_task_04(self):
+        class MyClass:
+            def __init__(self):
+                self.counter = 0
+
+            @asyncutil.exclusive_task(slot="another")
+            async def task(self):
+                self.counter += 1
+                await asyncio.sleep(8)
+                self.counter += 1
+
+        obj = MyClass()
+        await self._test(obj.task, lambda: obj.counter)
+
+    @with_fake_event_loop
+    async def test_exclusive_task_05(self):
+        class MyClass:
+            __slots__ = ("counter", "another",)
+
+            def __init__(self):
+                self.counter = 0
+
+            @asyncutil.exclusive_task(slot="another")
+            async def task(self):
+                self.counter += 1
+                await asyncio.sleep(8)
+                self.counter += 1
+
+        obj = MyClass()
+        await self._test(obj.task, lambda: obj.counter)
+
+    @with_fake_event_loop
+    async def test_exclusive_task_06(self):
+        class MyClass:
+            def __init__(self, factor: int):
+                self.counter = 0
+                self.factor = factor
+
+            @asyncutil.exclusive_task
+            async def task(self):
+                self.counter += self.factor
+                await asyncio.sleep(8)
+                self.counter += self.factor
+
+        obj1 = MyClass(1)
+        obj2 = MyClass(2)
+        async with asyncio.TaskGroup() as g:
+            g.create_task(
+                self._test(obj1.task, lambda: obj1.counter // obj1.factor)
+            )
+            await asyncio.sleep(3)
+            g.create_task(
+                self._test(obj2.task, lambda: obj2.counter // obj2.factor)
+            )
+
+    def test_exclusive_task_07(self):
+        with self.assertRaises(TypeError):
+            class MyClass:
+                __slots__ = ()
+
+                @asyncutil.exclusive_task
+                async def task(self):
+                    pass
+
+    def test_exclusive_task_08(self):
+        with self.assertRaises(TypeError):
+            class MyClass:
+                __slots__ = ()
+
+                @asyncutil.exclusive_task(slot="missing")
+                async def task(self):
+                    pass
+
+    def test_exclusive_task_09(self):
+        with self.assertRaises(TypeError):
+            @asyncutil.exclusive_task
+            async def task(*args, **kwargs):
+                pass
+
+    def test_exclusive_task_10(self):
+        with self.assertRaises(TypeError):
+            @asyncutil.exclusive_task
+            async def task(*, p):
+                pass
+
+    def test_exclusive_task_11(self):
+        with self.assertRaises(TypeError):
+            class MyClass:
+                @asyncutil.exclusive_task
+                async def task(self, p):
+                    pass
+
+    def test_exclusive_task_12(self):
+        with self.assertRaises(TypeError):
+            class MyClass:
+                @asyncutil.exclusive_task
+                @classmethod
+                async def task(cls):
+                    pass
+
+    @with_fake_event_loop
+    async def test_exclusive_task_13(self):
+        counter = 0
+
+        class MyClass:
+            @asyncutil.exclusive_task
+            @staticmethod
+            async def task():
+                nonlocal counter
+                counter += 1
+                await asyncio.sleep(8)
+                counter += 1
+
+        obj1 = MyClass()
+        obj2 = MyClass()
+
+        async with asyncio.TaskGroup() as g:
+            g.create_task(
+                self._test(obj1.task, lambda: counter)
+            )
+            g.create_task(
+                self._test(obj2.task, lambda: counter)
+            )


Code review comments:
diff_hunk: @@ -965,11 +1025,26 @@ async def acquire_pgcon(self, dbname: str) -> pgcon.PGConnection:
 
         for _ in range(self._pg_pool.max_capacity):
             conn = await self._pg_pool.acquire(dbname)
-            if conn.is_healthy():
-                return conn
+            if not conn.is_healthy():
+                logger.warning("acquired an unhealthy pgcon; discard now")
+            elif conn.last_init_con_data is not self._init_con_data:
+                if self._update_init_con_sql:
+                    try:
+                        await conn.sql_execute(self._update_init_con_sql)
path: edb/server/tenant.py
text: @user1:
Is there any way to avoid doing this every time?  Ideally we should track whether a connection has the updated state.  Maybe a dict with version counter or something like that?

@author:
`last_init_con_data` would be for "track whether a connection has the updated state"? The `is` reference check should be as fast as a version counter.

@user1:
I'm a bit confused about the interaction between `_update_init_con_sql` and that check then. Can there be a case when the state is not matching but `_update_init_con_sql` is `None`?

@author:
Theoretically no - whenever we update the state, we set a new update SQL. Let me change the code a bit to ensure that it is always done "atomically" with a single function.

In the worst case when it's really happening (someone is hacking the variables manually), it prints a warning, discard that pgcon and acquire a new one, and the new one may just `pg_connect()` with a proper state.