Topology

picopyn.topology

Client-side view of the Picodata cluster topology.

Based on the topology notifications RFC.

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 with the Topology Tracker instead.

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

See details in Topology.

Note

Currently this is read-only. The connection pool never changes based on topology -- Pool.connect() decides the pool once, for good. Topology just lets callers look at the current cluster state.

EventInstance dataclass

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

For replace, only set the fields that changed — leave the rest as None and they won't overwrite what we already know. For delete, only instance_uuid matters.

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

Source code in picopyn/topology.py

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

    For `replace`, only set the fields that changed — leave the rest as `None`
    and they won't overwrite what we already know. For `delete`, only
    `instance_uuid` matters.

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

    op: Op
    instance_uuid: str
    tier: str | None = None
    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.

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.
    """

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

Instance dataclass

A Picodata instance, assembled from _pico_instance and _pico_peer_address rows.

Source code in picopyn/topology.py

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

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

Replicaset dataclass

A Picodata replicaset, assembled from _pico_replicaset rows.

Source code in picopyn/topology.py

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

    uuid: str
    current_master_uuid: str | None = None

Topology

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

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

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

CONNECT["Pool.connect()"] --> CFG[/"Topology tracker settings<br>· update interval"/] --> START["start topology tracker"] --> WAIT

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 — never<br>changes pool membership"]:::note
NOTE -.-> READ

Currently, the refresh polls Picodata's system tables:

---
title: Topology refresh
---

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

TOPOLOGY[[Topology<br>instances + replicasets]]
SYS_TABLES[(Picodata system tables)]

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

Source code in picopyn/topology.py

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

    The Topology Tracker updates the topology on a schedule or when triggered:
    ```mermaid
    ---
    title: Topology tracking
    ---

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

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

    CONNECT["Pool.connect()"] --> CFG[/"Topology tracker settings<br>· update interval"/] --> START["start topology tracker"] --> WAIT

    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 — never<br>changes pool membership"]:::note
    NOTE -.-> READ
    ```

    Currently, the refresh polls Picodata's system tables:

    ```mermaid
    ---
    title: Topology refresh
    ---

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

    TOPOLOGY[[Topology<br>instances + replicasets]]
    SYS_TABLES[(Picodata system tables)]

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

    def __init__(self) -> None:
        self.instances: dict[str, Instance] = {}
        self.replicasets: dict[str, Replicaset] = {}

    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.
        """
        self.instances.clear()
        self.replicasets.clear()

        logger.debug("Topology: cleared")

    def apply_instance(self, event: EventInstance) -> None:
        """Apply an instance event.

        `replace` only updates the fields that were set — anything left as
        `None` stays as it was. `delete` removes the instance completely.
        """
        logger.debug(f"Topology: got instance event {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.tier is not None:
            instance.tier = event.tier
        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.

        `replace` only updates the fields that were set -- anything left as
        `None` stays as it was. `delete` removes the replicaset completely.
        """

        logger.debug(f"Topology: got replicaset event {event}")

        if event.op == "delete":
            self.replicasets.pop(event.replicaset_uuid, None)
            return

        replicaset = self.replicasets.setdefault(
            event.replicaset_uuid, Replicaset(uuid=event.replicaset_uuid)
        )
        if event.current_master_uuid is not None:
            replicaset.current_master_uuid = event.current_master_uuid

    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.current_state == "Online"
        }

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

apply_instance(event)

Apply an instance event.

replace only updates the fields that were set — anything left as None stays as it was. delete removes the instance completely.

Source code in picopyn/topology.py

def apply_instance(self, event: EventInstance) -> None:
    """Apply an instance event.

    `replace` only updates the fields that were set — anything left as
    `None` stays as it was. `delete` removes the instance completely.
    """
    logger.debug(f"Topology: got instance event {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.tier is not None:
        instance.tier = event.tier
    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.

replace only updates the fields that were set -- anything left as None stays as it was. delete removes the replicaset completely.

Source code in picopyn/topology.py

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

    `replace` only updates the fields that were set -- anything left as
    `None` stays as it was. `delete` removes the replicaset completely.
    """

    logger.debug(f"Topology: got replicaset event {event}")

    if event.op == "delete":
        self.replicasets.pop(event.replicaset_uuid, None)
        return

    replicaset = self.replicasets.setdefault(
        event.replicaset_uuid, Replicaset(uuid=event.replicaset_uuid)
    )
    if event.current_master_uuid is not None:
        replicaset.current_master_uuid = event.current_master_uuid

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.

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.
    """
    self.instances.clear()
    self.replicasets.clear()

    logger.debug("Topology: cleared")

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.current_state == "Online"
    }