AMQP Progress: Core Functionality Mostly Done, Consume Is What's Left

AMQP is the third protocol, after Kafka, to validate the "one data store, many protocols" kernel architecture. Its consumption model is different from both Kafka and MQTT: there's no offset-commit step, Get/Consume take messages one at a time, and it needs to support explicit ack, nack, and requeue.

Current state: exchange/queue management, all four routing types, and the entire Basic.Get path (cursor, unacked state, crash recovery) are all working. The one major piece left is wiring Basic.Consume into the same mechanism. This is a first-pass implementation — the goal was to prove the architecture can carry AMQP's semantics, not to reach production-readiness yet. Here's a concrete rundown of what's done and what isn't.

What's done

Exchange and queue declare, delete, bind, unbind, and purge all work. All four routing types (direct, fanout, topic, headers) are implemented, including exchange-to-exchange chained bindings. Basic.Publish stores message properties in full and returns them unchanged on delivery.

Basic.Get is the most complete part of this pass: a shared delivery cursor (not pinned to a connection, so no duplicate delivery across nodes), concurrency handled through meta-service's conditional writes rather than a lock, delivery_tag incrementing per channel, unacked messages tracked both in an in-memory table and in a persistent index written to an internal topic, Ack/Nack/Reject/Recover all going through one settlement path, and requeue implemented as a shared function called from graceful channel/connection close and from the crash-recovery scanner. Earlier posts already covered this in detail, so I won't repeat it here.

What's not done, or only half done:

1. Basic.Consume is still the original implementation: its own loop, its own cursor, entirely separate from Get's shared cursor and unacked index. If Get and Consume run against the same queue at the same time, they don't share progress — this is the biggest gap right now.
2. Basic.Cancel and Basic.Qos only acknowledge at the protocol level — they don't actually stop pushes or enforce a prefetch limit.
3. The Tx class is an empty implementation. Select/Commit/Rollback all just reply Ok, with no real transaction buffering behind them.
4. Non-durable queues persist messages the same way durable ones do, so they survive a restart. That's inconsistent with the protocol's "non-durable queues should be dropped" semantics — a gap left over from an earlier pass, not something fixed this time.
5. Integration tests currently only cover the Basic.Get path. Consume and the exchange/queue management APIs don't have test coverage yet.

What's next

The next step is migrating Basic.Consume onto the same shared cursor and unacked mechanism Get already uses, so both paths book-keep delivery through the same ledger — that way competing consumption on a queue is actually competing consumption, not two parallel tracks. Once that's done, AMQP's core path is complete end to end.

After that: filling in details and tests — making Cancel actually stop pushes, making Qos actually enforce prefetch, and extending integration tests to Consume and the management APIs. The transaction and non-durable-queue gaps stay open; they're not in scope for this phase.

Summary

The Kafka post's conclusion was that "one data store, many protocols" holds for Kafka just as much as for MQTT. AMQP adds something more specific: concurrency problems like "who gets to read the next message" can be solved with meta-service's existing conditional writes, without inventing a new lock or coordination mechanism for every new protocol. That's the real payoff of this architecture — it's not just that three protocols' worth of code shares one storage engine, it's that details like "how do we guarantee concurrency safety" get reused too.

Once AMQP is validated, the focus goes back to the main line: MQTT and Kafka. This round of architecture validation is essentially done. The next goal is building a genuinely production-ready MQTT broker, and connecting the path across edge IoT, big data, and AI.