Sync: 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.synchronous.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.synchronous.topology_tracker.TopologyTracker" "Topology tracker details"

    SOURCE_DETAILS_NOTE>"See topology source details"]:::note -.-> SOURCE
    click SOURCE_DETAILS_NOTE "#picopyn.synchronous.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/synchronous/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[[Topology], Connection | None]) -> None:
        self._connect = connect
        self._conn: Connection | None = None
        # guards the `_conn` reference only, not the queries running on it: close()
        # can be called from a thread that doesn't own the refresh, and must not drop
        # a connection the refresh has just opened
        self._conn_lock = threading.Lock()
        self._cancelled = threading.Event()
        # 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

    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._cancelled.is_set():
            return
        if self._conn is not None:
            try:
                self._conn._ensure_open()
            except Exception:
                self._conn = None

        if self._conn is None:
            with self._conn_lock:
                self._conn = self._connect(topology.snapshot())
            if self._conn is not None:
                topology.clear()
                self._tiers_fetched = False

        if self._conn is None:
            return
        if self._cancelled.is_set():
            return

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

        apply_replicaset_rows(topology, replicaset_rows)
        apply_instance_rows(topology, instance_rows)
        self._refresh_tiers(topology)
        self._refresh_buckets(topology)

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

    def _fetch_instance_rows(self) -> list[InstanceRow]:
        assert self._conn is not None

        cursor = self._conn.cursor()
        try:
            cursor.execute(TOPOLOGY_INSTANCES_QUERY)
            return [
                cast(
                    InstanceRow,
                    {
                        "instance_uuid": instance_uuid,
                        "replicaset_uuid": replicaset_uuid,
                        "current_state": current_state,
                        "address": address,
                    },
                )
                for instance_uuid, replicaset_uuid, current_state, address in cursor.fetchall()
            ]
        finally:
            cursor.close()

    def _fetch_replicaset_rows(self) -> list[ReplicasetRow]:
        assert self._conn is not None

        cursor = self._conn.cursor()
        try:
            cursor.execute(TOPOLOGY_REPLICASETS_QUERY)
            return [
                cast(
                    ReplicasetRow,
                    {
                        "replicaset_uuid": replicaset_uuid,
                        "tier": tier,
                        "current_master_uuid": current_master_uuid,
                    },
                )
                for replicaset_uuid, tier, current_master_uuid in cursor.fetchall()
            ]
        finally:
            cursor.close()

    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 or self._cancelled.is_set():
            return

        if self._tiers_fetched:
            return

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

        topology.apply_tiers({name: bucket_count for name, bucket_count in tier_rows})
        self._tiers_fetched = True

    def _fetch_tier_rows(self) -> list[tuple[str, int]]:
        assert self._conn is not None

        cursor = self._conn.cursor()
        try:
            cursor.execute(TOPOLOGY_TIERS_QUERY)
            return [cast(tuple[str, int], row) for row in cursor.fetchall()]
        finally:
            cursor.close()

    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():
            if self._cancelled.is_set():
                break
            try:
                # a scalar function call -- one row, one JSON-encoded column
                raw_buckets = self._fetch_bucket_row(tier)
            except Exception as e:
                logger.warning("Topology update: failed to fetch buckets for tier %s: %s", tier, e)
                continue

            apply_bucket_rows(topology, tier, raw_buckets)

    def _fetch_bucket_row(self, tier: str) -> object | None:
        assert self._conn is not None

        cursor = self._conn.cursor()
        try:
            cursor.execute(SYNC_TIER_BUCKETS_QUERY, (tier,))
            row = cursor.fetchone()
            return row[0] if row is not None else None
        finally:
            cursor.close()

    def close(self) -> None:
        with self._conn_lock:
            conn, self._conn = self._conn, None

        if conn is not None:
            conn.close()

    def cancel(self) -> None:
        self._cancelled.set()
        conn = self._conn
        if conn is None:
            return

        try:
            conn.cancel()
        except Exception as e:
            logger.warning("Topology update: could not cancel a running refresh: %s", e)

refresh(topology)

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/synchronous/topology_tracker.py

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._cancelled.is_set():
        return
    if self._conn is not None:
        try:
            self._conn._ensure_open()
        except Exception:
            self._conn = None

    if self._conn is None:
        with self._conn_lock:
            self._conn = self._connect(topology.snapshot())
        if self._conn is not None:
            topology.clear()
            self._tiers_fetched = False

    if self._conn is None:
        return
    if self._cancelled.is_set():
        return

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

    apply_replicaset_rows(topology, replicaset_rows)
    apply_instance_rows(topology, instance_rows)
    self._refresh_tiers(topology)
    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/synchronous/topology_tracker.py

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

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

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

        Only safe to call when no `refresh()` is running: it may drop resources
        that a running refresh is using. To interrupt a refresh running in another
        thread use `cancel()` instead.
        """
        ...

    def cancel(self) -> None:  # noqa: B027 - optional hook, not every source can be interrupted
        """Interrupt a `refresh()` running in another thread, if the source can.

        Unlike `close()` it must be safe to call while a refresh is in flight.
        Sources that cannot be interrupted just let the refresh finish, so the
        default is a no-op.
        """

cancel()

Interrupt a refresh() running in another thread, if the source can.

Unlike close() it must be safe to call while a refresh is in flight. Sources that cannot be interrupted just let the refresh finish, so the default is a no-op.

Source code in picopyn/synchronous/topology_tracker.py

def cancel(self) -> None:  # noqa: B027 - optional hook, not every source can be interrupted
    """Interrupt a `refresh()` running in another thread, if the source can.

    Unlike `close()` it must be safe to call while a refresh is in flight.
    Sources that cannot be interrupted just let the refresh finish, so the
    default is a no-op.
    """

close() abstractmethod

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

Only safe to call when no refresh() is running: it may drop resources that a running refresh is using. To interrupt a refresh running in another thread use cancel() instead.

Source code in picopyn/synchronous/topology_tracker.py

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

    Only safe to call when no `refresh()` is running: it may drop resources
    that a running refresh is using. To interrupt a refresh running in another
    thread use `cancel()` instead.
    """
    ...

TopologyTracker

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

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

Refreshes happen:

• on a timer, every interval seconds -- only if interval is not None and start() was called.
• on demand: refresh_now() refreshes immediately, regardless of the timer.
• via trigger(): it 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. Does nothing if the timer loop isn't running.

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

Source code in picopyn/synchronous/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:

    - on a timer, every `interval` seconds -- only if `interval` is not `None`
      and `start()` was called.
    - on demand: `refresh_now()` refreshes immediately, regardless of the timer.
    - via `trigger()`: it 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. Does nothing if the timer loop isn't running.

    All ways share 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 = threading.Lock()
        self._thread: threading.Thread | None = None
        self._triggered = threading.Event()
        self._stop_requested = threading.Event()
        self._on_refresh: Callable[[Topology], None] | None = None

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

        Args:
            on_refresh: Callback, called with `self._topology` after every
                successful refresh. Unlike the async tracker, sync `start()`
                doesn't run an initial refresh inline: doing so would block
                the caller (e.g.
                [`Pool.open()`][picopyn.synchronous.pool.Pool.open]) and could
                deadlock if the callback tried to acquire a lock the caller
                already holds.
        """
        if self._interval is None or self._thread is not None:
            return

        self._on_refresh = on_refresh
        self._stop_requested.clear()
        self._triggered.clear()
        self._thread = threading.Thread(
            target=self._loop,
            name="picopyn-topology-tracker",
            daemon=True,
        )
        self._thread.start()

    def stop(self, timeout: float | None = 5.0) -> None:
        """Stop the topology update loop.

        Args:
            timeout: Maximum number of seconds to wait for the update loop to stop.
                None waits until the loop has finished and closed its source.
        """
        self._stop_requested.set()
        thread = self._thread
        if thread is None:
            # the loop never ran, but refresh_now() may still have opened a connection
            with self._lock:
                self._source.close()
            return

        self._triggered.set()
        if thread is threading.current_thread():
            return

        thread.join(timeout=timeout)
        if thread.is_alive():
            logger.warning(
                "Topology update loop did not stop within %.1f seconds",
                timeout,
            )
            return

        # the loop has already closed the source before the join returned
        self._thread = None

    def terminate(self) -> None:
        """Stop the update loop without waiting for it to finish.

        Unlike [stop()][picopyn.synchronous.topology_tracker.TopologyTracker.stop]
        it never joins the loop thread. A refresh stuck on a query is cancelled instead,
        so it fails right away and the loop winds down on its own, closing the source
        from the thread that owns it. Closing the source here would mean closing a
        connection another thread is running a query on, which the driver doesn't allow.
        """
        thread = self._thread
        self._stop_requested.set()
        self._triggered.set()

        if thread is not None and thread is not threading.current_thread() and thread.is_alive():
            self._source.cancel()
            return

        # A direct refresh_now() may still be using the source without a loop thread.
        if self._lock.acquire(blocking=False):
            try:
                self._source.close()
            finally:
                self._lock.release()
        else:
            self._source.cancel()

    def snapshot(self) -> Topology:
        """Return a consistent detached copy of the current topology view."""
        with self._lock:
            return self._topology.snapshot()

    def get_bucket_count(self, tier: str) -> int | None:
        """Return the known bucket count for a tier."""
        with self._lock:
            return self._topology.tiers.get(tier)

    def has_tiers(self) -> bool:
        """Return whether any tier bucket counts are known."""
        with self._lock:
            return bool(self._topology.tiers)

    def find_bucket_master_uuid(self, tier: str, bucket_id: int) -> str | None:
        """Return the current master instance UUID for a bucket in a tier."""
        with self._lock:
            master = self._topology.find_bucket_master(tier, bucket_id)
            return master.uuid if master is not None else None

    def refresh_now(self) -> None:
        """Refresh topology immediately, regardless of the timer."""
        with self._lock:
            if self._stop_requested.is_set():
                return
            try:
                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._stop_requested.is_set() and self._on_refresh is not None:
                try:
                    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()

    def _loop(self) -> None:
        assert self._interval is not None
        logger.info("Topology update loop started (interval=%.1fs)", self._interval)
        try:
            while not self._stop_requested.is_set():
                self._triggered.wait(timeout=self._interval)
                self._triggered.clear()
                if self._stop_requested.is_set():
                    break
                self.refresh_now()
        finally:
            # A refresh from this loop has finished before control reaches finally,
            # so its source connection is no longer in use and can be closed safely.
            self._source.close()
            logger.debug("Topology update loop stopped")

find_bucket_master_uuid(tier, bucket_id)

Return the current master instance UUID for a bucket in a tier.

Source code in picopyn/synchronous/topology_tracker.py

def find_bucket_master_uuid(self, tier: str, bucket_id: int) -> str | None:
    """Return the current master instance UUID for a bucket in a tier."""
    with self._lock:
        master = self._topology.find_bucket_master(tier, bucket_id)
        return master.uuid if master is not None else None

get_bucket_count(tier)

Return the known bucket count for a tier.

Source code in picopyn/synchronous/topology_tracker.py

def get_bucket_count(self, tier: str) -> int | None:
    """Return the known bucket count for a tier."""
    with self._lock:
        return self._topology.tiers.get(tier)

has_tiers()

Return whether any tier bucket counts are known.

Source code in picopyn/synchronous/topology_tracker.py

def has_tiers(self) -> bool:
    """Return whether any tier bucket counts are known."""
    with self._lock:
        return bool(self._topology.tiers)

refresh_now()

Refresh topology immediately, regardless of the timer.

Source code in picopyn/synchronous/topology_tracker.py

def refresh_now(self) -> None:
    """Refresh topology immediately, regardless of the timer."""
    with self._lock:
        if self._stop_requested.is_set():
            return
        try:
            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._stop_requested.is_set() and self._on_refresh is not None:
            try:
                self._on_refresh(self._topology)
            except Exception as e:
                logger.warning("Topology update: on_refresh callback failed: %s", e)

snapshot()

Return a consistent detached copy of the current topology view.

Source code in picopyn/synchronous/topology_tracker.py

def snapshot(self) -> Topology:
    """Return a consistent detached copy of the current topology view."""
    with self._lock:
        return self._topology.snapshot()

start(on_refresh=None)

Start the tracker.

Parameters:

Name	Type	Description	Default
on_refresh	Callable[[Topology], None] | None	Callback, called with self._topology after every successful refresh. Unlike the async tracker, sync start() doesn't run an initial refresh inline: doing so would block the caller (e.g. Pool.open()) and could deadlock if the callback tried to acquire a lock the caller already holds.	None

Source code in picopyn/synchronous/topology_tracker.py

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

    Args:
        on_refresh: Callback, called with `self._topology` after every
            successful refresh. Unlike the async tracker, sync `start()`
            doesn't run an initial refresh inline: doing so would block
            the caller (e.g.
            [`Pool.open()`][picopyn.synchronous.pool.Pool.open]) and could
            deadlock if the callback tried to acquire a lock the caller
            already holds.
    """
    if self._interval is None or self._thread is not None:
        return

    self._on_refresh = on_refresh
    self._stop_requested.clear()
    self._triggered.clear()
    self._thread = threading.Thread(
        target=self._loop,
        name="picopyn-topology-tracker",
        daemon=True,
    )
    self._thread.start()

stop(timeout=5.0)

Stop the topology update loop.

Parameters:

Name	Type	Description	Default
timeout	float | None	Maximum number of seconds to wait for the update loop to stop. None waits until the loop has finished and closed its source.	5.0

Source code in picopyn/synchronous/topology_tracker.py

def stop(self, timeout: float | None = 5.0) -> None:
    """Stop the topology update loop.

    Args:
        timeout: Maximum number of seconds to wait for the update loop to stop.
            None waits until the loop has finished and closed its source.
    """
    self._stop_requested.set()
    thread = self._thread
    if thread is None:
        # the loop never ran, but refresh_now() may still have opened a connection
        with self._lock:
            self._source.close()
        return

    self._triggered.set()
    if thread is threading.current_thread():
        return

    thread.join(timeout=timeout)
    if thread.is_alive():
        logger.warning(
            "Topology update loop did not stop within %.1f seconds",
            timeout,
        )
        return

    # the loop has already closed the source before the join returned
    self._thread = None

terminate()

Stop the update loop without waiting for it to finish.

Unlike stop() it never joins the loop thread. A refresh stuck on a query is cancelled instead, so it fails right away and the loop winds down on its own, closing the source from the thread that owns it. Closing the source here would mean closing a connection another thread is running a query on, which the driver doesn't allow.

Source code in picopyn/synchronous/topology_tracker.py

def terminate(self) -> None:
    """Stop the update loop without waiting for it to finish.

    Unlike [stop()][picopyn.synchronous.topology_tracker.TopologyTracker.stop]
    it never joins the loop thread. A refresh stuck on a query is cancelled instead,
    so it fails right away and the loop winds down on its own, closing the source
    from the thread that owns it. Closing the source here would mean closing a
    connection another thread is running a query on, which the driver doesn't allow.
    """
    thread = self._thread
    self._stop_requested.set()
    self._triggered.set()

    if thread is not None and thread is not threading.current_thread() and thread.is_alive():
        self._source.cancel()
        return

    # A direct refresh_now() may still be using the source without a loop thread.
    if self._lock.acquire(blocking=False):
        try:
            self._source.close()
        finally:
            self._lock.release()
    else:
        self._source.cancel()

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/synchronous/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/synchronous/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.synchronous.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]], 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/synchronous/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.synchronous.pool.Pool].

    Args:
        connect: Opens a new dedicated
            [`Connection`][picopyn.synchronous.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]], Connection | None],
        cache_size: int,
    ) -> None:
        self._connect = connect
        self._cache = QueryMetadataCache(cache_size)
        self._conn: Connection | None = None
        # background threads (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
        # psycopg Notice callback writes to the cache while `prepare()` holds
        # this lock, so cache operations use a separate lock below
        self._lock = threading.RLock()
        # Cache operations, including resize, must not wait for network I/O
        # under the connection lock.
        self._cache_lock = threading.Lock()
        # one worker is enough: the dedicated connection cannot run overlapping
        # prepares anyway
        self._executor = ThreadPoolExecutor(
            max_workers=1,
            thread_name_prefix="picopyn-query-metadata",
        )
        self._pending_queries: set[str] = set()
        self._pending_queries_lock = threading.Lock()
        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.
        """
        with self._cache_lock:
            self._cache.resize(cache_size)
            with self._pending_queries_lock:
                self._max_pending_queries = cache_size

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

    def _drop_conn(self) -> None:
        """Close the dedicated connection, if one is open. Call it under `_lock`."""
        if self._conn is None:
            return
        try:
            self._conn.close()
        except Exception:
            logger.debug(
                "QueryMetadataService: closing the dedicated connection failed",
                exc_info=True,
            )
        self._conn = None

    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 = self._connect(self._set_metadata)
        if self._terminated:
            self._drop_conn()
            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
            with self._cache_lock:
                self._cache.clear()
        return self._conn is not None

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

    def _prepare(self, query: str, parameters: Params | None = None) -> 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).
        """
        with self._lock:
            if not self._ensure_connected():
                return False
            assert self._conn is not None
            accepted = False
            try:
                prepared_query = self._conn.prepare(query, parameters)
                accepted = True
                if prepared_query != query:
                    with self._cache_lock:
                        metadata = self._cache.get(prepared_query)
                        if metadata is not None:
                            self._cache.set(replace(metadata, query=query))
                            self._cache.evict(prepared_query)
            except Exception:
                logger.warning("QueryMetadataService: prepare failed for %r", query, exc_info=True)
                # TODO: close only on connection-level failures, not query/schema errors.
                self._drop_conn()
            if self._terminated:
                # terminate() could not close the connection while this prepare held
                # the lock, so close it here, from the thread that owns it
                self._drop_conn()
            return accepted

    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.
        """
        with self._cache_lock:
            if self._terminated:
                return None
            metadata = self._cache.get(query)
        if metadata is None:
            self._prepare(query)
            with self._cache_lock:
                metadata = self._cache.get(query)

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

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

        Unlike
        [`get_query_metadata`][picopyn.synchronous.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.synchronous.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"

        Only one background fetch per query can be queued at a time. If too
        many distinct queries are already pending, warming is skipped: the
        caller still executes the query normally, just without cache warming.
        A query Picodata already answered nothing about is not asked about
        again either, until the cached view goes stale.
        """
        with self._cache_lock:
            if self._terminated:
                return None
            metadata = self._cache.get(query)
            known_without_metadata = self._cache.is_known_without_metadata(query)
        if metadata is not None:
            return metadata
        if known_without_metadata:
            return None
        with self._pending_queries_lock:
            if self._terminated:
                return None
            if query in self._pending_queries:
                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)

        try:
            self._executor.submit(self._prepare_in_background, query, parameters)
        except RuntimeError:
            with self._pending_queries_lock:
                self._pending_queries.discard(query)
        return None

    def _prepare_in_background(self, query: str, parameters: Params | None = None) -> None:
        try:
            # the prepare can hang if the node does not respond: `Connection.prepare()`
            # goes through a blocking libpq call, and neither we nor psycopg have a
            # timeout to put on it. A server-side `statement_timeout` would not help
            # either if server just died
            # now only a cancel() from another thread stops this wait
            if self._prepare(query, parameters):
                with self._cache_lock:
                    if not self._terminated and self._cache.get(query) is None:
                        # Picodata accepted the query but sent no metadata Notice back:
                        # more likely it meaningless to ask again (Picodata can be older
                        # than 26.1.1 for example)
                        self._cache.set_no_metadata(query)
        finally:
            with self._pending_queries_lock:
                self._pending_queries.discard(query)

    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]).
        """
        with self._cache_lock:
            self._cache.evict(query)

    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.
        """
        self._executor.shutdown(wait=True)

        with self._lock:
            self._drop_conn()
            self._terminated = True
        with self._cache_lock:
            self._cache.clear()
        with self._pending_queries_lock:
            self._pending_queries.clear()

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

    def terminate(self) -> None:
        """Close the dedicated connection without waiting for background prepares.

        Unlike [close()][picopyn.synchronous.metadata_service.QueryMetadataService.close]
        it never joins the worker thread. A prepare stuck on a query is cancelled
        instead, so it fails right away and `_prepare` closes the connection from the
        thread that owns it. Closing it here would mean closing a connection another
        thread is running a query on, which the driver doesn't allow.
        """
        self._terminated = True
        with self._cache_lock:
            self._cache.clear()
        self._executor.shutdown(wait=False, cancel_futures=True)

        # nobody is inside a prepare, so the connection can be closed right here
        if self._lock.acquire(blocking=False):
            try:
                self._drop_conn()
            finally:
                self._lock.release()
            return

        conn = self._conn
        if conn is not None:
            conn.cancel()

close()

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/synchronous/metadata_service.py

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.
    """
    self._executor.shutdown(wait=True)

    with self._lock:
        self._drop_conn()
        self._terminated = True
    with self._cache_lock:
        self._cache.clear()
    with self._pending_queries_lock:
        self._pending_queries.clear()

evict(query)

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

Source code in picopyn/synchronous/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]).
    """
    with self._cache_lock:
        self._cache.evict(query)

get_query_metadata(query)

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/synchronous/metadata_service.py

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.
    """
    with self._cache_lock:
        if self._terminated:
            return None
        metadata = self._cache.get(query)
    if metadata is None:
        self._prepare(query)
        with self._cache_lock:
            metadata = self._cache.get(query)

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

request_if_missing(query, parameters=None)

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

Only one background fetch per query can be queued at a time. If too many distinct queries are already pending, warming is skipped: the caller still executes the query normally, just without cache warming. A query Picodata already answered nothing about is not asked about again either, until the cached view goes stale.

Source code in picopyn/synchronous/metadata_service.py

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

    Unlike
    [`get_query_metadata`][picopyn.synchronous.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.synchronous.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"

    Only one background fetch per query can be queued at a time. If too
    many distinct queries are already pending, warming is skipped: the
    caller still executes the query normally, just without cache warming.
    A query Picodata already answered nothing about is not asked about
    again either, until the cached view goes stale.
    """
    with self._cache_lock:
        if self._terminated:
            return None
        metadata = self._cache.get(query)
        known_without_metadata = self._cache.is_known_without_metadata(query)
    if metadata is not None:
        return metadata
    if known_without_metadata:
        return None
    with self._pending_queries_lock:
        if self._terminated:
            return None
        if query in self._pending_queries:
            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)

    try:
        self._executor.submit(self._prepare_in_background, query, parameters)
    except RuntimeError:
        with self._pending_queries_lock:
            self._pending_queries.discard(query)
    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/synchronous/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.
    """
    with self._cache_lock:
        self._cache.resize(cache_size)
        with self._pending_queries_lock:
            self._max_pending_queries = cache_size

start()

Eagerly opens the dedicated connection.

Source code in picopyn/synchronous/metadata_service.py

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

stop()

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

Source code in picopyn/synchronous/metadata_service.py

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

terminate()

Close the dedicated connection without waiting for background prepares.

Unlike close() it never joins the worker thread. A prepare stuck on a query is cancelled instead, so it fails right away and _prepare closes the connection from the thread that owns it. Closing it here would mean closing a connection another thread is running a query on, which the driver doesn't allow.

Source code in picopyn/synchronous/metadata_service.py

def terminate(self) -> None:
    """Close the dedicated connection without waiting for background prepares.

    Unlike [close()][picopyn.synchronous.metadata_service.QueryMetadataService.close]
    it never joins the worker thread. A prepare stuck on a query is cancelled
    instead, so it fails right away and `_prepare` closes the connection from the
    thread that owns it. Closing it here would mean closing a connection another
    thread is running a query on, which the driver doesn't allow.
    """
    self._terminated = True
    with self._cache_lock:
        self._cache.clear()
    self._executor.shutdown(wait=False, cancel_futures=True)

    # nobody is inside a prepare, so the connection can be closed right here
    if self._lock.acquire(blocking=False):
        try:
            self._drop_conn()
        finally:
            self._lock.release()
        return

    conn = self._conn
    if conn is not None:
        conn.cancel()