Topology

picopyn.topology

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 topology tracking actually works.

Client-side view of the Picodata cluster topology.

Based on the topology notifications ADR.

Picodata doesn't push topology changes yet as server-side events. So for now we get this data by periodically polling Picodata's system tables and scalar functions with the Topology Tracker instead.

Note

That function itself only exists starting with a 25.5.1 Picodata version, so on older clusters bucket tracking simply stays empty while instances/replicasets keep working.

Once Picodata starts pushing real events, we'll only need to change the source of topology (from Picodata's system tables/functions to NOTIFY messages).

See details in Topology.

Note

Topology itself is read-only -- it only lets callers look at the current cluster state. When topology tracking is enabled, the pool reacts to topology changes in both modes and adjusts its connections to match (see async reconcile and sync reconcile).

Bucket dataclass

A range of buckets owned by a replicaset.

Attributes:

Name	Type	Description
bucket_id_start	int	First bucket id in the range (inclusive). Also the second half of this entry's key in Topology.buckets.
bucket_id_end	int	Last bucket id in the range (inclusive).
current_replicaset_uuid	str	Uuid of the replicaset that currently owns this range.

Source code in picopyn/topology.py

@dataclass(slots=True)
class Bucket:
    """A range of buckets owned by a replicaset.

    Attributes:
        bucket_id_start: First bucket id in the range (inclusive). Also the
            second half of this entry's key in `Topology.buckets`.
        bucket_id_end: Last bucket id in the range (inclusive).
        current_replicaset_uuid: Uuid of the replicaset that currently owns
            this range.
    """

    bucket_id_start: int
    bucket_id_end: int
    current_replicaset_uuid: str

BucketRange dataclass

An inclusive range of bucket ids, e.g. [start, end].

Attributes:

Name	Type	Description
start	int	First bucket id in the range (inclusive).
end	int	Last bucket id in the range (inclusive).

Source code in picopyn/topology.py

@dataclass(slots=True, frozen=True)
class BucketRange:
    """An inclusive range of bucket ids, e.g. `[start, end]`.

    Attributes:
        start: First bucket id in the range (inclusive).
        end: Last bucket id in the range (inclusive).
    """

    start: int
    end: int

EventBuckets dataclass

A bucket range change: reassigned to a different replicaset, etc.

Attributes:

Name	Type	Description
op	Op	Carried for structural parity with the ADR event, but ignored -- there can be only replace value.
tier	str	Tier this bucket range belongs to.
state	str	vshard-reported state of the range, e.g. "active".
bucket_id	BucketRange	Range of bucket ids being (re)assigned.
current_replicaset_uuid	str	Uuid of the replicaset that now owns this range.

Source code in picopyn/topology.py

@dataclass(slots=True)
class EventBuckets:
    """A bucket range change: reassigned to a different replicaset, etc.

    Attributes:
        op: Carried for structural parity with the ADR event, but ignored --
            there can be only `replace` value.
        tier: Tier this bucket range belongs to.
        state: vshard-reported state of the range, e.g. `"active"`.
        bucket_id: Range of bucket ids being (re)assigned.
        current_replicaset_uuid: Uuid of the replicaset that now owns this range.
    """

    op: Op
    tier: str
    state: str
    bucket_id: BucketRange
    current_replicaset_uuid: str

EventInstance dataclass

An instance change: status changed, instance added, role changed, etc.

A few fields are missing here -- Picodata doesn't send this event yet, so there's nothing real to put in them.

Attributes:

Name	Type	Description
op	Op	"replace" to upsert the fields below, "delete" to remove the instance entirely.
instance_uuid	str	Which instance this event is about.
replicaset_uuid	str | None	New replicaset uuid, or None to leave unchanged.
current_state	str | None	New state, or None to leave unchanged.
address	str | None	New pgproto address, or None to leave unchanged.

Source code in picopyn/topology.py

@dataclass(slots=True)
class EventInstance:
    """An instance change: status changed, instance added, role changed, etc.

    A few fields are missing here -- Picodata doesn't send this event yet,
    so there's nothing real to put in them.

    Attributes:
        op: `"replace"` to upsert the fields below, `"delete"` to remove the
            instance entirely.
        instance_uuid: Which instance this event is about.
        replicaset_uuid: New replicaset uuid, or `None` to leave unchanged.
        current_state: New state, or `None` to leave unchanged.
        address: New `pgproto` address, or `None` to leave unchanged.
    """

    op: Op
    instance_uuid: str
    replicaset_uuid: str | None = None
    current_state: str | None = None
    address: str | None = None

EventReplicaset dataclass

A replicaset change: replicaset added, master changed, etc.

A few fields are missing here -- Picodata doesn't send this event yet, so there's nothing real to put in them.

Attributes:

Name	Type	Description
op	Op	"replace" to upsert the fields below, "delete" to remove the replicaset entirely.
replicaset_uuid	str	Which replicaset this event is about.
tier	str | None	New tier, or None to leave unchanged
current_master_uuid	str | None	New replication leader uuid, or None to leave unchanged.

Source code in picopyn/topology.py

@dataclass(slots=True)
class EventReplicaset:
    """A replicaset change: replicaset added, master changed, etc.

    A few fields are missing here -- Picodata doesn't send this event yet,
    so there's nothing real to put in them.

    Attributes:
        op: `"replace"` to upsert the fields below, `"delete"` to remove the
            replicaset entirely.
        replicaset_uuid: Which replicaset this event is about.
        tier: New tier, or `None` to leave unchanged
        current_master_uuid: New replication leader uuid, or `None` to leave
            unchanged.
    """

    op: Op
    replicaset_uuid: str
    tier: str | None = None
    current_master_uuid: str | None = None

Instance dataclass

A Picodata instance, assembled from _pico_instance and _pico_peer_address rows.

Not tier-tagged -- look up replicasets[replicaset_uuid].tier instead.

Attributes:

Name	Type	Description
uuid	str	Instance uuid.
replicaset_uuid	str | None	Uuid of the replicaset this instance belongs to, or None if not known yet.
current_state	str | None	e.g. "Online", or None if not changed.
address	str | None	pgproto address (host:port), or None if not changed.
is_online	bool	Whether current instance state is online (read-only property).

Source code in picopyn/topology.py

@dataclass(slots=True)
class Instance:
    """A Picodata instance, assembled from `_pico_instance` and
    `_pico_peer_address` rows.

    Not tier-tagged -- look up `replicasets[replicaset_uuid].tier` instead.

    Attributes:
        uuid: Instance uuid.
        replicaset_uuid: Uuid of the replicaset this instance belongs to,
            or `None` if not known yet.
        current_state: e.g. `"Online"`, or `None` if not changed.
        address: `pgproto` address (`host:port`), or `None` if not changed.
        is_online: Whether current instance state is online (read-only property).
    """

    uuid: str
    replicaset_uuid: str | None = None
    current_state: str | None = None
    address: str | None = None

    @property
    def is_online(self) -> bool:
        """Whether the instance is in the online state."""
        return self.current_state == ONLINE_STATE

is_online property

Whether the instance is in the online state.

Replicaset dataclass

A Picodata replicaset, assembled from _pico_replicaset rows.

Attributes:

Name	Type	Description
uuid	str	Replicaset uuid.
tier	str | None	Tier this replicaset belongs to, or None if not known yet.
current_master_uuid	str | None	Uuid of the current replication leader instance, or None if not known yet.

Source code in picopyn/topology.py

@dataclass(slots=True)
class Replicaset:
    """A Picodata replicaset, assembled from `_pico_replicaset` rows.

    Attributes:
        uuid: Replicaset uuid.
        tier: Tier this replicaset belongs to, or `None` if not known yet.
        current_master_uuid: Uuid of the current replication leader instance,
            or `None` if not known yet.
    """

    uuid: str
    tier: str | None = None
    current_master_uuid: str | None = None

Topology

Holds the current view of the cluster topology known to the client.

Attributes:

Name	Type	Description
instances	dict[str, Instance]	Known instances, keyed by instance uuid.
replicasets	dict[str, Replicaset]	Known replicasets, keyed by replicaset uuid.
buckets	dict[tuple[str, int], Bucket]	Known bucket ranges, keyed by (tier, bucket_id_start) -- ranges are only meaningful within their tier, so the tier name is part of the key, not just a field on Bucket.
tiers	dict[str, int]	Total bucket count per tier (tier name -> bucket_count). Static cluster configuration, fixed at cluster formation -- fetched once and never re-fetched or cleared.

The topology tracker updates the topology on a schedule or when triggered:

---
title: Topology tracking
---

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

ADR_DETAILS_NOTE>"Based on the Picodata topology<br>notifications ADR"]:::note
click ADR_DETAILS_NOTE "https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-topology-notifications.md" "Picodata ADR"

TOPOLOGY[[Topology<br>·instances<br>·replicasets<br>·buckets<br>·tiers]]

CONNECT["Pool.open()"] --> CFG[/"Topology tracker settings<br>· update interval"/] --> START["start topology tracker"] --> INITIAL["initial refresh<br>(sync, imitate ReadyForQuery)"] --> WAIT
INITAL_DETAILS_NOTE>"Picodata doesn't send it yet,<br>but it's planned: Picodata will<br>send ReadyForQuery once the<br>initial snapshot is fully<br>delivered -- meaning our<br>topology view is consistent"]:::note

INITAL_DETAILS_NOTE -.-> INITIAL

subgraph RefreshLoop["Topology tracker loop"]
    direction TB
    WAIT{"timer elapsed<br>or triggered?"} --> REFRESH["refresh topology"]
end

REFRESH -.->|next iteration| WAIT
REFRESH -.-> TOPOLOGY

QUERY["Pool.execute / fetch / fetchrow"] --> ERR{"connection error?"}
ERR -->|yes| TRIGGER["call tracker trigger"]
ERR -->|no| RAISE(["do not trigger tracker"])
TRIGGER -.-> WAIT

READ["pool.topology"] -.-> TOPOLOGY
NOTE>"read-only view<br>(but can be used to<br>change pool membership)"]:::note
NOTE -.-> READ

Currently, the refresh polls Picodata's system tables/functions:

---
title: Topology refresh
---

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

START(("START")) --> FETCH

TOPOLOGY[/Topology:<br>- instances<br>- replicasets<br>- buckets<br>- tiers/]

FETCH["fetch Picodata system tables"] --> PARSE["parse db rows"]
FETCH <-. read .-> SYS_TABLES[(Picodata system tables)]
PARSE --> DIFF["compare current topology<br>keys with rows"]
TOPOLOGY1[/Topology/] <-. read .-> DIFF
DIFF -.-> EVENTS[["- missing entries become delete events<br>- the rest become replace events<br>"]]
EVENTS -.-> APPLY["topology apply<br>instance/replicaset events"]
TOPOLOGY2[/Topology/] <-. write .-> APPLY

APPLY --> TIERS_FETCHED{"tiers already<br>fetched for this<br>connection?"}
TIERS_FETCHED -->|no| TIER_FETCH["fetch tier names<br>+ bucket counts"] --> TIER_APPLY["topology apply<br>tiers"]
TIER_FETCH <-. read .-> TIER_TABLE[("_pico_tier")]
TIER_APPLY <-. write .-> TOPOLOGY_TIERS[/Topology/]
TIERS_FETCHED -->|yes| COLLECT_TIER
TIER_APPLY --> COLLECT_TIER[["collect tier list<br>from topology"]]

TIER_NOTE>"_pico_tier rows are written once<br>at cluster formation and never<br>rewritten -- fetched at most once<br>per connection, then trusted"]:::note
TIER_NOTE -.-> TIERS_FETCHED

COLLECT_TIER <-. read .-> TOPOLOGY3[/Topology/]
COLLECT_TIER --> PerTier

subgraph PerTier["for each known tier"]
    direction TB
    BUCKET_FETCH["fetch tier's bucket info"] --> BUCKET_PARSE["parse db rows"]
    BUCKET_FETCH <-. read .-> BUCKET_FN[("_pico_bucket(tier)"<br>function)]
    BUCKET_PARSE -.-> BUCKET_EVENTS[["generated bucket events"]]
    BUCKET_EVENTS -.-> BUCKET_APPLY["topology apply<br>bucket events"]
end

BUCKETS_DETAILS_NOTE>"Bucket event details"]:::note
click BUCKETS_DETAILS_NOTE "#picopyn.topology.Topology.apply_bucket" "Bucket event details"
BUCKETS_DETAILS_NOTE -.-> BUCKET_APPLY

NOTE>"_pico_bucket(tier) may not exist<br>on older Picodata -- failure<br>is logged and skipped per tier,<br>not fatal"]:::note
NOTE -.-> BUCKET_FETCH
BUCKET_APPLY <-. write .-> TOPOLOGY4[/Topology/]

BUCKET_APPLY --> STOP(("STOP"))

Source code in picopyn/topology.py

class Topology:
    """
    Holds the current view of the cluster topology known to the client.

    Attributes:
        instances: Known instances, keyed by instance uuid.
        replicasets: Known replicasets, keyed by replicaset uuid.
        buckets: Known bucket ranges, keyed by `(tier, bucket_id_start)` --
            ranges are only meaningful within their tier, so the tier name
            is part of the key, not just a field on `Bucket`.
        tiers: Total bucket count per tier (`tier name -> bucket_count`).
            Static cluster configuration, fixed at cluster formation --
            fetched once and never re-fetched or cleared.

    The topology tracker updates the topology on a schedule or when triggered:
    --8<-- "_topology_tracking.md"

    Currently, the refresh polls Picodata's system tables/functions:

    --8<-- "_topology_refresh.md"
    """

    def __init__(self) -> None:
        self.instances: dict[str, Instance] = {}  # instance_uuid -> Instance
        self.replicasets: dict[str, Replicaset] = {}  # replicaset_uuid -> Replicaset
        self.buckets: dict[tuple[str, int], Bucket] = {}  # (tier, bucket_id_start) -> Bucket
        # sorted range starts per tier, so bucket search can bisect instead of
        # scanning the whole map
        self._tier_range_starts: dict[str, list[int]] = {}
        self.tiers: dict[str, int] = (
            {}
        )  # tier name -> bucket_count (fetched once and never cleared)

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

    def clear(self) -> None:
        """Forget everything we know about the topology.

        Do this before reading a fresh snapshot from a new connection -- the
        new connection may be talking to an instance that's behind or ahead
        of the one we had before.

        Note:
            Does not clear `tiers` -- it's static cluster configuration, not
            tied to which instance we're connected to.
        """
        self.instances.clear()
        self.replicasets.clear()
        self.buckets.clear()
        self._tier_range_starts.clear()

        logger.debug("Topology: cleared")

    def apply_tiers(self, tiers: dict[str, int]) -> None:
        """Apply a `tier name -> bucket_count` snapshot read from the system table.

        Unlike applying instance/replicaset/bucket events, there's no replace/delete
        event semantics for tiers. We need it only because Picodata does not send
        topology Notifications right now. Tiers are static cluster configuration,
        fixed at cluster formation, so once applied, later calls are ignored.

        Note:
            Currently read from system table. Once Picodata starts sending
            Notifications, tier bucket counts can simply be calculated as the sum
            of each tier's replicaset buckets once Picodata sends
            [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.

            (see Picodata's
            [topology notifications ADR](https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-topology-notifications.md?plain=0#изначальное-получение-топологии))
        """
        if self.tiers:
            return
        logger.debug("Topology: got tiers %s", tiers)
        self.tiers = dict(tiers)

    def apply_instance(self, event: EventInstance) -> None:
        """Apply an instance event."""
        logger.debug("Topology: got instance event %s", event)

        if event.op == "delete":
            self.instances.pop(event.instance_uuid, None)
            return

        instance = self.instances.setdefault(
            event.instance_uuid, Instance(uuid=event.instance_uuid)
        )
        if event.replicaset_uuid is not None:
            instance.replicaset_uuid = event.replicaset_uuid
        if event.current_state is not None:
            instance.current_state = event.current_state
        if event.address is not None:
            instance.address = event.address

    def apply_replicaset(self, event: EventReplicaset) -> None:
        """Apply a replicaset event.

        A replicaset joining or leaving a tier changes how that tier's buckets
        are distributed, so the replicaset event affects bucket map.
        """

        logger.debug("Topology: got replicaset event %s", event)

        if event.op == "delete":
            replicaset = self.replicasets.pop(event.replicaset_uuid, None)
            if replicaset is not None and replicaset.tier is not None:
                self._clear_tier_buckets(replicaset.tier)
            return

        existing = self.replicasets.get(event.replicaset_uuid)
        if existing is None:
            tier_already_known = event.tier is not None and any(
                r.tier == event.tier for r in self.replicasets.values()
            )
            replicaset = Replicaset(uuid=event.replicaset_uuid)
            self.replicasets[event.replicaset_uuid] = replicaset
            if event.tier is not None:
                replicaset.tier = event.tier
                if tier_already_known:
                    self._clear_tier_buckets(event.tier)
        else:
            replicaset = existing
            if event.tier is not None:
                replicaset.tier = event.tier

        if event.current_master_uuid is not None:
            replicaset.current_master_uuid = event.current_master_uuid

    def _clear_tier_buckets(self, tier: str) -> None:
        """Drop every known bucket range for a tier -- the old map can no
        longer be trusted once its replicaset membership changes.
        """
        for key in [key for key in self.buckets if key[0] == tier]:
            del self.buckets[key]
        self._tier_range_starts.pop(tier, None)

    def apply_bucket(self, event: EventBuckets) -> None:
        """Apply an `EventBuckets` message by updating the bucket range it describes.

        There are several cases that can trigger bucket migration, for example:

        - a replicaset is added to the cluster
        - a replicaset is removed from the cluster
        - a replicaset's weight changes

        Note:
            Picodata doesn't specify how it will notify about bucket migrations
            or replicaset membership changes -- the
            [topology notifications ADR](https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-topology-notifications.md)
            only covers the initial topology snapshot. We accept the risk that
            this algorithm may need to change once real behavior is observed.
            For now we assume:

            1. Events keep arriving replicaset -> instance -> bucket (the ADR's
               initial-snapshot order) even for later changes, so a
               replicaset's uuid/tier is always known by the time its bucket
               events arrive.
            2. A distribution change resends a bucket event for every
               replicaset affected in that tier -- not for unaffected
               replicasets, and not for replicasets that no longer exist.

        Therefore, replicaset events can also affect the bucket map:
        --8<-- "_topology_apply_bucket_replicaset_event.md"


        --8<-- "_topology_apply_bucket_event.md"

        """
        logger.debug("Topology: got bucket event %s", event)

        start, end = event.bucket_id.start, event.bucket_id.end
        # if intersection found clear bucket map
        # because the map can no longer be trusted
        if self._overlaps_tier_ranges(event.tier, start, end, event.current_replicaset_uuid):
            self._clear_tier_buckets(event.tier)

        # take the list before the insert below, so it cannot already carry `start`
        starts = self._tier_range_starts_for(event.tier)

        key = (event.tier, start)
        is_new_range = key not in self.buckets
        self.buckets[key] = Bucket(
            bucket_id_start=start,
            bucket_id_end=end,
            current_replicaset_uuid=event.current_replicaset_uuid,
        )

        # keep the sorted starts in sync with the map. topology polls re-apply
        # existing bucket events, so only newly seen ranges need to be inserted;
        # re-sorting the whole tier on every refresh would be unnecessary.
        # clear-methods remove the starts together with the corresponding ranges,
        # so the list always stays in sync with `self.buckets`
        if is_new_range:
            insort(starts, start)

    def _tier_range_starts_for(self, tier: str) -> list[int]:
        """The tier's range starts, sorted -- built on first use and kept in step
        by `apply_bucket` from then on.
        """
        starts = self._tier_range_starts.get(tier)
        if starts is None:
            starts = sorted(key[1] for key in self.buckets if key[0] == tier)
            self._tier_range_starts[tier] = starts
        return starts

    def _overlaps_tier_ranges(
        self, tier: str, start: int, end: int, current_replicaset_uuid: str
    ) -> bool:
        """Whether `[start, end]` claims a bucket id some other known range of the
        tier already claims.

        A range redelivered exactly as it is known (same bounds, same replicaset)
        is not an overlap -- currently every poll redelivers the whole tier.
        """
        starts = self._tier_range_starts_for(tier)

        # check only the ranges that can overlap `[start, end]`:
        # - the range just before `start`, if it extends to `start`
        # - all ranges whose starts are within `[start, end]`
        # any earlier range would have to overlap the range immediately before
        # `start` to reach this interval
        index = bisect_left(starts, start)
        if index > 0 and self.buckets[(tier, starts[index - 1])].bucket_id_end >= start:
            return True

        for range_start in starts[index : bisect_right(starts, end)]:
            bucket = self.buckets[(tier, range_start)]
            if (range_start, bucket.bucket_id_end, bucket.current_replicaset_uuid) != (
                start,
                end,
                current_replicaset_uuid,
            ):
                return True

        return False

    def online_addresses(self) -> set[str]:
        """Addresses of instances currently in the `Online` state."""
        return {
            instance.address
            for instance in self.instances.values()
            if instance.address and instance.is_online
        }

    def find_bucket_master(self, tier: str, bucket_id: int) -> Instance | None:
        """Find the current master instance for a bucket in a tier.

        Returns `None` if the topology snapshot is incomplete: the bucket range
        is unknown, its replicaset is unknown, the replicaset has no known
        master yet, or the master instance itself has not been observed.
        """
        # inline the hit path: this runs on every routed query, and the call
        # costs more than the get it wraps
        starts = self._tier_range_starts.get(tier)
        if starts is None:
            starts = self._tier_range_starts_for(tier)

        # ranges within a tier never overlap
        # so the last range starting <= `bucket_id` is the only one that includes it
        index = bisect_right(starts, bucket_id) - 1
        if index < 0:
            return None

        bucket = self.buckets[(tier, starts[index])]
        if bucket_id > bucket.bucket_id_end:
            return None

        replicaset = self.replicasets.get(bucket.current_replicaset_uuid)
        if replicaset is None or replicaset.current_master_uuid is None:
            return None
        return self.instances.get(replicaset.current_master_uuid)

    def known_tiers(self) -> set[str]:
        """Names of all tiers in the cluster.

        Empty until `tiers` has been populated -- and once it is, the
        result never changes, since tiers are fixed for the cluster's
        lifetime.

        Returns:
            The set of tier names, or an empty set if `tiers` hasn't
            been populated yet.
        """
        return set(self.tiers.keys())

    def __repr__(self) -> str:
        return (
            f"Topology(instances={self.instances}, replicasets={self.replicasets}, "
            f"buckets={self.buckets}), tiers = {self.tiers}"
        )

apply_bucket(event)

Apply an EventBuckets message by updating the bucket range it describes.

There are several cases that can trigger bucket migration, for example:

• a replicaset is added to the cluster
• a replicaset is removed from the cluster
• a replicaset's weight changes

Note

Picodata doesn't specify how it will notify about bucket migrations or replicaset membership changes -- the topology notifications ADR only covers the initial topology snapshot. We accept the risk that this algorithm may need to change once real behavior is observed. For now we assume:

1. Events keep arriving replicaset -> instance -> bucket (the ADR's initial-snapshot order) even for later changes, so a replicaset's uuid/tier is always known by the time its bucket events arrive.
2. A distribution change resends a bucket event for every replicaset affected in that tier -- not for unaffected replicasets, and not for replicasets that no longer exist.

Therefore, replicaset events can also affect the bucket map:

---
title: Replicaset Event Processing (Bucket Context)
---
flowchart TD
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242
classDef note fill:#fffde7,stroke:#f9a825,stroke-dasharray:4,color:#555
classDef spacer fill:none,stroke:none

NOTE>"If a tier's bucket map becomes<br>empty after processing a replicaset<br>event, it will be populated later<br>by bucket events"]:::note

NOTE2>"Topology consists of:<br>- replicaset list<br>- instances list<br>- bucket maps<br>Bucket maps are indexed by tier name<br>and store bucket ranges<br>and replicaset UUIDs."]:::note

EVENT[/"Replicaset Event<br>- event type<br>- replicaset uuid<br>- current master uuid"/]:::neutral --> APPLY["apply replicaset event<br>to topology"]
TOPOLOGY_REPL[/"Topology.replicasets"/] -. update .- APPLY

APPLY --> EVENT_TYPE{"event type?"}
EVENT_TYPE --> DelEvent
EVENT_TYPE --> NewEvent
EVENT_TYPE --> ChangeEvent

subgraph DelEvent["delete replicaset"]
    direction TB

    CLEAR_DEL["clear tier bucket map"]
    CLEAR_DEL -. update .- TOPOLOGY_DEL_BUCK[/"Topology.buckets[tier]"/]
end

subgraph NewEvent["new replicaset"]
    direction TB

    TIER_CHECK{"is tier already registered?"}
    TOPOLOGY_NEW_BUCK1[/"Topology.buckets"/]
    TIER_CHECK -. read .- TOPOLOGY_NEW_BUCK1

    TIER_CHECK --> |yes| CLEAR_NEW["clear tier bucket map"]
    TIER_CHECK --> |no| TIER_ADD["create an empty<br>tier bucket map"]

    TOPOLOGY_NEW_BUCK2[/"Topology.buckets[tier]"/]
    TIER_ADD -. update .- TOPOLOGY_NEW_BUCK2
    CLEAR_NEW -. update .- TOPOLOGY_NEW_BUCK2
end

subgraph ChangeEvent["existing replicaset changed<br>(e.g. master switched)"]
    direction TB
    S1[" "]:::spacer --> NOOP["nothing to do"] --> S2[" "]:::spacer
end

DelEvent --> DONE(["replicaset event<br>processed"]):::neutral
NewEvent --> DONE2(["replicaset event<br>processed"]):::neutral
ChangeEvent --> DONE3(["replicaset event<br>processed"]):::neutral

---
title: Bucket Event Processing
---
flowchart TD
classDef neutral fill:#f5f5f5,stroke:#9e9e9e,color:#424242

EVENT[/"Bucket Event<br>- tier T<br>- bucket start A<br>- bucket end B<br>- replicaset uuid X"/]:::neutral

TOPOLOGY_BUCK[/"Topology.buckets[tier=T]<br>- bucket range 1 -> uuid1<br>- ...<br>- bucket range N -> uuidN"/] -. read .- CHECK
EVENT --> CHECK{"Do the received<br>buckets [A, B] overlap<br>existing bucket ranges in tier T?"}

CHECK -->|yes| CLEAR["clear tier T bucket map<br>(the map can no longer be trusted)"]
CLEAR -. update .- TOPOLOGY_EMPTY_BUCK[/"Topology.buckets[tier=T]<br>(empty)"/] --> APPLY
CHECK -->|no| APPLY["upsert the received<br>(bucket range -> replicaset uuid)<br>into tier T bucket map"]

APPLY -. upsert .- TOPOLOGY_NEW_BUCK[/"Topology.buckets[tier=T]<br>- range [A, B] -> uuid X<br>- ...(unchanged mappings)"/]

TOPOLOGY_NEW_BUCK --> DONE(["bucket event<br>processed"]):::neutral

Source code in picopyn/topology.py

def apply_bucket(self, event: EventBuckets) -> None:
    """Apply an `EventBuckets` message by updating the bucket range it describes.

    There are several cases that can trigger bucket migration, for example:

    - a replicaset is added to the cluster
    - a replicaset is removed from the cluster
    - a replicaset's weight changes

    Note:
        Picodata doesn't specify how it will notify about bucket migrations
        or replicaset membership changes -- the
        [topology notifications ADR](https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-topology-notifications.md)
        only covers the initial topology snapshot. We accept the risk that
        this algorithm may need to change once real behavior is observed.
        For now we assume:

        1. Events keep arriving replicaset -> instance -> bucket (the ADR's
           initial-snapshot order) even for later changes, so a
           replicaset's uuid/tier is always known by the time its bucket
           events arrive.
        2. A distribution change resends a bucket event for every
           replicaset affected in that tier -- not for unaffected
           replicasets, and not for replicasets that no longer exist.

    Therefore, replicaset events can also affect the bucket map:
    --8<-- "_topology_apply_bucket_replicaset_event.md"


    --8<-- "_topology_apply_bucket_event.md"

    """
    logger.debug("Topology: got bucket event %s", event)

    start, end = event.bucket_id.start, event.bucket_id.end
    # if intersection found clear bucket map
    # because the map can no longer be trusted
    if self._overlaps_tier_ranges(event.tier, start, end, event.current_replicaset_uuid):
        self._clear_tier_buckets(event.tier)

    # take the list before the insert below, so it cannot already carry `start`
    starts = self._tier_range_starts_for(event.tier)

    key = (event.tier, start)
    is_new_range = key not in self.buckets
    self.buckets[key] = Bucket(
        bucket_id_start=start,
        bucket_id_end=end,
        current_replicaset_uuid=event.current_replicaset_uuid,
    )

    # keep the sorted starts in sync with the map. topology polls re-apply
    # existing bucket events, so only newly seen ranges need to be inserted;
    # re-sorting the whole tier on every refresh would be unnecessary.
    # clear-methods remove the starts together with the corresponding ranges,
    # so the list always stays in sync with `self.buckets`
    if is_new_range:
        insort(starts, start)

apply_instance(event)

Apply an instance event.

Source code in picopyn/topology.py

def apply_instance(self, event: EventInstance) -> None:
    """Apply an instance event."""
    logger.debug("Topology: got instance event %s", event)

    if event.op == "delete":
        self.instances.pop(event.instance_uuid, None)
        return

    instance = self.instances.setdefault(
        event.instance_uuid, Instance(uuid=event.instance_uuid)
    )
    if event.replicaset_uuid is not None:
        instance.replicaset_uuid = event.replicaset_uuid
    if event.current_state is not None:
        instance.current_state = event.current_state
    if event.address is not None:
        instance.address = event.address

apply_replicaset(event)

Apply a replicaset event.

A replicaset joining or leaving a tier changes how that tier's buckets are distributed, so the replicaset event affects bucket map.

Source code in picopyn/topology.py

def apply_replicaset(self, event: EventReplicaset) -> None:
    """Apply a replicaset event.

    A replicaset joining or leaving a tier changes how that tier's buckets
    are distributed, so the replicaset event affects bucket map.
    """

    logger.debug("Topology: got replicaset event %s", event)

    if event.op == "delete":
        replicaset = self.replicasets.pop(event.replicaset_uuid, None)
        if replicaset is not None and replicaset.tier is not None:
            self._clear_tier_buckets(replicaset.tier)
        return

    existing = self.replicasets.get(event.replicaset_uuid)
    if existing is None:
        tier_already_known = event.tier is not None and any(
            r.tier == event.tier for r in self.replicasets.values()
        )
        replicaset = Replicaset(uuid=event.replicaset_uuid)
        self.replicasets[event.replicaset_uuid] = replicaset
        if event.tier is not None:
            replicaset.tier = event.tier
            if tier_already_known:
                self._clear_tier_buckets(event.tier)
    else:
        replicaset = existing
        if event.tier is not None:
            replicaset.tier = event.tier

    if event.current_master_uuid is not None:
        replicaset.current_master_uuid = event.current_master_uuid

apply_tiers(tiers)

Apply a tier name -> bucket_count snapshot read from the system table.

Unlike applying instance/replicaset/bucket events, there's no replace/delete event semantics for tiers. We need it only because Picodata does not send topology Notifications right now. Tiers are static cluster configuration, fixed at cluster formation, so once applied, later calls are ignored.

Note

Currently read from system table. Once Picodata starts sending Notifications, tier bucket counts can simply be calculated as the sum of each tier's replicaset buckets once Picodata sends ReadyForQuery message telling us our Topology view is now consistent.

(see Picodata's topology notifications ADR)

Source code in picopyn/topology.py

def apply_tiers(self, tiers: dict[str, int]) -> None:
    """Apply a `tier name -> bucket_count` snapshot read from the system table.

    Unlike applying instance/replicaset/bucket events, there's no replace/delete
    event semantics for tiers. We need it only because Picodata does not send
    topology Notifications right now. Tiers are static cluster configuration,
    fixed at cluster formation, so once applied, later calls are ignored.

    Note:
        Currently read from system table. Once Picodata starts sending
        Notifications, tier bucket counts can simply be calculated as the sum
        of each tier's replicaset buckets once Picodata sends
        [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.

        (see Picodata's
        [topology notifications ADR](https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-topology-notifications.md?plain=0#изначальное-получение-топологии))
    """
    if self.tiers:
        return
    logger.debug("Topology: got tiers %s", tiers)
    self.tiers = dict(tiers)

clear()

Forget everything we know about the topology.

Do this before reading a fresh snapshot from a new connection -- the new connection may be talking to an instance that's behind or ahead of the one we had before.

Note

Does not clear tiers -- it's static cluster configuration, not tied to which instance we're connected to.

Source code in picopyn/topology.py

def clear(self) -> None:
    """Forget everything we know about the topology.

    Do this before reading a fresh snapshot from a new connection -- the
    new connection may be talking to an instance that's behind or ahead
    of the one we had before.

    Note:
        Does not clear `tiers` -- it's static cluster configuration, not
        tied to which instance we're connected to.
    """
    self.instances.clear()
    self.replicasets.clear()
    self.buckets.clear()
    self._tier_range_starts.clear()

    logger.debug("Topology: cleared")

find_bucket_master(tier, bucket_id)

Find the current master instance for a bucket in a tier.

Returns None if the topology snapshot is incomplete: the bucket range is unknown, its replicaset is unknown, the replicaset has no known master yet, or the master instance itself has not been observed.

Source code in picopyn/topology.py

def find_bucket_master(self, tier: str, bucket_id: int) -> Instance | None:
    """Find the current master instance for a bucket in a tier.

    Returns `None` if the topology snapshot is incomplete: the bucket range
    is unknown, its replicaset is unknown, the replicaset has no known
    master yet, or the master instance itself has not been observed.
    """
    # inline the hit path: this runs on every routed query, and the call
    # costs more than the get it wraps
    starts = self._tier_range_starts.get(tier)
    if starts is None:
        starts = self._tier_range_starts_for(tier)

    # ranges within a tier never overlap
    # so the last range starting <= `bucket_id` is the only one that includes it
    index = bisect_right(starts, bucket_id) - 1
    if index < 0:
        return None

    bucket = self.buckets[(tier, starts[index])]
    if bucket_id > bucket.bucket_id_end:
        return None

    replicaset = self.replicasets.get(bucket.current_replicaset_uuid)
    if replicaset is None or replicaset.current_master_uuid is None:
        return None
    return self.instances.get(replicaset.current_master_uuid)

known_tiers()

Names of all tiers in the cluster.

Empty until tiers has been populated -- and once it is, the result never changes, since tiers are fixed for the cluster's lifetime.

Returns:

Type	Description
set[str]	The set of tier names, or an empty set if tiers hasn't
set[str]	been populated yet.

Source code in picopyn/topology.py

def known_tiers(self) -> set[str]:
    """Names of all tiers in the cluster.

    Empty until `tiers` has been populated -- and once it is, the
    result never changes, since tiers are fixed for the cluster's
    lifetime.

    Returns:
        The set of tier names, or an empty set if `tiers` hasn't
        been populated yet.
    """
    return set(self.tiers.keys())

online_addresses()

Addresses of instances currently in the Online state.

Source code in picopyn/topology.py

def online_addresses(self) -> set[str]:
    """Addresses of instances currently in the `Online` state."""
    return {
        instance.address
        for instance in self.instances.values()
        if instance.address and instance.is_online
    }

snapshot()

Return a detached copy of the current topology view.

Source code in picopyn/topology.py

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