Async: Internal

Warning

Not public API. Driver wires this up and owns it; nothing here is meant to be constructed or called directly. Documented anyway so you can see how it actually works

picopyn.asynchronous.topology_tracker

Keeps a Topology object up to date, using a swappable source.

Three pieces, three narrow jobs:

• Topology: only consumes events and applies them consistently. It doesn't know or care where events come from.
• TopologyTracker: listens to a TopologySource and applies whatever events it gets to Topology. It doesn't know or care how events are produced.
• TopologySource: produces events and hands them to the tracker. Currently PollingTopologySource builds them by polling Picodata's system tables/functions. Later, once Picodata starts pushing topology changes as NOTICE messages (follow the issue), NotifyTopologySource will replace it -- at that point Picodata itself effectively becomes the source.

---
title: Topology components
---

flowchart TB
    classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
    classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

    subgraph SOURCE["TopologySource"]
        direction TB
        POLL["PollingTopologySource<br>(today)"]:::neutral
        NOTIFY["NotifyTopologySource<br>(planned)"]:::neutral
    end

    SOURCE -->|produces events| TRACKER["TopologyTracker<br>listens to a source,<br>applies events it gets"]
    TRACKER -->|apply_instance /<br>apply_replicaset /<br>apply_bucket /<br>apply_tiers| TOPOLOGY[["Topology<br>only job: consume events<br>and apply them consistently"]]

    TOPOLOGY_DETAILS_NOTE>"See topology details"]:::note -.-> TOPOLOGY
    click TOPOLOGY_DETAILS_NOTE "../topology/#picopyn.topology.Topology" "See topology details"

    TRACKER_DETAILS_NOTE>"See topology tracker details"]:::note -.-> TRACKER
    click TRACKER_DETAILS_NOTE "#picopyn.asynchronous.topology_tracker.TopologyTracker" "Topology tracker details"

    SOURCE_DETAILS_NOTE>"See topology source details"]:::note -.-> SOURCE
    click SOURCE_DETAILS_NOTE "#picopyn.asynchronous.topology_tracker.PollingTopologySource" "Topology source details"

    NOTE>"today: events are built by driver<br>polled Picodata's system<br>tables/functions.<br>planned: Picodata itself<br>is the source, pushing<br>events directly"]:::note
    NOTE -.-> SOURCE

Note

The tracker only keeps Topology up to date, so other driver parts (e.g. Pool) can use it for pool membership changes or shard-aware routing. The tracker also accepts a refresh callback that runs right after each refresh -- the pool registers its reconcile there.

PollingTopologySource

Bases: TopologySource

Polls Picodata system tables/functions over one connection, opened on first use and reopened if it dies.

Should be replaced by real NOTIFY messages once Picodata supports them (follow the issue).

Note

Scalar function used to poll bucket ranges, supported by Picodata >=25.5.1

Source code in picopyn/asynchronous/topology_tracker.py

class PollingTopologySource(TopologySource):
    """Polls Picodata system tables/functions over one connection, opened
    on first use and reopened if it dies.

    Should be replaced by real `NOTIFY` messages once Picodata supports them
    (follow [the issue](https://git.picodata.io/core/picodata/-/issues/1306)).

    Note:
        [Scalar function](https://docs.picodata.io/picodata/devel/reference/sql/system_functions/#_pico_bucket)
        used to poll bucket ranges, supported by Picodata >=25.5.1
    """

    def __init__(self, connect: Callable[[], Awaitable[Connection | None]]) -> None:
        self._connect = connect
        self._conn: Connection | None = None
        # tier names and bucket counts are fixed once the cluster is formed,
        # so we only need to fetch them once per connection (reset on reconnect)
        self._tiers_fetched = False

    async def refresh(self, topology: Topology) -> None:
        """Fetch the current topology and apply it, (re)connecting if needed.

        Reconnects first if there's no live connection, clearing `topology`
        in that case -- the new connection may be talking to an instance
        that's behind or ahead of the one we had before. If reconnecting
        fails, returns without touching `topology`.

        Sequence of applying changes:

        - [replicaset][picopyn.topology.EventReplicaset] events
        - [instance][picopyn.topology.EventInstance] events
        - tiers

        Note:
            According to Picodata's
            [topology notifications ADR](https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-topology-notifications.md?plain=0#изначальное-получение-топологии),
            once Picodata has delivered every event of the current snapshot, it sends a
            [ReadyForQuery message](https://www.postgresql.org/docs/current/protocol-message-formats.html#PROTOCOL-MESSAGE-FORMATS-READYFORQUERY),
            telling us our [Topology][picopyn.topology.Topology] view is now consistent.

            This matters because we need to be sure we've received every tier's info
            (via [replicaset events][picopyn.topology.EventReplicaset]) and its buckets
            (via [bucket events][picopyn.topology.EventBuckets]) before trusting the
            snapshot. Picodata doesn't send ReadyForQuery yet, so we have no way of
            knowing when all tiers have shown up in replicaset events.

            Fortunately, we can always get each tier's bucket count from the
            [`_pico_tier`](https://docs.picodata.io/picodata/devel/architecture/system_tables/#_pico_tier)
            system table and trust it as is -- no extra checks needed. It's populated
            with every tier and its bucket count
            [exactly once](https://git.picodata.io/core/picodata/-/blob/master/src/bootstrap_entries.rs?ref_type=heads#L123),
            all rows at once. So all we need to do is read the table once and store it
            in `topology`.

        - [bucket][picopyn.topology.EventBuckets] events (based on
            [the tier list already known][picopyn.topology.Topology.known_tiers] to the topology)

        Note:
            The bucket map is only loosely coupled to the rest of the
            topology (replicasets/instances). Failing to fetch buckets for a
            tier does not mean the rest of the topology is wrong -- the
            topology itself can be known and correct while its buckets are
            temporarily not. This happens e.g. when a replicaset's weight
            changes: replicasets/tiers/instances don't change and are known
            right away, but bucket migration is still in progress, so
            `_pico_bucket(tier)` responds with an error like "rebalance in
            progress" until it settles.

            This is expected to go away once we switch to `NOTIFY`-based
            topology: Picodata will only send us an event once a migration
            has actually completed, so there won't be an in-between state
            to poll into an error -- we'll just apply the bucket map change
            when told.
        """
        if self._conn is None or self._conn.is_closed():
            self._conn = await self._connect()
            if self._conn is not None:
                topology.clear()
                self._tiers_fetched = False

        if self._conn is None:
            return

        try:
            # fetch replicasets and instances info
            # buckets are fetched later when we already know replicasets
            replicaset_rows = await self._conn.fetch(TOPOLOGY_REPLICASETS_QUERY)
            instance_rows = await self._conn.fetch(TOPOLOGY_INSTANCES_QUERY)
        except Exception as e:
            logger.warning("Topology update: failed to fetch topology: %s", e)
            await self._conn.close()
            self._conn = None
            return

        apply_replicaset_rows(topology, cast(Iterable[ReplicasetRow], replicaset_rows))
        apply_instance_rows(topology, cast(Iterable[InstanceRow], instance_rows))
        await self._refresh_tiers(topology)
        await self._refresh_buckets(topology)

        logger.debug("Topology update: %s", topology)

    async def _refresh_tiers(self, topology: Topology) -> None:
        """Fetch tier names and their bucket counts and apply them, but only
        once per connection.

        This cluster configuration is never rewritten, so there's no point
        re-fetching them more than once per connection. If the fetch fails,
        we just try again on the next refresh (no dedicated retry mechanism).
        """
        if self._conn is None:
            return

        if self._tiers_fetched:
            return

        try:
            tier_rows = await self._conn.fetch(TOPOLOGY_TIERS_QUERY)
        except Exception as e:
            logger.warning("Topology update: failed to fetch tiers: %s", e)
            return

        topology.apply_tiers({row["name"]: row["bucket_count"] for row in tier_rows})
        self._tiers_fetched = True

    async def _refresh_buckets(self, topology: Topology) -> None:
        assert self._conn is not None

        # get the known tiers to get their bucket ranges
        # empty means tier info hasn't been fetched yet, so there's nothing to do
        for tier in topology.known_tiers():
            try:
                # a scalar function call -- one row, one JSON-encoded column
                bucket_row = await self._conn.fetchrow(ASYNC_TIER_BUCKETS_QUERY, tier)
            except Exception as e:
                logger.warning("Topology update: failed to fetch buckets for tier %s: %s", tier, e)
                continue

            raw_buckets = bucket_row["buckets"] if bucket_row is not None else None
            apply_bucket_rows(topology, tier, raw_buckets)

    async def close(self) -> None:
        if self._conn is not None:
            await self._conn.close()
            self._conn = None

    def terminate(self) -> None:
        if self._conn is not None:
            self._conn.terminate()
            self._conn = None

refresh(topology) async

Fetch the current topology and apply it, (re)connecting if needed.

Reconnects first if there's no live connection, clearing topology in that case -- the new connection may be talking to an instance that's behind or ahead of the one we had before. If reconnecting fails, returns without touching topology.

Sequence of applying changes:

• replicaset events
• instance events
• tiers

Note

According to Picodata's topology notifications ADR, once Picodata has delivered every event of the current snapshot, it sends a ReadyForQuery message, telling us our Topology view is now consistent.

This matters because we need to be sure we've received every tier's info (via replicaset events) and its buckets (via bucket events) before trusting the snapshot. Picodata doesn't send ReadyForQuery yet, so we have no way of knowing when all tiers have shown up in replicaset events.

Fortunately, we can always get each tier's bucket count from the _pico_tier system table and trust it as is -- no extra checks needed. It's populated with every tier and its bucket count exactly once, all rows at once. So all we need to do is read the table once and store it in topology.

• bucket events (based on the tier list already known to the topology)

Note

The bucket map is only loosely coupled to the rest of the topology (replicasets/instances). Failing to fetch buckets for a tier does not mean the rest of the topology is wrong -- the topology itself can be known and correct while its buckets are temporarily not. This happens e.g. when a replicaset's weight changes: replicasets/tiers/instances don't change and are known right away, but bucket migration is still in progress, so _pico_bucket(tier) responds with an error like "rebalance in progress" until it settles.

This is expected to go away once we switch to NOTIFY-based topology: Picodata will only send us an event once a migration has actually completed, so there won't be an in-between state to poll into an error -- we'll just apply the bucket map change when told.

Source code in picopyn/asynchronous/topology_tracker.py

async def refresh(self, topology: Topology) -> None:
    """Fetch the current topology and apply it, (re)connecting if needed.

    Reconnects first if there's no live connection, clearing `topology`
    in that case -- the new connection may be talking to an instance
    that's behind or ahead of the one we had before. If reconnecting
    fails, returns without touching `topology`.

    Sequence of applying changes:

    - [replicaset][picopyn.topology.EventReplicaset] events
    - [instance][picopyn.topology.EventInstance] events
    - tiers

    Note:
        According to Picodata's
        [topology notifications ADR](https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-topology-notifications.md?plain=0#изначальное-получение-топологии),
        once Picodata has delivered every event of the current snapshot, it sends a
        [ReadyForQuery message](https://www.postgresql.org/docs/current/protocol-message-formats.html#PROTOCOL-MESSAGE-FORMATS-READYFORQUERY),
        telling us our [Topology][picopyn.topology.Topology] view is now consistent.

        This matters because we need to be sure we've received every tier's info
        (via [replicaset events][picopyn.topology.EventReplicaset]) and its buckets
        (via [bucket events][picopyn.topology.EventBuckets]) before trusting the
        snapshot. Picodata doesn't send ReadyForQuery yet, so we have no way of
        knowing when all tiers have shown up in replicaset events.

        Fortunately, we can always get each tier's bucket count from the
        [`_pico_tier`](https://docs.picodata.io/picodata/devel/architecture/system_tables/#_pico_tier)
        system table and trust it as is -- no extra checks needed. It's populated
        with every tier and its bucket count
        [exactly once](https://git.picodata.io/core/picodata/-/blob/master/src/bootstrap_entries.rs?ref_type=heads#L123),
        all rows at once. So all we need to do is read the table once and store it
        in `topology`.

    - [bucket][picopyn.topology.EventBuckets] events (based on
        [the tier list already known][picopyn.topology.Topology.known_tiers] to the topology)

    Note:
        The bucket map is only loosely coupled to the rest of the
        topology (replicasets/instances). Failing to fetch buckets for a
        tier does not mean the rest of the topology is wrong -- the
        topology itself can be known and correct while its buckets are
        temporarily not. This happens e.g. when a replicaset's weight
        changes: replicasets/tiers/instances don't change and are known
        right away, but bucket migration is still in progress, so
        `_pico_bucket(tier)` responds with an error like "rebalance in
        progress" until it settles.

        This is expected to go away once we switch to `NOTIFY`-based
        topology: Picodata will only send us an event once a migration
        has actually completed, so there won't be an in-between state
        to poll into an error -- we'll just apply the bucket map change
        when told.
    """
    if self._conn is None or self._conn.is_closed():
        self._conn = await self._connect()
        if self._conn is not None:
            topology.clear()
            self._tiers_fetched = False

    if self._conn is None:
        return

    try:
        # fetch replicasets and instances info
        # buckets are fetched later when we already know replicasets
        replicaset_rows = await self._conn.fetch(TOPOLOGY_REPLICASETS_QUERY)
        instance_rows = await self._conn.fetch(TOPOLOGY_INSTANCES_QUERY)
    except Exception as e:
        logger.warning("Topology update: failed to fetch topology: %s", e)
        await self._conn.close()
        self._conn = None
        return

    apply_replicaset_rows(topology, cast(Iterable[ReplicasetRow], replicaset_rows))
    apply_instance_rows(topology, cast(Iterable[InstanceRow], instance_rows))
    await self._refresh_tiers(topology)
    await self._refresh_buckets(topology)

    logger.debug("Topology update: %s", topology)

TopologySource

Bases: ABC

Something that can bring a Topology up to date.

Source code in picopyn/asynchronous/topology_tracker.py

class TopologySource(ABC):
    """Something that can bring a `Topology` up to date."""

    @abstractmethod
    async def refresh(self, topology: Topology) -> None: ...

    @abstractmethod
    async def close(self) -> None:
        """Release any resources (e.g. connections) this source holds open."""
        ...

    @abstractmethod
    def terminate(self) -> None:
        """Terminate any resources this source holds open without graceful close."""
        ...

close() abstractmethod async

Release any resources (e.g. connections) this source holds open.

Source code in picopyn/asynchronous/topology_tracker.py

@abstractmethod
async def close(self) -> None:
    """Release any resources (e.g. connections) this source holds open."""
    ...

terminate() abstractmethod

Terminate any resources this source holds open without graceful close.

Source code in picopyn/asynchronous/topology_tracker.py

@abstractmethod
def terminate(self) -> None:
    """Terminate any resources this source holds open without graceful close."""
    ...

TopologyTracker

Holds a topology view and refreshes it through a topology source.

Currently the topology source is Picodata's system tables/functions.

Refreshes happen:

• once, synchronously, inside start() (before the timer loop task is created)
• on a timer, every interval seconds -- only if interval is not None and start() was called.
• on demand: await refresh_now() refreshes immediately, regardless of the timer.
• via the sync trigger(): doesn't refresh by itself, it just wakes the timer loop so it calls refresh_now() right away instead of waiting out the rest of the interval (e.g. the pool does this when a query fails with a connection error). Does nothing if the timer loop isn't running.

All ways funnel through the same lock, so only one refresh runs at a time.

Source code in picopyn/asynchronous/topology_tracker.py

class TopologyTracker:
    """Holds a topology view and refreshes it through a topology source.

    Currently the topology source is Picodata's system tables/functions.

    Refreshes happen:

    - once, synchronously, inside
      [start()][picopyn.asynchronous.topology_tracker.TopologyTracker.start]
      (before the timer loop task is created)
    - on a timer, every `interval` seconds -- only if `interval` is not
      `None` and `start()` was called.
    - on demand:
      [await refresh_now()][picopyn.asynchronous.topology_tracker.TopologyTracker.refresh_now]
      refreshes immediately, regardless of the timer.
    - via the sync
      [trigger()][picopyn.asynchronous.topology_tracker.TopologyTracker.trigger]:
      doesn't refresh by itself, it just wakes the timer loop so it calls
      [refresh_now()][picopyn.asynchronous.topology_tracker.TopologyTracker.refresh_now]
      right away instead of waiting out the rest of the interval (e.g. the
      pool does this when a query fails with a connection error). Does
      nothing if the timer loop isn't running.

    All ways funnel through the same lock, so only one refresh runs at a time.
    """

    def __init__(self, source: TopologySource, interval: float | None) -> None:
        self.topology = Topology()
        self._source = source
        self._interval = interval
        self._lock = asyncio.Lock()
        self._task: asyncio.Task[None] | None = None
        self._refresh_task: asyncio.Task[None] | None = None
        self._stopped = False
        self._triggered = asyncio.Event()
        self._on_refresh: Callable[[Topology], Awaitable[None]] | None = None

    async def start(self, on_refresh: Callable[[Topology], Awaitable[None]] | None = None) -> None:
        """Start the tracker.

        Args:
            on_refresh: Awaitable callback, called with `self.topology` after
                every successful refresh performed by the background loop.
                It is not called after the initial synchronous refresh, since
                that one runs while the caller (e.g.
                [`Pool.open()`][picopyn.asynchronous.pool.Pool.open]) may
                itself be mid-setup -- e.g. holding a lock that the callback
                would deadlock on if invoked immediately.
        """
        if self._interval is not None and self._task is None:
            await self.refresh_now()
            # set the callback only after the initial topology refresh.
            self._on_refresh = on_refresh
            self._task = asyncio.create_task(self._loop())

    async def stop(self) -> None:
        """Stop the tracker."""
        self._stopped = True
        refresh_task = self._refresh_task
        if refresh_task is not None and refresh_task is not asyncio.current_task():
            refresh_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await refresh_task
        if self._task is not None:
            self._task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._task
            self._task = None
        # cancel the loop first so a refresh() in flight can't still be
        # using the source's connection when we close it
        await self._source.close()

    def _on_terminated_task_done(self, task: asyncio.Task[None]) -> None:
        if self._task is task:
            self._task = None
        with contextlib.suppress(asyncio.CancelledError):
            task.exception()

    def terminate(self) -> None:
        """Stop the tracker without waiting for graceful source close."""
        self._stopped = True
        if self._refresh_task is not None and self._refresh_task is not asyncio.current_task():
            self._refresh_task.cancel()
        task = self._task
        if task is not None:
            task.cancel()
            task.add_done_callback(self._on_terminated_task_done)
        self._source.terminate()

    async def refresh_now(self) -> None:
        """Refresh topology immediately, regardless of the timer."""
        async with self._lock:
            if self._stopped:
                return
            refresh_task = asyncio.create_task(self._refresh())
            self._refresh_task = refresh_task
            try:
                await asyncio.shield(refresh_task)
            except asyncio.CancelledError:
                if self._stopped:
                    return
                refresh_task.cancel()
                with contextlib.suppress(asyncio.CancelledError):
                    await refresh_task
                raise
            finally:
                if self._refresh_task is refresh_task:
                    self._refresh_task = None

    async def _refresh(self) -> None:
        try:
            await self._source.refresh(self.topology)
        except Exception as e:
            logger.warning("Topology update failed: %s", e)
            # skip the on_refresh callback if an error occurred here.
            return

        if not self._stopped and self._on_refresh is not None:
            try:
                await self._on_refresh(self.topology)
            except Exception as e:
                logger.warning("Topology update: on_refresh callback failed: %s", e)

    def update_interval(self, interval: float) -> None:
        """Change the refresh interval and wake the background loop.

        A refresh already in progress is allowed to finish.

        Args:
            interval: Number of seconds between topology refreshes.

        Raises:
            ValueError: If `interval` is less than 1s.
            RuntimeError: If periodic topology tracking is disabled.
        """
        if interval < 1.0:
            raise ValueError("topology_update_interval (if set) cannot be less than 1s")
        if self._interval is None:
            raise RuntimeError("Cannot enable topology tracking by changing its interval")
        if self._interval == interval:
            return
        self._interval = interval
        self.trigger()

    def trigger(self) -> None:
        """Wake the timer loop so it refreshes right now, instead of waiting.

        It should be called if some SQL handler (fetch, execute etc.) fails with
        a closed connection.

        Does nothing if the loop isn't running (`interval=None`, or `start()` was
        never called).
        """
        self._triggered.set()

    async def _loop(self) -> None:
        assert self._interval is not None
        logger.info("Topology update loop started (interval=%.1fs)", self._interval)
        try:
            while True:
                # if trigger() and stop()/cancel() land in the same event loop tick,
                # wait_for can swallow the cancellation and this loop never exits
                # (https://github.com/python/cpython/issues/86296, fixed only in 3.12).
                waiter = asyncio.ensure_future(self._triggered.wait())
                try:
                    await asyncio.wait({waiter}, timeout=self._interval)
                finally:
                    if not waiter.done():
                        waiter.cancel()
                        with contextlib.suppress(asyncio.CancelledError):
                            await waiter
                self._triggered.clear()
                await self.refresh_now()
        except asyncio.CancelledError:
            logger.debug("Topology update loop cancelled")
            raise

refresh_now() async

Refresh topology immediately, regardless of the timer.

Source code in picopyn/asynchronous/topology_tracker.py

async def refresh_now(self) -> None:
    """Refresh topology immediately, regardless of the timer."""
    async with self._lock:
        if self._stopped:
            return
        refresh_task = asyncio.create_task(self._refresh())
        self._refresh_task = refresh_task
        try:
            await asyncio.shield(refresh_task)
        except asyncio.CancelledError:
            if self._stopped:
                return
            refresh_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await refresh_task
            raise
        finally:
            if self._refresh_task is refresh_task:
                self._refresh_task = None

start(on_refresh=None) async

Start the tracker.

Parameters:

Name	Type	Description	Default
on_refresh	Callable[[Topology], Awaitable[None]] | None	Awaitable callback, called with self.topology after every successful refresh performed by the background loop. It is not called after the initial synchronous refresh, since that one runs while the caller (e.g. Pool.open()) may itself be mid-setup -- e.g. holding a lock that the callback would deadlock on if invoked immediately.	None

Source code in picopyn/asynchronous/topology_tracker.py

async def start(self, on_refresh: Callable[[Topology], Awaitable[None]] | None = None) -> None:
    """Start the tracker.

    Args:
        on_refresh: Awaitable callback, called with `self.topology` after
            every successful refresh performed by the background loop.
            It is not called after the initial synchronous refresh, since
            that one runs while the caller (e.g.
            [`Pool.open()`][picopyn.asynchronous.pool.Pool.open]) may
            itself be mid-setup -- e.g. holding a lock that the callback
            would deadlock on if invoked immediately.
    """
    if self._interval is not None and self._task is None:
        await self.refresh_now()
        # set the callback only after the initial topology refresh.
        self._on_refresh = on_refresh
        self._task = asyncio.create_task(self._loop())

stop() async

Stop the tracker.

Source code in picopyn/asynchronous/topology_tracker.py

async def stop(self) -> None:
    """Stop the tracker."""
    self._stopped = True
    refresh_task = self._refresh_task
    if refresh_task is not None and refresh_task is not asyncio.current_task():
        refresh_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await refresh_task
    if self._task is not None:
        self._task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._task
        self._task = None
    # cancel the loop first so a refresh() in flight can't still be
    # using the source's connection when we close it
    await self._source.close()

terminate()

Stop the tracker without waiting for graceful source close.

Source code in picopyn/asynchronous/topology_tracker.py

def terminate(self) -> None:
    """Stop the tracker without waiting for graceful source close."""
    self._stopped = True
    if self._refresh_task is not None and self._refresh_task is not asyncio.current_task():
        self._refresh_task.cancel()
    task = self._task
    if task is not None:
        task.cancel()
        task.add_done_callback(self._on_terminated_task_done)
    self._source.terminate()

trigger()

Wake the timer loop so it refreshes right now, instead of waiting.

It should be called if some SQL handler (fetch, execute etc.) fails with a closed connection.

Does nothing if the loop isn't running (interval=None, or start() was never called).

Source code in picopyn/asynchronous/topology_tracker.py

def trigger(self) -> None:
    """Wake the timer loop so it refreshes right now, instead of waiting.

    It should be called if some SQL handler (fetch, execute etc.) fails with
    a closed connection.

    Does nothing if the loop isn't running (`interval=None`, or `start()` was
    never called).
    """
    self._triggered.set()

update_interval(interval)

Change the refresh interval and wake the background loop.

A refresh already in progress is allowed to finish.

Parameters:

Name	Type	Description	Default
interval	float	Number of seconds between topology refreshes.	required

Raises:

Type	Description
ValueError	If interval is less than 1s.
RuntimeError	If periodic topology tracking is disabled.

Source code in picopyn/asynchronous/topology_tracker.py

def update_interval(self, interval: float) -> None:
    """Change the refresh interval and wake the background loop.

    A refresh already in progress is allowed to finish.

    Args:
        interval: Number of seconds between topology refreshes.

    Raises:
        ValueError: If `interval` is less than 1s.
        RuntimeError: If periodic topology tracking is disabled.
    """
    if interval < 1.0:
        raise ValueError("topology_update_interval (if set) cannot be less than 1s")
    if self._interval is None:
        raise RuntimeError("Cannot enable topology tracking by changing its interval")
    if self._interval == interval:
        return
    self._interval = interval
    self.trigger()

picopyn.asynchronous.metadata_service

QueryMetadataService

Dedicated connection for discovering Picodata's query metadata.

Instead of every connection in the pool requesting metadata and passively filling a shared cache, one dedicated connection handles discovery -- separate from the connections that run real queries. Metadata depends only on the query text and the current schema, not on which connection asked/node answered, so a single connection and cache are enough, and ordinary queries never pay for this extra Notice traffic.

So, service owns the single connection and cache used to resolve query meta for a Pool.

Parameters:

Name	Type	Description	Default
connect	Callable[[Callable[[PreparedStatementMetadata], None]], Awaitable[Connection | None]]	Opens a new dedicated Connection with the given callback wired in, or returns None if no configured node could be reached. Called lazily, on first use and again on every reconnect.	required
cache_size	int	Maximum number of distinct queries to keep meta for. Oldest entries are evicted once exceeded. Must be at least 1.	required

Raises:

Type	Description
ValueError	invalid cache_size (see QueryMetadataCache).

Source code in picopyn/asynchronous/metadata_service.py

class QueryMetadataService:
    """Dedicated connection for discovering [Picodata's query metadata][picopyn.query_metadata].

    Instead of every connection in the pool requesting metadata and passively
    filling a shared cache, one dedicated connection handles discovery --
    separate from the connections that run real queries. Metadata depends only
    on the query text and the current schema, not on which connection asked/node
    answered, so a single connection and cache are enough, and ordinary queries
    never pay for this extra Notice traffic.


    So, service owns the single connection and cache used to resolve query meta for
    a [`Pool`][picopyn.asynchronous.pool.Pool].

    Args:
        connect: Opens a new dedicated
            [`Connection`][picopyn.asynchronous.connection.Connection]
            with the given callback wired in, or returns None if no
            configured node could be reached. Called lazily, on first
            use and again on every reconnect.
        cache_size: Maximum number of distinct queries to keep meta for.
            Oldest entries are evicted once exceeded. Must be at least 1.

    Raises:
        ValueError: invalid `cache_size` (see
            [`QueryMetadataCache`][picopyn.query_metadata.QueryMetadataCache.__init__]).
    """

    def __init__(
        self,
        connect: Callable[
            [Callable[[PreparedStatementMetadata], None]], Awaitable[Connection | None]
        ],
        cache_size: int,
    ) -> None:
        self._connect = connect
        self._cache = QueryMetadataCache(cache_size)
        self._conn: Connection | None = None
        # background tasks (see request_if_missing) and direct calls (see
        # get_query_metadata) can race on connect, and the dedicated
        # connection can't run overlapping operations anyway
        self._lock = asyncio.Lock()
        # keeps references to fire-and-forget background tasks (see
        # request_if_missing) so asyncio doesn't garbage-collect them
        # mid-execution
        self._background_tasks: set[asyncio.Task[None]] = set()
        # queries a background prepare is already running for: without this the
        # arrival rate of cache misses can outrun the single dedicated connection
        # and the task queue grows without bound
        self._pending_queries: set[str] = set()
        self._max_pending_queries = cache_size
        # keeps a prepare that is already in flight from reopening the dedicated
        # connection after the service is stopped
        self._terminated = False

    def resize(self, cache_size: int) -> None:
        """Resize the cache and the limit for pending background prepares.

        Already pending prepares are allowed to finish. New ones are admitted
        only when their count is below the new limit.

        Args:
            cache_size: Maximum number of cached and pending queries.
        """
        self._cache.resize(cache_size)
        self._max_pending_queries = cache_size

    async def start(self) -> None:
        """Eagerly opens the dedicated connection."""
        async with self._lock:
            if not await self._ensure_connected():
                logger.warning("QueryMetadataService: no node reachable for dk_meta discovery")

    async def _ensure_connected(self) -> bool:
        if self._terminated:
            return False
        if self._conn is not None and not self._conn.is_closed():
            return True
        self._conn = await self._connect(self._set_metadata)
        if self._terminated:
            if self._conn is not None:
                self._conn.terminate()
                self._conn = None
            return False
        if self._conn is not None:
            # according to the
            # [DDL invalidation rules](https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-query-metadata.md?ref_type=heads#ddl-invalidation).
            # we must clear cached meta when connection established
            self._cache.clear()
        return self._conn is not None

    def _set_metadata(self, metadata: PreparedStatementMetadata) -> None:
        if not self._terminated:
            self._cache.set(metadata)

    async def _prepare(self, query: str) -> bool:
        """(Re)connects if needed, then asks Picodata to Parse `query` so it
        sends back the metadata Notice that fills the cache.

        On failure, the connection is dropped so the next attempt starts
        fresh instead of reusing a possibly broken one.

        Returns:
            True if the Parse was accepted by Picodata, so we have chance to get
            the metadata Notice response. False if no node was reachable or
            Picodata rejected the query (meaning we should not expect the response).
        """
        async with self._lock:
            if not await self._ensure_connected():
                return False
            assert self._conn is not None
            conn = self._conn
            try:
                await conn.prepare(query)
                return True
            except Exception:
                logger.warning("QueryMetadataService: prepare failed for %r", query, exc_info=True)
                # TODO: close only on connection-level failures, not query/schema errors.
                try:
                    await conn.close()
                except Exception:
                    logger.debug(
                        "QueryMetadataService: close after failed prepare also failed",
                        exc_info=True,
                    )
                self._conn = None
                return False

    async def get_query_metadata(self, query: str) -> PreparedStatementMetadata | None:
        """Return Picodata's distribution key metadata for `query`, waiting
        for it if needed.

        Returns the cached metadata right away if present. On a cache miss,
        it asks Picodata for the metadata and waits for the answer -- the
        result is `None` if Picodata never sends metadata for this query
        (e.g. it couldn't resolve a single bucket_id for it), or if no node
        was reachable to ask.
        """
        if self._terminated:
            return None
        metadata = self._cache.get(query)
        if metadata is None:
            await self._prepare(query)
            metadata = None if self._terminated else self._cache.get(query)

        logger.debug("Query metadata for %r: %s", query, metadata)
        return metadata

    def request_if_missing(self, query: str) -> PreparedStatementMetadata | None:
        """Return cached metadata for `query`, warming it in the background if missing.

        Unlike
        [`get_query_metadata`][picopyn.asynchronous.metadata_service.QueryMetadataService.get_query_metadata],
        this returns immediately. On a cache hit it returns the cached metadata.
        On a cache miss it schedules metadata discovery and returns `None`.

        [`Pool.execute`][picopyn.asynchronous.pool.Pool.execute] calls this
        (only for parameterized queries; meta is only ever populated
        for `INSERT`, see
        [`PreparedStatementMetadata`][picopyn.query_metadata.PreparedStatementMetadata])
        so a query with no cached metadata yet still runs right away on an
        ordinary pool connection, while Picodata is asked for the metadata
        in the background on the dedicated connection.

        --8<-- "_request_if_missing.md"

        """
        if self._terminated:
            return None
        metadata = self._cache.get(query)
        if metadata is not None:
            return metadata
        if query in self._pending_queries or self._cache.is_known_without_metadata(query):
            return None
        if len(self._pending_queries) >= self._max_pending_queries:
            logger.debug("QueryMetadataService: too many pending prepares, skipping %r", query)
            return None

        self._pending_queries.add(query)
        task = asyncio.create_task(self._prepare_in_background(query))
        self._background_tasks.add(task)  # link safe task from GC
        task.add_done_callback(self._on_background_task_done)
        return None

    async def _prepare_in_background(self, query: str) -> None:
        try:
            if (
                await self._prepare(query)
                and not self._terminated
                and self._cache.get(query) is None
            ):
                # Picodata accepted the query but sent no metadata Notice back (only ParseComplete):
                # more likely it meaningless to ask again (Picodata can be older than 26.1.1 for example)
                self._cache.set_no_metadata(query)
        finally:
            self._pending_queries.discard(query)

    def _on_background_task_done(self, task: asyncio.Task[None]) -> None:
        self._background_tasks.discard(task)
        with contextlib.suppress(asyncio.CancelledError):
            task.exception()

    def evict(self, query: str) -> None:
        """Drop a single cached entry, e.g. after a stale-plan error (see
        [`is_stmt_invalidated_error`][picopyn.query_metadata.is_stmt_invalidated_error]).

        Also lets background warming ask about the query again, if it will be requested by Pool etc
        """
        self._cache.evict(query)

    async def close(self) -> None:
        """Waits for any background prepares still running, then closes the
        dedicated connection, if one is open.

        Clears the cache. The stopped service no longer accepts metadata requests.
        """
        if self._background_tasks:
            await asyncio.gather(*self._background_tasks, return_exceptions=True)
        async with self._lock:
            if self._conn is not None:
                await self._conn.close()
                self._conn = None
            self._terminated = True
            self._cache.clear()
            self._pending_queries.clear()

    async def stop(self) -> None:
        """Stop the service, cancel pending work and close its dedicated connection."""
        self.terminate()
        await self.close()

    def terminate(self) -> None:
        """Terminate the dedicated connection without waiting for graceful close."""
        self._terminated = True
        self._cache.clear()
        for task in self._background_tasks:
            task.cancel()
        if self._conn is not None:
            self._conn.terminate()
            self._conn = None

close() async

Waits for any background prepares still running, then closes the dedicated connection, if one is open.

Clears the cache. The stopped service no longer accepts metadata requests.

Source code in picopyn/asynchronous/metadata_service.py

async def close(self) -> None:
    """Waits for any background prepares still running, then closes the
    dedicated connection, if one is open.

    Clears the cache. The stopped service no longer accepts metadata requests.
    """
    if self._background_tasks:
        await asyncio.gather(*self._background_tasks, return_exceptions=True)
    async with self._lock:
        if self._conn is not None:
            await self._conn.close()
            self._conn = None
        self._terminated = True
        self._cache.clear()
        self._pending_queries.clear()

evict(query)

Drop a single cached entry, e.g. after a stale-plan error (see is_stmt_invalidated_error).

Also lets background warming ask about the query again, if it will be requested by Pool etc

Source code in picopyn/asynchronous/metadata_service.py

def evict(self, query: str) -> None:
    """Drop a single cached entry, e.g. after a stale-plan error (see
    [`is_stmt_invalidated_error`][picopyn.query_metadata.is_stmt_invalidated_error]).

    Also lets background warming ask about the query again, if it will be requested by Pool etc
    """
    self._cache.evict(query)

get_query_metadata(query) async

Return Picodata's distribution key metadata for query, waiting for it if needed.

Returns the cached metadata right away if present. On a cache miss, it asks Picodata for the metadata and waits for the answer -- the result is None if Picodata never sends metadata for this query (e.g. it couldn't resolve a single bucket_id for it), or if no node was reachable to ask.

Source code in picopyn/asynchronous/metadata_service.py

async def get_query_metadata(self, query: str) -> PreparedStatementMetadata | None:
    """Return Picodata's distribution key metadata for `query`, waiting
    for it if needed.

    Returns the cached metadata right away if present. On a cache miss,
    it asks Picodata for the metadata and waits for the answer -- the
    result is `None` if Picodata never sends metadata for this query
    (e.g. it couldn't resolve a single bucket_id for it), or if no node
    was reachable to ask.
    """
    if self._terminated:
        return None
    metadata = self._cache.get(query)
    if metadata is None:
        await self._prepare(query)
        metadata = None if self._terminated else self._cache.get(query)

    logger.debug("Query metadata for %r: %s", query, metadata)
    return metadata

request_if_missing(query)

Return cached metadata for query, warming it in the background if missing.

Unlike get_query_metadata, this returns immediately. On a cache hit it returns the cached metadata. On a cache miss it schedules metadata discovery and returns None.

Pool.execute calls this (only for parameterized queries; meta is only ever populated for INSERT, see PreparedStatementMetadata) so a query with no cached metadata yet still runs right away on an ordinary pool connection, while Picodata is asked for the metadata in the background on the dedicated connection.

---
title: Background job -- get and cache query meta
---

flowchart TD

classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555

QUERY(["parametrized query"]):::neutral

QUERY --> PARSE_QUERY
PARSE_QUERY["ask Picodata to parse query"] -. Parse .-> PICO[("Picodata")] -. Notice message .-> GET

GET["receive the notice"] --> PARSE_MSG{"is this query metadata?"}

NOTE_ASYNC>"the callback must run<br>synchronously -- metadata has<br>to be ready before the caller's<br>request completes"]:::note -.-> GET

CACHE[/"Pool query meta cache"/] -. write .- FORWARD

PARSE_MSG --> |parsed| FORWARD(["store parsed meta<br>in the cache"]):::neutral

PARSE_MSG --> |could not parse picodata<br>notice message, e.g. Picodata<br>couldn't resolve a distribution<br>key for this query| EXIT(["stop silently"]):::neutral

Source code in picopyn/asynchronous/metadata_service.py

def request_if_missing(self, query: str) -> PreparedStatementMetadata | None:
    """Return cached metadata for `query`, warming it in the background if missing.

    Unlike
    [`get_query_metadata`][picopyn.asynchronous.metadata_service.QueryMetadataService.get_query_metadata],
    this returns immediately. On a cache hit it returns the cached metadata.
    On a cache miss it schedules metadata discovery and returns `None`.

    [`Pool.execute`][picopyn.asynchronous.pool.Pool.execute] calls this
    (only for parameterized queries; meta is only ever populated
    for `INSERT`, see
    [`PreparedStatementMetadata`][picopyn.query_metadata.PreparedStatementMetadata])
    so a query with no cached metadata yet still runs right away on an
    ordinary pool connection, while Picodata is asked for the metadata
    in the background on the dedicated connection.

    --8<-- "_request_if_missing.md"

    """
    if self._terminated:
        return None
    metadata = self._cache.get(query)
    if metadata is not None:
        return metadata
    if query in self._pending_queries or self._cache.is_known_without_metadata(query):
        return None
    if len(self._pending_queries) >= self._max_pending_queries:
        logger.debug("QueryMetadataService: too many pending prepares, skipping %r", query)
        return None

    self._pending_queries.add(query)
    task = asyncio.create_task(self._prepare_in_background(query))
    self._background_tasks.add(task)  # link safe task from GC
    task.add_done_callback(self._on_background_task_done)
    return None

resize(cache_size)

Resize the cache and the limit for pending background prepares.

Already pending prepares are allowed to finish. New ones are admitted only when their count is below the new limit.

Parameters:

Name	Type	Description	Default
cache_size	int	Maximum number of cached and pending queries.	required

Source code in picopyn/asynchronous/metadata_service.py

def resize(self, cache_size: int) -> None:
    """Resize the cache and the limit for pending background prepares.

    Already pending prepares are allowed to finish. New ones are admitted
    only when their count is below the new limit.

    Args:
        cache_size: Maximum number of cached and pending queries.
    """
    self._cache.resize(cache_size)
    self._max_pending_queries = cache_size

start() async

Eagerly opens the dedicated connection.

Source code in picopyn/asynchronous/metadata_service.py

async def start(self) -> None:
    """Eagerly opens the dedicated connection."""
    async with self._lock:
        if not await self._ensure_connected():
            logger.warning("QueryMetadataService: no node reachable for dk_meta discovery")

stop() async

Stop the service, cancel pending work and close its dedicated connection.

Source code in picopyn/asynchronous/metadata_service.py

async def stop(self) -> None:
    """Stop the service, cancel pending work and close its dedicated connection."""
    self.terminate()
    await self.close()

terminate()

Terminate the dedicated connection without waiting for graceful close.

Source code in picopyn/asynchronous/metadata_service.py

def terminate(self) -> None:
    """Terminate the dedicated connection without waiting for graceful close."""
    self._terminated = True
    self._cache.clear()
    for task in self._background_tasks:
        task.cancel()
    if self._conn is not None:
        self._conn.terminate()
        self._conn = None