OxiDB'de MQTT ve AMQP: PUBACK diskte demek

OxiDB'nin içinde iki mesaj broker'ı var: bir MQTT 3.1.1 broker'ı ve bir AMQP 0-9-1 (RabbitMQ'nun konuştuğu protokol) broker'ı. İkisi de aynı ikili dosyanın içinde, aynı süreçte çalışıyor ve aynı yerden dayanıklılık alıyor: doküman motorunun fsync'li yazma yolu. Bu yazıda bu katmanı baştan sona anlatıyorum: neden bir veritabanının içine broker koyduğumu, MQTT tarafında "QoS 1" sözünü nasıl gerçekten tutar hale getirdiğimi, AMQP'yi neden eklediğimi, bir MQTT sensörünün yayınladığını bir AMQP worker'ının nasıl tükettiğini ve bütün bunların içeride hangi veri yapılarıyla döndüğünü. Her örneği kendi makinemde (Apple M4, macOS, yerel) gerçek bir oxidb-server 0.44.2 sürecine karşı çalıştırdım; istemci tarafında hiçbir şey OxiDB'ye özel değil: mosquitto_pub/mosquitto_sub, paho-mqtt 2.1.0 ve pika 1.4.4, yani hazır, değiştirilmemiş araçlar.

Bir veritabanının içinde neden broker var?

OxiDB'nin MQTT broker'ı ilk olarak bir IoT senaryosu için doğdu: soğuk zincir (ColdChain) demosunda sıcaklık probları ölçümlerini MQTT ile gönderiyor, bu ölçümler bir yerde saklanıp sorgulanıyor. Klasik kurulumda bunun için üç parça gerekir: bir broker (Mosquitto, EMQX), bir köprü servisi ve bir veritabanı. Ben broker'ı veritabanının kenarına koydum: ölçüm geldiği anda doküman motoruna yazılabiliyor, zaman serisi motoruna akıtılabiliyor, OxiMem (Redis uyumlu katman) aboneleri aynı mesajı görebiliyor.

Bir noktada karşıma şu soru çıktı: "Madem bir broker'ımız var, Kafka veya RabbitMQ tarzı ayrı bir mesajlaşma motoru ekleyelim mi?" Bu sorunun cevabı ADR-0015'te (projedeki mimari karar kayıtlarından biri) yazılı ve cevap hayır. Kafka'nın değeri partition başına liderlik ve ISR replikasyonunda; bunu OxiDB'nin tek liderli Raft kümesinin üstüne koymak ya da her mesajı tek bir Raft liderinden geçirmek (partition paralelliğini çöpe atmak) ya da konsensüs modeliyle kavga etmek demek. Sahte bir Kafka ya da kümeyi zorlayan bir yük olacaktı. Onun yerine elimdeki broker'ı dürüst hale getirmeye karar verdim, sonra da onun üstüne AMQP'yi ekledim (ADR-0016).

Bu yazının ana fikri şu: OxiDB'de bir mesajı onayladığım an (MQTT'de PUBACK, AMQP'de publisher confirm) o mesaj diskte, fsync edilmiş halde duruyor. Broker'ın yeni bir dosya formatı yok; oturumlar, kuyruktaki mesajlar, retained değerler sıradan dokümanlar olarak _mqtt ve _amqp koleksiyonlarında yaşıyor.

Açmak: dört ortam değişkeni

Bütün protokol dinleyicileri varsayılan olarak kapalı, açık olmadıklarında maliyetleri de sıfır. Bu yazıdaki denemeler için sunucuyu şöyle başlattım (portlar bana ayrılan aralıktan):

OXIDB_ADDR=127.0.0.1:15800 \
OXIDB_DATA=$PWD/data \
OXIDB_MQTT_PORT=15804 \
OXIDB_AMQP_PORT=15805 \
OXIDB_MQTT_PERSIST=1 \
OXIDB_MQTT_INGEST="sensors/#:sensor_readings,fleet/+/pos:pings" \
oxidb-server

Sunucunun açılış logu, hangi parçaların devrede olduğunu söylüyor (sunucu bu satırlarda uzun tire kullanıyor; kuralım gereği burada kısa çizgiyle yazdım, gerisi birebir):

MQTT listening on 0.0.0.0:15804
MQTT persistence on - recovered 0 sessions, 0 pending messages
MQTT ingest bridge on - 2 route(s)
AMQP listening on 0.0.0.0:15805 - recovered 0 durable queues, 0 messages

Dikkat edilecek bir nokta: MQTT ve AMQP dinleyicileri 0.0.0.0'a bağlanıyor, yani OXIDB_ADDR'deki 127.0.0.1 onları kapsamıyor. Üretimde kimlik doğrulamayı açmak ya da portu güvenlik duvarıyla kapatmak gerekiyor.

MQTT 3.1.1: önce sıradan kısım

Bir broker'dan beklenen temel şeylerle başlayayım: topic'ler, joker karakterli abonelikler ve retained mesajlar. İki odanın sıcaklığını retained olarak yayınlıyorum, sonra home/+/temp filtresiyle abone oluyorum, sonra iki mesaj daha gönderiyorum:

mosquitto_pub -p 15804 -t 'home/livingroom/temp' -m '22.4' -r -q 1
mosquitto_pub -p 15804 -t 'home/kitchen/temp'    -m '24.1' -r -q 1
mosquitto_sub -p 15804 -t 'home/+/temp' -v -q 1 &
mosquitto_pub -p 15804 -t 'home/bedroom/temp'     -m '19.8' -q 1
mosquitto_pub -p 15804 -t 'home/bedroom/humidity' -m '55'   -q 1

home/livingroom/temp 22.4
home/kitchen/temp 24.1
home/bedroom/temp 19.8

İlk iki satır, abone bağlandığı anda gelen retained değerler. Üçüncüsü canlı mesaj. home/bedroom/humidity filtreye uymadığı için gelmiyor. + tam bir seviyeyle, sondaki # ise geri kalan her şeyle (hiçbir şey dahil) eşleşiyor.

Bu kısım her broker'da var. OxiDB'yi farklı kılan kısım, QoS'un ne anlama geldiği.

Dürüst QoS: ADR-0015'in hikayesi

MQTT'de üç teslim seviyesi var. QoS 0 "en fazla bir kez": gönder ve unut. QoS 1 "en az bir kez": broker mesajı, abone PUBACK ile onaylayana kadar saklamak, bağlantı koparsa DUP bayrağıyla yeniden göndermek ve clean_session=false ile bağlanan bir istemci çevrimdışıyken gelen mesajları biriktirip geri döndüğünde teslim etmek zorunda. QoS 2 "tam bir kez": PUBREC/PUBREL/PUBCOMP ile iki aşamalı el sıkışma ve paket kimliğine göre tekrar eleme.

OxiDB'nin ilk MQTT broker'ı bu sözü vermiyordu, ama veriyormuş gibi davranıyordu. ADR-0015'i yazarken kendi kodumu incelediğimde şunu gördüm: teslim döngüsü QoS 1 bayrağını set ediyor, paket kimliği ayırıyor, sitede "QoS 1 delivery with packet ids" yazıyordu. Ama:

• teslim edilen QoS 1 mesajı PUBACK beklenirken hiçbir yerde saklanmıyordu, yani yeniden gönderilemezdi;
• aboneden gelen PUBACK okunup atılıyordu, onaylayacağı bir kayıt yoktu;
• yayın yalnızca o anda canlı kanalı olan abonelere gidiyordu, bir anlığına kopan abone mesajı tamamen kaybediyordu;
• clean_session=false hiçbir şey yapmıyordu, CONNACK her zaman session_present=0 diyordu.

Yani kablodaki protokol "en az bir kez" diyordu, davranış "en fazla bir kez"di. Bunu bir doğruluk hatası olarak sınıflandırdım: sistem tutmadığı bir garantiyi iddia ediyordu. Önce belgeleri gerçeğe uydurdum, sonra davranışı belgelerin eski iddiasına yükselttim. Kural basit: az söz verip fazlasını yapan bir broker kabul edilebilir, tersi hatanın ta kendisi.

Kalıcı oturum: alıcı bağlantıya değil oturuma ait

Çözümün merkezinde küçük bir tasarım dönüşü var. Eski broker'da SUBSCRIBE, bağlantıya bir mpsc::Receiver veriyordu; soket kapanınca alıcı da düşüyor, kanala gelen mesajlar buharlaşıyordu. Yeni tasarımda alıcı oturuma ait. mpsc kanalı, alıcısı hayatta ama okuyanı olmadığında tampona almaya devam ediyor; yani çevrimdışı bir oturuma gelen mesaj kanala düşmeye devam ediyor. Bağlantı takılıyken mesajları bağlantı boşaltıyor, değilken broker genelinde tek bir "reaper" iş parçacığı (500 ms'de bir) onları oturumun sınırlı kuyruğuna aktarıyor.

Oturum, istemci kimliğine göre tutulan bir yapı: abonelik listesi, PUBACK bekleyen giden mesajlar (paket kimliği sıralı bir BTreeMap'te, böylece yeniden gönderim sıralı), çevrimdışıyken gelenlerin kuyruğu, QoS 2 için iki küme ve bir "generation" sayacı. Aynı istemci kimliğiyle ikinci bir CONNECT geldiğinde (MQTT-3.1.4-2: yeni gelen kazanır) generation artıyor; yarışı kaybeden eski bağlantı sayacın değiştiğini görüp çift teslim yapmadan çıkıyor.

Çevrimdışı kuyruk sınırsız değil: oturum başına MAX_QUEUED = 4096. Dolunca en eski mesaj atılıyor ve sayaç (dropped) artıyor. Telemetri için en taze ölçüm en değerli olandır; QoS 1 "en az bir kez" der, "ölü bir tüketiciye karşı sınırsız birikimde hiç kaybetmem" demez. Yavaş bir tüketicinin broker'ı belleksiz bırakmaması gerekiyor.

OXIDB_MQTT_PERSIST: SIGKILL ile deneme

Bellek içi oturumlar, istemci kopmalarına karşı korur ama sunucu yeniden başlarsa her şey sıfırlanır. OXIDB_MQTT_PERSIST=1 açıkken oturumlar, kuyruktaki mesajlar ve retained değerler doküman motorunun WAL'ı üzerinden diske yazılıyor. Burada ADR'deki kuralı ciddiye aldım: "SIGKILL edilmiş bir alt süreç olmadan geçen bir test hiçbir şey kanıtlamaz." O yüzden ben de denemeyi kill -9 ile yaptım.

Senaryo: bir soğuk oda worker'ı plant/+/alarm filtresine QoS 1 ile, kalıcı oturumla abone oluyor ve kopuyor. Bir prob üç alarm yayınlıyor ve her biri için PUBACK bekliyor. Sonra sunucuyu öldürüyorum, yeniden başlatıyorum ve worker'ı tekrar bağlıyorum. Python tarafı paho-mqtt:

import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion

def client(cid, clean):
    return mqtt.Client(CallbackAPIVersion.VERSION2, client_id=cid, clean_session=clean)

# 1) abone ol ve kop
c = client("cold-room-worker", False)
def on_connect(c, u, flags, rc, props):
    print(f"CONNACK rc={rc} session_present={flags.session_present}")
    c.subscribe("plant/+/alarm", qos=1)
def on_subscribe(c, u, mid, rcs, props):
    print(f"SUBACK granted={[r.value for r in rcs]}")
    c.disconnect()

# 2) yayıncı: her mesaj için PUBACK bekle
p = client("probe-17", True)
info = p.publish("plant/room1/alarm", '{"seq": 1, "temp_c": -12.5}', qos=1)
info.wait_for_publish()

Abone olup kopma ve üç yayın:

CONNACK rc=Success session_present=False
SUBACK granted=[1]
PUBACK mid=1 topic=plant/room1/alarm
PUBACK mid=2 topic=plant/room2/alarm
PUBACK mid=3 topic=plant/room1/alarm
PUBACK mid=4 topic=plant/status (retained)

Bu noktada _mqtt koleksiyonuna OxiDB'nin kendi Python istemcisiyle (native wire protokolü) baktım. Broker'ın bütün durumu burada, düz doküman olarak:

{"_kind": "retain", "topic": "home/livingroom/temp", "payload": "22.4", "_id": 1, "_version": 1}
{"_kind": "retain", "topic": "home/kitchen/temp", "payload": "24.1", "_id": 2, "_version": 1}
{"_kind": "sub", "client": "cold-room-worker", "filter": "plant/+/alarm", "qos": 1, "_id": 3, "_version": 1}
{"_kind": "msg", "client": "cold-room-worker", "seq": 1, "topic": "plant/room1/alarm", "payload": "{\"seq\": 1, \"temp_c\": -12.5}", "qos": 1, "_id": 4, "_version": 1}
{"_kind": "msg", "client": "cold-room-worker", "seq": 2, "topic": "plant/room2/alarm", "payload": "{\"seq\": 2, \"temp_c\": -3.1}", "qos": 1, "_id": 5, "_version": 1}
{"_kind": "msg", "client": "cold-room-worker", "seq": 3, "topic": "plant/room1/alarm", "payload": "{\"seq\": 3, \"temp_c\": 4.8}", "qos": 1, "_id": 6, "_version": 1}
{"_kind": "retain", "topic": "plant/status", "payload": "online", "_id": 7, "_version": 1}

Şimdi kill -9, yeniden başlatma ve worker'ın geri dönüşü. Açılış logu:

MQTT persistence on - recovered 1 sessions, 3 pending messages

Worker aynı istemci kimliğiyle, yine clean_session=false ile bağlanıyor; ayrıca temiz oturumlu bir "dashboard" istemcisi plant/status'a abone oluyor:

CONNACK rc=Success session_present=True
PUBLISH topic=plant/room1/alarm qos=1 dup=0 payload={"seq": 1, "temp_c": -12.5}
PUBLISH topic=plant/room2/alarm qos=1 dup=0 payload={"seq": 2, "temp_c": -3.1}
PUBLISH topic=plant/room1/alarm qos=1 dup=0 payload={"seq": 3, "temp_c": 4.8}
PUBLISH topic=plant/status retain=1 payload=online

session_present=True artık bir anlam taşıyor: abonelik, öldürülmüş bir süreçten geri geldi. Üç alarm seq sırasıyla teslim edildi ve retained online değeri de yeniden başlatmadan sağ çıktı. Worker her mesajı PUBACK'leyince _kind: "msg" kayıtları siliniyor; denemeden sonra aynı sorgu boş döndü.

Silme işinde ufak ama önemli bir detay var: PUBACK geldiğinde kayıt içeriğe göre değil (client, seq) çiftine göre siliniyor. İçeriğe göre silmek tekrarları birleştirir: aynı yükü iki kez yayınla, bir kez onayla, içerik eşleşmesi iki kaydı birden siler ve ikinci kopya için "en az bir kez" sessizce bozulur. Bu yüzden bellekteki her inflight mesaj, diskteki kaydının sıra numarasını yanında taşıyor.

Kurtarmanın da kendi sınırı var: bir çökme penceresi diskte MAX_QUEUED'dan fazla kayıt bırakabilir. Açılışta bunların en yenileri yükleniyor, fazlası diskten de siliniyor; aksi halde sınırın attığı mesajlar bir yeniden başlatmayla dirilirdi.

QoS 2: tam bir kez

QoS 2'yi her iki yönde de destekliyorum. mosquitto istemcilerinin -d bayrağıyla paket akışı görünüyor. Bir sayaç meter/42/kwh'ye QoS 2 ile yayın yapıyor, bir faturalama istemcisi meter/+/kwh'ye QoS 2 ile abone:

Client meter-42 sending CONNECT
Client meter-42 received CONNACK (0)
Client meter-42 sending PUBLISH (d0, q2, r0, m1, 'meter/42/kwh', ... (6 bytes))
Client meter-42 received PUBREC (Mid: 1)
Client meter-42 sending PUBREL (m1)
Client meter-42 received PUBCOMP (Mid: 1, RC:0)
Client meter-42 sending DISCONNECT

Client billing sending SUBSCRIBE (Mid: 1, Topic: meter/+/kwh, QoS: 2, Options: 0x00)
Client billing received SUBACK
Subscribed (mid: 1): 2
Client billing received PUBLISH (d0, q2, r0, m1, 'meter/42/kwh', ... (6 bytes))
Client billing sending PUBREC (m1, rc0)
Client billing received PUBREL (Mid: 1)
Client billing sending PUBCOMP (m1)
1532.7

İçeride iki küme bu işi taşıyor. Gelen tarafta rx2: PUBLISH'i dağıttığım ama PUBREL'ini henüz görmediğim paket kimlikleri. Yayıncı aynı kimlikle mesajı yeniden gönderirse (DUP), kimlik kümede olduğu için yalnızca PUBREC'i tekrarlıyorum ve mesajı ikinci kez dağıtmıyorum; "tam bir kez" ile "daha süslü bir el sıkışmalı en az bir kez" arasındaki farkın tamamı bu küme. Giden tarafta pubrel_pending: PUBREC aldığım ve PUBCOMP beklediğim kimlikler. PUBREC anında mesaj kaydını siliyorum (alıcı artık mesajın sahibi) ama PUBREL borcu kalıcı olmak zorunda, yoksa abonenin durum makinesi sonsuza kadar bekleyen bir kimlikte takılır. Kalıcılık açıkken bu iki küme de _kind: "rx2" ve _kind: "rel" dokümanları olarak diske gidiyor; tam bir kez garantisi de çökmeden sağ çıkmalı.

Yazmadan ack yok, peki bedeli ne?

Yayın yolunun sırası mqtt.rs'te açıkça yazılı: önce bellek içi bus'a dağıt, sonra kalıcı oturumlar için fsync'li insert yap, sonra AMQP köprüsüne ve ingest köprüsüne ver, en son PUBACK (QoS 2'de PUBREC) gönder. Yani PUBACK'in anlamı "mesajı aldım" değil, "mesaj ona ihtiyaç duyacak her kalıcı abone için diskte".

Bunun bir bedeli olmalı ve ölçtüm. Aynı makinede (Apple M4, veri dizini dahili SSD'de) paho ile 500 ardışık QoS 1 yayın yaptım, her birinde PUBACK'i bekleyerek. İlk turda topic'e uyan kalıcı abone yoktu, ikinci turda bir tane vardı (çevrimdışı):

bench/nobody: 500 sequential QoS1 publishes, p50=0.059 ms p99=0.185 ms total=0.03 s
bench/durable: 500 sequential QoS1 publishes, p50=4.164 ms p99=9.006 ms total=2.25 s

Fark yaklaşık 70 kat ve tamamı fsync. macOS'ta OxiDB tam diske yazma (F_FULLFSYNC) kullanıyor, bu da birkaç milisaniye demek. Topic'e uyan kalıcı abone yoksa fsync de yok; sunucuda hiç kalıcı oturum yoksa yol tek bir atomik okumayla (has_durable bayrağı) tamamen atlanıyor. Bedeli yalnızca ona ihtiyaç duyan ödüyor. Bu sayılar ardışık, tek bağlantılı en kötü durum; birden fazla yayıncı olduğunda tablo değişiyor, AMQP tarafında anlatacağım grup commit tam da bunun için var.

OXIDB_MQTT_INGEST: topic'ten sorgulanabilir dokümana

İngest köprüsü, MQTT'yi bir veri giriş kapısı olarak kullananlar için yazdığım en pratik parça. OXIDB_MQTT_INGEST virgülle ayrılmış topic-filtresi:koleksiyon kurallarından oluşuyor. Eşleşen her yayın bir dokümana dönüşüp koleksiyona yazılıyor:

• yük bir JSON nesnesiyse, nesnenin kendisi doküman oluyor ({"value": ..} alanı payload.value değil, doğrudan value olarak sorgulanabiliyor);
• başka bir şeyse {"payload": ...} olarak sarılıyor;
• her iki durumda da doküman _topic (yayının yapıldığı gerçek topic; joker kurallarda satırları ayırt etmenin yolu bu) ve _received_at (epoch milisaniye) alanlarını kazanıyor.

Ayrıştırmada son :'den bölüyorum, çünkü bir MQTT topic seviyesi kendisi : içerebilir. Hatalı bir kural sunucunun açılmasını engelliyor: ortada # olan bir filtre (a/#/b) MQTT spesifikasyonuna göre hiçbir şeyle eşleşmez, yani bu bir yazım hatasıdır ve sessizce hiç çalışmayan bir köprüye dönüşmesini istemedim.

Beş yayın yaptım, biri hiçbir kurala uymuyor:

mosquitto_pub -p 15804 -q 1 -t sensors/greenhouse/temp     -m '{"value": 27.3, "unit": "C"}'
mosquitto_pub -p 15804 -q 1 -t sensors/greenhouse/humidity -m '{"value": 71, "unit": "%"}'
mosquitto_pub -p 15804 -q 1 -t sensors/garage/temp         -m '18.9'
mosquitto_pub -p 15804 -q 1 -t fleet/u42/pos -m '{"lat": 41.0151, "lon": 28.9795, "speed": 34}'
mosquitto_pub -p 15804 -q 1 -t chat/general  -m '{"msg": "not mapped"}'

Sonra native istemciyle koleksiyonlara baktım:

with OxiDbClient(port=15800) as db:
    print("collections:", sorted(...db.list_collections()))
    for d in db.find("sensor_readings", {}):
        print(json.dumps(d, ensure_ascii=False))
    for d in db.find("sensor_readings", {"value": {"$gte": 25}}):
        print(d["_topic"], d["value"], d["unit"])
    for d in db.find("pings", {}):
        print(json.dumps(d))

collections: ['_alerts', '_amqp', '_mqtt', '_schedules', 'pings', 'sensor_readings']
{"value": 27.3, "unit": "C", "_topic": "sensors/greenhouse/temp", "_received_at": 1791754335817, "_id": 1, "_version": 1}
{"value": 71, "unit": "%", "_topic": "sensors/greenhouse/humidity", "_received_at": 1791754335830, "_id": 2, "_version": 1}
{"payload": 18.9, "_topic": "sensors/garage/temp", "_received_at": 1791754335839, "_id": 3, "_version": 1}
--- find value >= 25
sensors/greenhouse/temp 27.3 C
sensors/greenhouse/humidity 71 %
--- pings
{"lat": 41.0151, "lon": 28.9795, "speed": 34, "_topic": "fleet/u42/pos", "_received_at": 1791754335848, "_id": 1, "_version": 1}

chat/general hiçbir yere yazılmadı. 18.9 gibi düz bir sayı payload altına sarıldı. Gerisi artık normal birer doküman: index kurabilir, aggregation çalıştırabilir, doküman motorunun bütün sorgu dilini kullanabilirsiniz. Koordinatları olan pings koleksiyonu da coğrafi sorgulara hazır.

Insert, broker'ın QoS ack'inden önce yapılıyor; yani onaylanmış bir yayın koleksiyonda da kalıcı. Köprü yapılandırılmamışsa her yayının bedeli tek bir atomik okuma (OnceLock::get).

AMQP 0-9-1: MQTT'nin ifade edemediği şey

ADR-0015'te RabbitMQ tarzı kalıcı bir kuyruk motorunu reddetmemiş, ertelemiştim: "kalıcı oturumlar, inflight deposu ve WAL destekli yeniden teslim var olduğunda, kalıcı bir kuyruk yüzeyi ayrı bir motor değil, bunun üstüne küçük bir artış olur." Sonra soru protokol biçiminde geldi: mevcut RabbitMQ istemcileri OxiDB ile konuşabilir mi?

AMQP, MQTT'nin bir uzantısı değil; tek bir gerçekten yeni anlamı olan ikinci bir protokol. O anlam: rakip tüketiciler. Bir MQTT aboneliği her mesajın bir kopyasını eşleşen her aboneye verir. Bir AMQP kuyruğu ise her mesajı tüketicilerinden yalnızca birine verir. İş dağıtımı budur, RabbitMQ'nun üreticilerle worker havuzları arasında durmasının sebebi de bu. OxiDB'de pub/sub vardı (MQTT, OxiMem), iş kuyruğu yoktu.

Kapsamı bilerek dar tuttum ve sınırı dürüstçe çizdim. Desteklenenler: bağlantı el sıkışması (PLAIN), kanallar, heartbeat, varsayılan exchange ("", kuyruk adına göre yönlendirir), direct, fanout ve topic exchange'leri, Queue.Declare (durable, exclusive, auto-delete, sunucu adlı kuyruklar, passive), Queue.Bind, çok frame'li gövdeler, Basic.Consume/Deliver/Get/Cancel, Basic.Ack (multiple dahil), Basic.Nack/Reject, Basic.Qos prefetch, mandatory bayrağı ve Basic.Return, publisher confirm. Kapsam dışı: tx sınıfı, headers exchange, exchange'ten exchange'e bağlama, mesaj başına TTL ve öncelik, dead-letter argümanları, alternate exchange, AMQP 1.0 (bambaşka bir protokol), stream'ler ve yönetim arayüzü.

Exchange'ler: direct, fanout, topic

Hepsini tek bir pika betiğiyle denedim:

import pika
conn = pika.BlockingConnection(pika.ConnectionParameters("127.0.0.1", 15805))
ch = conn.channel()

ch.exchange_declare("orders", exchange_type="direct")
for q, key in [("q.paid", "paid"), ("q.refund", "refunded")]:
    ch.queue_declare(q); ch.queue_bind(q, "orders", routing_key=key)
for key, body in [("paid", "order-1001"), ("refunded", "order-0999"),
                  ("paid", "order-1002"), ("cancelled", "order-1003")]:
    ch.basic_publish("orders", key, body)

ch.exchange_declare("cache-bust", exchange_type="fanout")
for q in ("node-a", "node-b", "node-c"):
    ch.queue_declare(q); ch.queue_bind(q, "cache-bust")
ch.basic_publish("cache-bust", "ignored", "invalidate:/products/42")

ch.exchange_declare("logs", exchange_type="topic")
binds = {"all-errors": "*.error", "eu-everything": "eu.#", "eu-west-db": "eu.west.db.*"}
for q, pat in binds.items():
    ch.queue_declare(q); ch.queue_bind(q, "logs", routing_key=pat)
for key in ["api.error", "eu.west.db.error", "eu.west.db.slow", "eu", "us.east.api.info"]:
    ch.basic_publish("logs", key, key)

Kuyrukları basic_get ile boşalttığımda:

server product: OxiDB

== direct exchange
q.paid   -> ['order-1001', 'order-1002']
q.refund -> ['order-0999']

== fanout exchange
node-a -> ['invalidate:/products/42']
node-b -> ['invalidate:/products/42']
node-c -> ['invalidate:/products/42']

== topic exchange
all-errors     *.error        -> ['api.error']
eu-everything  eu.#           -> ['eu.west.db.error', 'eu.west.db.slow', 'eu']
eu-west-db     eu.west.db.*   -> ['eu.west.db.error', 'eu.west.db.slow']

cancelled anahtarlı sipariş hiçbir bağlamaya uymadığı için düştü (mandatory olmadan bu AMQP'de normal davranış). Topic exchange'de * tam bir kelimeyle, # sıfır ya da daha fazla kelimeyle eşleşiyor: eu.# tek başına eu anahtarını da yakaladı, *.error ise iki kelimeli olmayan eu.west.db.error'ı almadı.

Confirm, mandatory, nack ve dürüst ret

ch.confirm_delivery()
ch.queue_declare("jobs")
ch.basic_publish("", "jobs", "resize:img-1.png")
try:
    ch.basic_publish("", "no-such-queue", "lost?", mandatory=True)
except pika.exceptions.UnroutableError as e:
    ...
m, _, body = ch.basic_get("jobs");  ch.basic_nack(m.delivery_tag, requeue=True)
m, _, body = ch.basic_get("jobs");  ch.basic_ack(m.delivery_tag)
ch.exchange_declare("hdr", exchange_type="headers")

== publisher confirms + mandatory
confirmed: resize:img-1.png
UnroutableError: ['NO_ROUTE'] [312]

== nack with requeue
got resize:img-1.png redelivered = False -> basic_nack(requeue=True)
got resize:img-1.png redelivered = True -> basic_ack

== unsupported exchange type
ChannelClosedByBroker: 540 exchange type 'headers' is outside ADR-0016's scope; supported: direct, fanout, topic

Son satır bu tasarımın en sevdiğim kısmı. Desteklemediğim bir özelliği kullanan istemci sessiz bir kabul değil, ADR'nin adını veren bir kanal hatası (540 NOT_IMPLEMENTED) alıyor. Sessizce kabul edip hiç yönlendirmeyen bir headers exchange, hata ayıklaması günler süren türden bir sorundur.

mandatory ile ilgili bir ayrıntı: Basic.Return, confirm'den önce gönderiliyor. pika önce iadeyi topluyor, sonra ack'i görüp UnroutableError fırlatıyor; RabbitMQ da böyle davranıyor. .NET istemcisi bana arkasında ack olmayan bir Return'ün bir heartbeat'e kadar tamponda beklememesi gerektiğini öğretti; artık her toplu işlemin sonunda tek bir flush var.

Rakip tüketiciler ve prefetch

Bu, AMQP'yi eklememin sebebi olan özellik. İki worker aynı kuyruğu tüketiyor: biri hızlı (iş başına 50 ms), biri yavaş (500 ms). Kuyruğa 12 iş atıyorum. Önce prefetch olmadan, sonra basic_qos(prefetch_count=1) ile:

def worker(name, cost):
    conn = pika.BlockingConnection(P); ch = conn.channel()
    ch.queue_declare(Q, durable=True)
    if prefetch: ch.basic_qos(prefetch_count=prefetch)
    def on_msg(ch, m, props, body):
        time.sleep(cost)
        log.append((name, body.decode()))
        ch.basic_ack(m.delivery_tag)
    ch.basic_consume(Q, on_msg)
    ...

prefetch=unlimited fast:  6 jobs ['job-02', 'job-04', 'job-06', 'job-08', 'job-10', 'job-12']
prefetch=unlimited slow:  6 jobs ['job-01', 'job-03', 'job-05', 'job-07', 'job-09', 'job-11']
prefetch=1 fast: 10 jobs ['job-01', 'job-03', 'job-04', 'job-05', 'job-06', 'job-07', 'job-08', 'job-09', 'job-10', 'job-12']
prefetch=1 slow:  2 jobs ['job-02', 'job-11']

Prefetch olmadan saf round-robin: her iş sırayla bir sonraki tüketiciye gidiyor, yavaş worker kendi altı işini üç saniyede bitirirken hızlı olan boşta bekliyor. prefetch=1 ile her worker aynı anda en fazla bir onaylanmamış mesaj taşıyabiliyor; limitindeki tüketicinin sırası kapasitesi olana geçiyor. Hızlı worker 10 iş, yavaş olan 2 iş aldı. Prefetch'in bütün amacı bu: meşgul bir worker kuyruğu arkasında bekletmemeli.

Onaylanmamış mesajlar kanal ya da bağlantı ölürse kuyruğun önüne redelivered=true ile geri dönüyor. Diskteki kayıtları hiç silinmediği için bu pencerede bir çökme hiçbir şeyi değiştirmiyor.

Kalıcılık protokolden gelir: durable kuyruk ve kill -9

MQTT'de OXIDB_MQTT_PERSIST gerekiyordu, çünkü MQTT 3.1.1'in kablo seviyesinde depolama kavramı yok. AMQP'de var: istemci kuyruğu durable ilan ediyor ve mesajı delivery_mode=2 (persistent) ile yayınlıyor. Niyetini protokolde söylüyor. Bu yüzden OXIDB_AMQP_PERSIST diye bir şey yok: kalıcı mesajlı kalıcı bir kuyruk kalıcıdır, nokta. Geçici kuyruk bellekte yaşar, çünkü istemci durable demeyerek bunu istemiştir.

Deneme: invoices kalıcı kuyruğuna üç kalıcı fatura, bir de delivery_mode=1 taslak; ayrıca geçici scratch kuyruğuna bir mesaj. Hepsi publisher confirm ile. Ardından _amqp koleksiyonu:

{"_kind": "queue", "name": "invoices", "_id": 31, "_version": 1}
{"_kind": "qmsg", "queue": "invoices", "seq": 25, "exchange": "", "rkey": "invoices", "props": "kAAKdGV4dC9wbGFpbgI=", "body": "aW52b2ljZS0x", "_id": 32, "_version": 1}
{"_kind": "qmsg", "queue": "invoices", "seq": 26, "exchange": "", "rkey": "invoices", "props": "kAAKdGV4dC9wbGFpbgI=", "body": "aW52b2ljZS0y", "_id": 33, "_version": 1}
{"_kind": "qmsg", "queue": "invoices", "seq": 27, "exchange": "", "rkey": "invoices", "props": "kAAKdGV4dC9wbGFpbgI=", "body": "aW52b2ljZS0z", "_id": 34, "_version": 1}

(Koleksiyonda önceki denemelerden kalan başka kuyruk kayıtları da vardı, burada yalnızca invoices ile ilgili olanları bıraktım.) Taslak mesaj ve scratch'teki mesaj diskte yok: biri persistent değil, diğerinin kuyruğu durable değil. Gövde ve AMQP özellikleri base64 ile saklanıyor. props, içerik başlığının özellik bölümünün kablodan geldiği gibi kopyası; broker content_type, korelasyon kimlikleri veya uygulama başlıklarının hiçbirini modellemeden hepsini teslimde aynen geri veriyor.

Sonra kill -9 ve yeniden başlatma:

Kalıcı mesajlar geri geldi, content_type=text/plain korundu, geçici kuyruk ise yok (404). Kurtarılan mesajların hepsi redelivered=True. Bu bilinçli olarak temkinli bir seçim: bayrağın var olma sebebi "bu mesaj çökmeden önce teslim edilmiş olabilir" sözünü tutmak; tam teslim durumunu izlemek her teslimde bir disk yazması demek olurdu.

Burada ufak bir şey de fark ettim: log "recovered 5 durable queues" diyor ama gerçekte üç farklı kalıcı kuyruk vardı. Sebep, Queue.Declare'in kaydı broker kilidi tutulmadan "sil, sonra ekle" ile yazması; aynı kuyruğu üç bağlantı aynı anda ilan edince (prefetch denemesindeki iki worker ve kurulum bağlantısı) iki kayıt oluştu. Kurtarma entry().or_insert ile tekilleştirdiği için davranışa etkisi yok, yalnızca açılıştaki sayaç kayıt sayıyor. Sıradaki sürümde düzelteceğim küçük bir kusur.

MQTT ile AMQP arasında köprü

ADR-0016'nın son aşaması, iki protokolü tek ikili dosyada birbirine bağlamak. RabbitMQ'nun kendi MQTT eklentisinin yaptığı gibi: her MQTT yayını amq.topic adlı topic exchange'e / yerine . konarak yönlendiriliyor; amq.topic'e yapılan her AMQP yayını da MQTT tarafına . yerine / ile çıkıyor. amq.topic açılışta önceden ilan edilmiş halde geliyor, ona bağlanmak için exchange_declare gerekmiyor.

Senaryo: bir kazan sensörü MQTT ile sensors/boiler/temp'e yayın yapıyor, bir AMQP worker'ı kalıcı temp-alerts kuyruğunu amq.topic'e sensors.*.temp deseniyle bağlamış:

ch.queue_declare("temp-alerts", durable=True)
ch.queue_bind("temp-alerts", "amq.topic", routing_key="sensors.*.temp")   # amq.topic önceden ilan edilmiş
ch.basic_qos(prefetch_count=10)
def on_msg(ch, m, props, body):
    print(f"[amqp worker] exchange={m.exchange} routing_key={m.routing_key} "
          f"delivery_mode={props.delivery_mode} body={body.decode()}")
    ch.basic_ack(m.delivery_tag)
ch.basic_consume("temp-alerts", on_msg)

mosquitto_pub -p 15804 -i boiler-probe -q 1 -t sensors/boiler/temp -m '{"value": 91.5, "unit": "C"}' -d
mosquitto_pub -p 15804 -q 1 -t sensors/boiler/pressure -m '{"value": 2.1, "unit": "bar"}'

Client boiler-probe sending CONNECT
Client boiler-probe received CONNACK (0)
Client boiler-probe sending PUBLISH (d0, q1, r0, m1, 'sensors/boiler/temp', ... (28 bytes))
Client boiler-probe received PUBACK (Mid: 1, RC:0)

[amqp worker] waiting on temp-alerts <- amq.topic 'sensors.*.temp'
[amqp worker] exchange=amq.topic routing_key=sensors.boiler.temp delivery_mode=2 body={"value": 91.5, "unit": "C"}

Basınç ölçümü desene uymadığı için worker'a gelmedi. QoS 1 bir MQTT yayını AMQP tarafına delivery_mode=2 olarak geçiyor: dayanıklılık sözüyle gelen bir mesaj o sözü köprüde de koruyor. Ayrıca bu yayın sensors/# ingest kuralına da uyduğu için aynı anda sensor_readings koleksiyonuna yazıldı:

{"value": 91.5, "unit": "C", "_topic": "sensors/boiler/temp", "_received_at": 1791754410602, "_id": 4, "_version": 1}

Tek bir MQTT PUBLISH: canlı MQTT abonelerine, bir AMQP iş kuyruğuna ve sorgulanabilir bir koleksiyona. Hepsi PUBACK'ten önce.

Ters yön de çalışıyor. AMQP'den amq.topic'e devices.valve7.cmd anahtarıyla bir komut yayınladım, devices/+/cmd'ye abone bir mosquitto_sub aldı:

[amqp publisher] confirmed amq.topic devices.valve7.cmd
devices/valve7/cmd {"open": false}

Köprünün bir döngü koruması var: MQTT'den gelen bir yayın amq.topic'e girerken bridge: false ile işaretleniyor ve MQTT tarafına geri gönderilmiyor.

Son olarak köprünün en zor durumunu denedim: worker kapalıyken sensör iki QoS 1 ölçüm yayınlıyor, sunucu kill -9 ile ölüyor, yeniden açılıyor, worker sonra bağlanıyor:

published 2 (worker offline)
AMQP listening on 0.0.0.0:15805 - recovered 6 durable queues, 2 messages
[amqp worker] waiting on temp-alerts <- amq.topic 'sensors.*.temp'
[amqp worker] exchange=amq.topic routing_key=sensors.boiler.temp delivery_mode=2 body={"value": 92.8, "unit": "C"}
[amqp worker] exchange=amq.topic routing_key=sensors.boiler.temp delivery_mode=2 body={"value": 95.1, "unit": "C"}

Sensörün aldığı PUBACK, AMQP kuyruğundaki kopyanın da diskte olduğu anlamına geliyordu ve öyleydi. amq.topic kalıcı sayıldığı için kalıcı bir kuyruğun ona bağlanması da (_kind: "bind") diske yazılıyor; bağlama da yeniden başlatmadan sağ çıktı.

İçeride nasıl çalışıyor

Depolama: yeni format yok, doküman var

İki broker'ın durumu iki ayrı koleksiyonda, _kind alanıyla ayrılan dokümanlar olarak tutuluyor:

Bu kararın sebebi, ADR-0015'te yazdığım gibi, ikinci bir dayanıklılık yolu test etmek istememem. Doküman motorunun koleksiyon ve WAL makinesi zaten çökme kurtarma, checkpoint ve PITR'yi çözmüş durumda; broker bunları bedavaya alıyor. Sunucu "strict ACID-D" modunda açılıyor: her commit fsync ediliyor. Koleksiyonları ayırmamın sebebi de, iki broker'dan birinin diğerinin kurtarmasını bozamaması.

Kurtarma açılışta, dinleyici ilk bağlantıyı kabul etmeden önce çalışıyor: _kind alanına göre find sorguları, mesajlar seq'e göre sıralanıp kuyruklara yeniden diziliyor, sıra sayacı görülen en büyük seq'in bir fazlasından devam ediyor.

Topic eşleştirme: iki protokol, iki kural

MQTT ve AMQP joker karakterleri benzer görünür ama farklıdır, kodda da iki ayrı eşleştirici var:

• MQTT (/ ile ayrılmış seviyeler): + tam bir seviye, # yalnızca son seviyede ve "geri kalan her şey, hiçbir şey dahil". Broker abonelik filtrelerini açılışta bir kez sabitlenmiş (^...$) bir regex'e derliyor: + → [^/]+, sondaki /# → (/.*)? (bu sayede sensors/# tek başına sensors'ı da yakalar). Joker içermeyen bir filtre regex::escape ile birebir eşleşmeye dönüşüyor. İngest köprüsü ise aynı kuralı regex'siz, seviye dizileri üzerinde özyinelemeli küçük bir fonksiyonla uyguluyor.
• AMQP (. ile ayrılmış kelimeler): * tam bir kelime, # sıfır ya da daha fazla kelime ve desenin herhangi bir yerinde olabilir. topic_matches, # için iki dallı bir özyineleme: #'yi tüket ya da anahtardan bir kelime yut ve #'de kal.

Köprü bu iki dünyayı yalnızca ayırıcıyı değiştirerek birleştiriyor (/ ↔ .), tıpkı RabbitMQ'nun MQTT eklentisi gibi. Bir AMQP bağlaması desen olarak kendi kurallarını kullanıyor (sensors.*.temp), MQTT tarafındaki filtre de kendi kurallarını.

Bir exchange'e birden fazla bağlama aynı kuyruğa eşleşse bile kuyruk mesajın tek kopyasını alıyor: hedef listesi sıralanıp tekilleştiriliyor (RabbitMQ semantiği).

AMQP kuyruğu: paylaşılan bir nesne

MQTT'de dayanıklı durum istemciye aittir (oturum). AMQP'de ise kuyruk, bir tüketici listesi ve bir round-robin işaretçisi olan paylaşılan bir nesne. Her bağlantı kendi iş parçacığında çalışıyor ve broker'a "bana borçlu olduğun teslimler neler?" diye soruyor (poll). Hazırdaki her mesaj sıradaki tüketiciye gidiyor; o sıra başka bir bağlantıya aitse mesaj o bağlantının kendi turunu bekliyor. Tüketicilerin gerçekten rekabet etmesini, en hızlı sorgulayanın her şeyi kapmamasını sağlayan bu. Prefetch limitindeki bir tüketici atlanıyor.

Bu tasarım ilk başta 50 ms'lik bir poll aralığı yüzünden uçtan uca 51 ms gecikme veriyordu. Proje notlarına göre bu önce uyarlanabilir 1 ms'lik bir aralıkla 1,15 ms'ye, sonra çapraz iş parçacığı uyandırmasıyla 0,02 ms'ye indi: her tüketici bağlantının, okuma ucu poll(2) kümesinde TCP soketinin yanında duran engellemesiz bir pipe'ı var; bir kuyruğa mesaj düştüğünde yayıncı o kuyruğun tüketicilerinin pipe'ına bir bayt yazıyor ve çekirdek beklemesi anında bitiyor. (Yalnızca Unix'te; diğer platformlarda uyarlanabilir aralık kalıyor.)

Grup commit: fsync başına bir mesaj değil

Kalıcı yayının ilk sürümü her mesaj için bir fsync yapıyordu; proje ölçümlerine göre bu tek bağlantıda saniyede 264 mesaj demekti. İki katman bunu değiştirdi:

1. Bağlantı içi toplu yazma. Bağlantı, bir pipeline patlamasındaki yayınları prepare_publish ile yönlendirip biriktiriyor (en fazla 256 ya da soket tamponu boşalana kadar), sonra hepsini tek bir insert_many ile yazıyor ve confirm'leri yayın sırasıyla gönderiyor. Burada sıra önemli: disk yazması önce. Ack'in kuyruğa ekleme ile yarışıp henüz yazılmamış bir kaydı silmesi, bir çökmede tüketilmiş bir mesajı diriltirdi; mesaj başına eski yolda tam bu pencere yanlıştı.
2. Bağlantılar arası grup commit. O anda diskte bir commit yoksa gelen bağlantı lider oluyor ve bekleyen her şeyi tur tur yazıyor; diskteyken gelenler bir sonraki turu paylaşıyor ve kendi turlarının bitmesini bir Condvar üzerinde bekliyor. Pencerenin kendisi fsync'in süresi; tek başına bir yayıncı hemen kendi lideri oluyor ve bunun için hiçbir şey ödemiyor.

Sınırlar

İki broker da aynı disiplini izliyor: yavaş ya da ölü bir tüketici broker'ı belleksiz bırakamaz ve kayıp sessiz olmamalı. MQTT oturum kuyruğu 4096, AMQP hazır kuyruğu 100.000 mesajla sınırlı; ikisinde de en eskisi atılıyor, sayaç artıyor ve atılan mesajın diskteki kaydı da siliniyor. Sonuncusu önemli: sınır diske de uygulanmazsa ölü bir tüketici koleksiyonu sonsuza kadar büyütür ve bir yeniden başlatma, sınırın çoktan attığı mesajları diriltir. Bunu önce MQTT'de öğrendim, AMQP'de baştan doğru yaptım.

Proje ölçümleri: RabbitMQ ile karşılaştırma

Repoda, aynı Go istemcisiyle (rabbitmq/amqp091-go), aynı senaryolarla OxiDB'yi ve RabbitMQ'yu karşılaştıran bir benchmark var. Bu sayıları bu yazı için ben ölçmedim; projenin 2026-07-17 tarihli ölçümleri, bir arm64 MacBook üzerinde OxiDB 0.36.12 ile RabbitMQ 4.x (brew), 100 baytlık gövde:

Kaybedilen tek satır dürüst bir kayıp: OxiDB'de her kalıcı confirm, içinde geldiği toplu yazmanın gerçek bir F_FULLFSYNC'inin arkasında duruyor. Projenin notuna göre RabbitMQ'nun klasik kuyrukları diske tembel, aralıklı yazıyor; yani onun kalıcı confirm'i tek başına mesajın kalıcı depolamada olduğunu kanıtlamıyor. Sekiz bağlantıda grup commit devreye girince tablo dönüyor.

Sınırlar ve dürüst ödünler

• Düğüme yerel. MQTT oturumları ve AMQP kuyrukları, bağlantıların yaşadığı tek düğümde. Bir failover'dan sonra başka bir düğüme bağlanan istemci oturumuna devam edemez. Bu bir kaza değil: düğümler arası bir kuyruk, iki ADR'nin de reddettiği paylaşılan log (Kafka) tasarımı demek. Kuyruğun arkasındaki kalıcı veri (koleksiyonlar, zaman serileri) ise motorlar tarafından zaten replike ediliyor.
• MQTT 3.1.1, MQTT 5 değil. Paylaşılan abonelikler, mesaj süresi veya sebep kodları gibi MQTT 5 özellikleri yok. Rakip tüketici istiyorsanız cevap AMQP kuyruğu.
• AMQP alt kümesi. Yukarıdaki kapsam dışı listesi kesin: tx, headers exchange, TTL, öncelik, dead-letter, AMQP 1.0, yönetim arayüzü yok. Kullanan istemci açık bir hata alır.
• Sınırlı kuyruklar en eskiyi atar. Bu telemetri için doğru, bir ödeme kuyruğu için tartışmalı. Sınıra yaklaşan bir kuyruk bir alarm konusudur.
• Yeniden teslim temkinli. Çökmeden kurtarılan her AMQP mesajı redelivered=true gelir. Tüketicilerin idempotent olması zaten AMQP'nin "en az bir kez" sözünün gereği.
• Fsync'in bedeli gerçek. Tek yayıncılı, ardışık, kalıcı bir akışta ölçtüğüm fark yaklaşık 70 kattı. Kalıcılık gerekmeyen telemetri için QoS 0 ya da temiz oturum hâlâ en ucuz yol.
• Kimlik doğrulama basit. MQTT kullanıcı adı/parola ve AMQP PLAIN; ayarlanmadıklarında dinleyiciler açık ve 0.0.0.0'a bağlı.

OxiDB'nin geri kalanıyla nasıl birleşiyor

Mesajlaşma katmanı tek başına bir ürün değil, OxiDB'nin giriş kapısı. MQTT yayınları bellek içi bus üzerinden OxiMem'in Redis uyumlu SUBSCRIBE abonelerine de ulaşıyor (ve tersi). İngest köprüsü ölçümleri doküman motoruna yazıyor; oradan SQL motoruyla sorgulanabilir, uzun süreli metrikler için zaman serisi motoruna taşınabilirler. Bir AMQP worker havuzu, MQTT'den gelen iş akışını tüketip sonuçları yine OxiDB'ye yazabiliyor. Logları ve alarmları ise serinin log ve alarm yazısında anlatıyorum.

Benim için bu katmanın asıl dersi şuydu: bir broker'ın en önemli özelliği hızı değil, verdiği sözün kablodaki ile davranıştaki halinin aynı olması. OxiDB'de PUBACK artık "diskte" demek ve bunu her seferinde bir kill -9 ile kanıtlıyorum.

Denemek için

OxiDB'nin hazır derlenmiş ikili dosyaları ve Docker imajları GitHub releases sayfasında. MQTT ve AMQP'yi açmak için iki ortam değişkeni yeterli:

OXIDB_MQTT_PORT=1883 OXIDB_AMQP_PORT=5672 OXIDB_MQTT_PERSIST=1 oxidb-server

Sonrası, zaten kullandığınız araçlar: mosquitto_pub, paho-mqtt, pika, amqp091-go ya da RabbitMQ.Client. Hiçbiri OxiDB için değiştirilmedi. Serinin bir sonraki yazısında log ve alarm motoruna geçiyorum.