добавить карту бакетов

Алгоритм обновления карты бакетов по событиям топологии

В рамках #42 (closed) реализован сбор топологии (instances/replicasets). Нужно добавить карту бакетов -- обработку Bucket Event в соответствии с ADR pgproto topology notifications.

При этом нужно учитывать, что пикодата пока не отправляет данные нотификации. Необходимо реализовать интерфейс обновления, который позже можно будет переключить на прослушивание пикодаты.

На текущий момент бакеты берутся не из событий напрямую, а через функцию _pico_bucket(tier), после чего строки конвертируются в события -- так же, как для остального topology по #42 (closed).

Пример работы функции:

# (admin) sql> SELECT _pico_bucket('non-exists') AS buckets;
# sbroad: _pico_bucket: tier 'non-exists' doesn't exist
# (admin) sql> SELECT _pico_bucket('default') AS buckets;
# +----------------------------------------------------------------------------------------------------------+
# | buckets                                                                                                  |
# +==========================================================================================================+
# | [{"bucket_id_start": 1, "target_replicaset_uuid": "31f6051e-13aa-43fb-beb2-ae17155ac810", "tier_name":   |
# | "default", "state": "active", "bucket_id_end": 1500, "current_replicaset_uuid":                          |
# | "31f6051e-13aa-43fb-beb2-ae17155ac810"}, {"bucket_id_start": 1501, "target_replicaset_uuid":             |
# | "efb27727-df65-42bd-9ca4-001efea63d33", "tier_name": "default", "state": "active", "bucket_id_end":      |
# | 3000, "current_replicaset_uuid": "efb27727-df65-42bd-9ca4-001efea63d33"}]                                |
# +----------------------------------------------------------------------------------------------------------+
# (1 rows)

Предлагаемая структура

Выбор конкретных структур данных остается за разработчиком на этапе реализации. В данном разделе опишем логические связи внутри топологии:

Topology:
    instances: dict[i_uuid, Instance] = {}     # изменения по instance event
    replicasets: dict[r_uuid, Replicaset] = {} # изменения по replicaset event
    buckets: dict[tier_name, BucketMap] = {}   # изменения по gо bucket event (миграция бакетов
                                               # и по replicaset event (удаление/добавление репликасетов)
BucketMap:
    b_start, b_end, uuid_replicaset

Бакеты нужны в разрезе тира, поэтому карта бакетов всегда привязывается к тиру. Такая структура позволит в будущем по вычисленному bucket id (и тиру) опеределять узел пикодата для вызова (см #8 (closed) )

Кейсы событий

Примечание: на самом деле мы не знаем, как именно пикодата будет формировать нотификации в случае миграций бакетов, изменения состава репликасетов. ADR данный вопрос не затрагивает. По договоренности внутри команды, принимаем риски, что алгоритм обработки, возможно, нужно будет поменять. На текущем же этапе мы будем исходить из следующих предположений:

порядок сформированных событий от пикодата: репликасет -> узел -> бакеты

Данный порядок зафиксирован в ADR, но в случае начального сбора топологии. Мы же будем предполагать, что такой порядок соблюдается и при дальнейшей работе.

Пример: в кластер добавляется новый репликасет. Мы ожидаем сообщения:

replicaset event (new, count = 1)
instance event (new, count >= 1)
bucket event (new, count >= 1)

Данная последовательность гарантирует нам (при корректной работе), что мы узнаем о новых/удаленных репликасетов до получения информации о миграции бакетов. А также то, что вся информация о кластере, полученная из bucket event (uuid репликасета, тир), уже будет на стороне драйвера

При изменениях в распределении бакетов пикодата отправит последовательно нотификации для каждого существующего репликасета в кластере, в котором было изменение

Пример: кластер состоит из репликасетов R1 R2, куда добавляется репликасет R3. Ожидаем сообщения bucket events: R1, R2, R3

Пример: кластер состоит из репликасетов R1 R2 R3, откуда удаляется репликасет R3. Ожидаем сообщения bucket events: R1, R2

Пример: кластер состоит из тиров и репликасетов T1(R1, R2), T2(R3). в T1 добавляется еще один репликасет R4. Мы не ожидаем нотификации по T2(R3), поскольку там не будет изменений. Мы ожидаем нотификации по T1(R1), T1(R2), T1(R4) (если произошли изменения по бакетам).

Это означает, что мы не ожидаем от пикодаты нотификацию об (уже) несуществующих репликасетах (для которых bucket range был бы пуст) -- в предположении, что данная информация избыточна, кроме того, мы не ожидаем сообщения о репликасетах, которых не затронули изменения в кластере

Удаление репликасетов

Вводные:

начальное состояние тира: три репликасета R1(1-1000), R2(1001-2000), R3(2001-3000)
репликасет R2 переходит в expelled -> происходит миграция бакетов
мы не знаем, как именно пикодата будет формировать нотификации о миграции и не знаем, как именно будет "раскидывать" бакеты. поэтому сделаем предположения, что
- 1. бакеты размажутся равномерно между оставшимися репликасетами
- 2. пикодата отправит последовательно нотификации с миграцией бакета для каждого существующего (R1, R2) репликасета в кластере

получаемые события и предполагаемая обработка:

replicaset event: R2 replicaset delete -> убрать R2 из self.replicasets (или пометить статусом, не принципиально) и очистить карту бакетов, то есть новая карта self.buckets.tier_name: R2(), R3(). мы очищаем карту бакетов, поскольку не можем быть уверены, как именно перераспределятся бакеты по оставшимся репликасетам пикодатой. поэтому мы не можем пользоваться "старой" картой бакетов, пока не дождемся и не обработаем нотификации пикодаты о бакетах
bucket event: R1 bucket range 1-1500 => применяем к self.buckets.tier_name, новая карта: R1(1-1500), R3()
bucket event: R3 bucket range 1501-3000 => применяем к self.buckets.tier_name, новая карта: R1(1-1500), R3(1501-3000)

Добавление репликасетов

Вводные:

начальное состояние тира: два репликасета R1(1-1500), R2(1501-3000)
репликасет R3 добавляется в тир -> происходит миграция бакетов
мы не знаем, как именно пикодата будет формировать нотификации о миграции и не знаем, как именно будет "раскидывать" бакеты. поэтому сделаем предположения, что
- 1. бакеты размажутся равномерно между новым набором репликасетами
- 2. пикодата отправит последовательно нотификации с миграцией бакета для каждого существующего (R1, R2, R3) репликасета в кластере

получаемые события и предполагаемая обработка:

replicaset event: R3 replicaset replace (insert) -> добавить R3 в self.replicasets и очистить всю карту бакетов, то есть новая карта self.buckets.tier_name: R1(), R2(), R3(), поскольку мы не можем знать, какие ренжи бакетов "уедут" из стартых репликасетов. поэтому мы не можем пользоваться "старой" картой бакетов, пока не дождемся и не обработаем нотификации пикодаты о бакетах
bucket event: R1 bucket range 1-1000 => применяем к self.buckets.tier_name, новая карта: R1(1-1000), R2(), R3()
bucket event: R2 bucket range 1001-2000 => применяем к self.buckets.tier_name, новая карта: R1(1-1000), R2(1001-2000)
bucket event: R3 bucket range 2001-3000 => применяем к self.buckets.tier_name, новая карта: R1(1-1000), R2(1001-2000), R3(2001-3000)

Изменение весов репликасетов

Вводные:

начальное состояние тира: три репликасета R1(1-1000), R2(1001-2000), R3(2001-3000) с равным весом
репликасеты меняют веса -> происходит миграция бакетов
мы не знаем, как именно пикодата будет формировать нотификации о миграции и не знаем, как именно будет "раскидывать" бакеты. поэтому сделаем предположения для нашего примера, что
- 1. бакеты размажутся между оставшимися репликасетами в пропорции 0.25 0.25 0.5
- 2. пикодата отправит последовательно нотификации с миграцией бакета для каждого существующего (R1, R2, R3) репликасета в кластере

получаемые события и предполагаемая обработка:

replicaset event: отсутствует (пикодата не присылает нотификации об изменении весов)
bucket event: R1 bucket range 1-750 => проверяем пересечение с "старой" картой и находим его (был R1 1 -1000, стал R1 1 - 750) => очищаем карту бакетов (так как не можем знать о корректности остальной карты - по R2 и R3) => применяем пришедший event R1 bucket range 1-750 к карте, таким образом новая карта self.buckets.tier_name: R1(1-750), R2(), R3()
bucket event R2 bucket range 751-1500 => проверяем пересечение с "старой" картой и не находим его => не очищаем карту бакетов => применяем пришедший event R2 bucket range 751-1500 к карте, таким образом новая карта self.buckets.tier_name: R1(1-750), R2(751-1500), R3()
bucket event: R3 bucket range 1501-3000 => проверяем пересечение с "старой" картой и не находим его => не очищаем карту бакетов => применяем пришедший event R3 bucket range 1501-3000 к карте, таким образом новая карта self.buckets.tier_name: R1(1-750), R2(751-1500), R3(1501-3000)

Общий алгоритм

Оба события (replicaset и bucket) независимо приводят к одному и тому же эффекту на карту бакетов -- очистке тира при малейшем подозрении, что старые данные больше не надёжны, плюс upsert для Bucket Event. Кейс 3 показывает, что сам по себе Bucket Event-триггер уже достаточен, чтобы восстановить карту корректно даже без явного Replicaset Event; очистка на Replicaset Event нужна дополнительно, чтобы не было окна, где карта уверенно отдаёт устаревший (а не просто неизвестный) ответ -- между приходом replicaset-события и первого bucket-события по этому тиру.

Обработка `Replicaset Event`

---
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 empty after<br>processing a replicaset event,<br>it will be populated later 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 and store bucket ranges 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

Обработка `Bucket Event`

---
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 buckets [A, B] overlap 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

Edited Jul 21, 2026 by Asya Lomakina

добавить карту бакетов

Алгоритм обновления карты бакетов по событиям топологии

Предлагаемая структура

Кейсы событий

Удаление репликасетов

Добавление репликасетов

Изменение весов репликасетов

Общий алгоритм

Обработка Replicaset Event

Обработка Bucket Event

Обработка `Replicaset Event`

Обработка `Bucket Event`