Transactions in MoonMQ

Status: read_committed isolation + Last Stable Offset shipped (2026-07-18), completing the exactly-once story on top of atomic multi-partition transactions (2026-07-10) and the session-scoped idempotent producer (2026-06-21).

What now ships:

• Durable producers (src/storage/producer_state.lua): a producer opened with a stable producer_name (Kafka's transactional_id) gets a PID + epoch that survive reconnects and broker restarts. Each new session bumps the epoch, fencing the old one (ERR_PRODUCER_FENCED). Idempotent-produce sequence memos are persisted, so a retry replays the original ack across a restart.
• Atomic multi-partition transactions (src/broker/txn_coordinator.lua): BEGIN_TXN / transactional PRODUCE_IDEMPOTENT / TXN_OFFSET_COMMIT / END_TXN. On commit the coordinator writes COMMIT control records to every participant partition and applies buffered offset commits; on abort it writes ABORT markers and drops the offsets. State lives in the internal __transaction_state topic and is re-driven idempotently on crash recovery (ONGOING → abort, PREPARE_COMMIT → roll forward).
• read_committed isolation + Last Stable Offset (2026-07-18). How it fits together:
  • Transactional data records carry the producer session in the record header (ATTR_TXN bit + pid/epoch — see src/record/message.lua), so a reader can attribute any record to its transaction.
  • The produce path enrols a partition with the coordinator BEFORE the transaction's first append there, capturing the partition's LEO as the txn's first_offset — the lower bound of everything it writes there.
  • LSO = min first_offset over unresolved transactions touching the partition (Coordinator:lso); a read_committed consumer never reads at or past it, so every record it does read belongs to a resolved txn.
  • On abort, the coordinator durably records the range [first_offset, abort_marker) per participant in the abort index (src/broker/abort_index.lua, <data_dir>/txn-aborts.json); below the LSO, records matching an aborted (pid, epoch, range) are filtered out. Entries are pruned once retention deletes their whole range.
  • On the wire, FETCH/SUBSCRIBE take an optional trailing isolation byte (0 = read_uncommitted — the default, and what old clients send; 1 = read_committed). The Lua client: Client.new{ isolation = "read_committed" }.
  • Boundary: transactional produce to a partition owned by a peer broker is refused (Producer:produce) — markers could not be written on the owner.
  • Boundary: abort-index ranges are offset-based, so they assume offsets are stable — true on the segmented backend (the default for user topics; its retention deletes whole segments without renumbering). Don't run transactions against a commitlog-backend topic with cleanup_policy=compact: compaction renumbers surviving records, which would break the aborted-range filter.

This document records:

1. What ships today.
2. What is intentionally NOT in scope today.
3. The original design surface, retained for context.

1. Today: idempotent and transactional producers

A client asks for a producer id with INIT_PRODUCER_ID. Without a producer_name (Client.new{ idempotent = true }) the id is ephemeral: it lives for one connection, and the guarantees and non-guarantees below apply. With a producer_name it is durable: PID, epoch and sequence memos survive reconnects and restarts (__producer_state), a new session fences the old one, and the producer may run transactions (see the client example at the end of this section).

Guarantees

Within the lifetime of a single TCP connection that called INIT_PRODUCER_ID (and, for a named producer, across connections), the broker guarantees:

• A PRODUCE_IDEMPOTENT retry with the same (PID, topic, seq) returns the original (partition, offset) instead of appending a duplicate.
• An out-of-order seq (gap or regression) is rejected with ERR_OUT_OF_ORDER_SEQUENCE so the client can detect lost messages.

Non-guarantees (unnamed producers)

• No durability across the broker process. Ephemeral PIDs and sequence memos live in broker memory. A restart wipes them; clients must call INIT_PRODUCER_ID again on reconnect.
• No cross-connection idempotence. Two unnamed connections get two distinct PIDs; the broker can't dedup retries that move across connections. Give the producer a producer_name for that.
• No transaction boundary. Without BEGIN_TXN/END_TXN, multiple PRODUCE_IDEMPOTENT calls do NOT form an atomic group; a crash mid-batch leaves a partial write. Transactions require a named producer.

Wire format

Opcodes (additive on PROTOCOL_VERSION 1):

Acks reuse OP_PRODUCE_ACK (0x82) with the original (partition, offset). The optional trailing headers block is described in headers.md. Errors use OP_ERROR with codes:

Why per-(PID, topic) and not per-(PID, partition)?

Real Kafka tracks sequences per (PID, topic, partition). We track per (PID, topic) because:

• TCP guarantees per-connection in-order delivery, so a single monotonic counter across partitions is internally consistent.
• The client doesn't have to know the broker's partition count or hashing function to pick the right seq slot.
• It costs: idempotent producers can't batch across partitions in parallel (we don't anyway).

Consumer groups and offset commits have since landed (see §3); moving to the full Kafka per-partition layout remains possible but has not been needed.

Client API

local Client = require("src.client")
 
-- Open a connection AND ask the broker for a producer ID.
local c = Client.new{
    host       = "127.0.0.1",
    port       = 9092,
    username   = "admin",
    password   = "admin",
    idempotent = true,
}
 
-- produce() auto-uses the idempotent path when pid is set.
local ack = c:produce("orders", "key-1", "value-1")
-- ack.seq is the sequence number the broker dedup-keyed on.
 
-- For testing: send a specific seq (e.g. simulate a retry).
local replay = c:produce_at_seq("orders", "key-1", "value-1", 0)
-- replay.partition == ack.partition, replay.offset == ack.offset

A named producer runs transactions — here a consume-transform-produce step that commits its output and its input offset together:

local p = assert(Client.new{ host = "127.0.0.1", port = 9092,
                             username = "admin", password = "admin",
                             producer_name = "billing-writer" })
 
assert(p:begin_transaction())
assert(p:produce("invoices", "order-42", "…"))
assert(p:send_offsets_to_transaction("billing", {
    { topic = "orders", partition = 2, offset = 43 },
}))
assert(p:commit_transaction())          -- or p:abort_transaction()

2. Out of scope (today)

Still deferred: nothing from the original design.

Now implemented (were previously out of scope):

• Transactional produce to a partition owned by a peer broker (2026-07-19). The coordinator enrols the OWNER before the forwarded append (POST /cluster/txn/enroll floors the owner's LSO at its pre-append LEO), forwards the record, and on END_TXN writes the COMMIT/ABORT marker on the owner and resolves there (POST /cluster/txn/resolve — on abort this also records the aborted range in the owner's abort index, since the owner's consumers do the read_committed filtering). Caveats: the owner's LSO floor for a REMOTE transaction is in-memory — if the owner restarts while the peer's transaction is unresolved, read_committed readers there may see the transaction's records until the markers arrive; and reassigning a partition away mid-transaction is unsupported (ownership is re-checked at marker time, but offsets captured at enrol time are not migrated).
• PID expiry / garbage-collecting idle producer state (2026-07-19): durable producers idle for Server.ProducerExpirySeconds (default 1 day, 0 = off) have their identity + memos tombstoned in __producer_state; compaction drops the tombstones. Producers with an unresolved transaction or a live connection are never expired, and a persisted allocator watermark guarantees an expired PID is never re-issued.
• COMMIT / ABORT control records in the log (v2 record format, control bit).
• Cross-partition atomic commit via the transaction coordinator.
• Transactional offset-commit (TXN_OFFSET_COMMIT, applied on commit).
• __transaction_state coordinator topic + crash recovery.
• PID generation epochs and producer fencing across reconnects.
• read_committed consumer isolation level (2026-07-18).
• Last Stable Offset (LSO) tracking on partitions (2026-07-18).
• Filtering aborted records out at read time via the durable abort index (2026-07-18).

3. Original design: full Kafka-style transactions

Everything below has since landed (including read_committed + LSO as of 2026-07-18); the section is retained as the design rationale.

• Consumer subsystem. commit_offset / load_offset are now durable (src/broker/consumer.lua → OffsetManager).
• __consumer_offsets topic. Implemented (src/storage/offset_manager.lua); __producer_state and __transaction_state follow the same internal-topic pattern.

Components

1. Transaction coordinator. A new broker subsystem that owns __transaction_state. Producers route INIT_PRODUCER_ID to the coordinator, which tracks (transactional_id → (PID, epoch, state)) durably. As shipped: the server assigns PIDs inline from __producer_state, and src/broker/txn_coordinator.lua owns the transaction state.

2. Producer epochs. PIDs become 2-field: (pid: u64, epoch: u16). On reconnect with a known transactional_id, the coordinator bumps epoch and fences older sessions (rejects writes with stale epoch via ERR_PRODUCER_FENCED).

3. Control records. Every transactional log append carries a marker bit. The coordinator writes COMMIT / ABORT control records to every affected partition at txn boundary. The records are written through the same log path so consumers see them.

4. Read-committed isolation. Consumers in read_committed mode stop at the Last Stable Offset (LSO) — the smallest offset above which any partition has uncommitted transactional records. Aborted records below LSO are filtered out at read time using the ABORT markers.

5. Transactional offset commit. OFFSET_COMMIT_REQUEST carries the PID + epoch + sequence; the offset commit becomes part of the txn and only takes effect when the COMMIT control record lands.

Opcodes (proposed)

As shipped: INIT_TRANSACTIONAL_ID became INIT_PRODUCER_ID with a producer_name; ADD_PARTITIONS_TO_TXN became implicit enrolment on the first transactional produce to a partition; WRITE_TXN_MARKERS became the inter-broker /cluster/txn/enroll and /cluster/txn/resolve endpoints. BEGIN_TXN, END_TXN and TXN_OFFSET_COMMIT shipped as proposed.

Error codes (proposed)

Storage shape

• __transaction_state partition-of-N stores (transactional_id, pid, epoch, state, participating_partitions[], last_update_ms, timeout_ms).
• Per-partition log adds a 1-bit transactional flag in the record header (claim one of the high bits of key_size, or add a v2 record format with a leading control byte).

Migration

Plan:

1. Land consumer groups + durable offset commit (blocking).
2. Add __consumer_offsets and __transaction_state as internal topics (broker-managed, not user-creatable).
3. Add v2 record format with the control flag.
4. Wire coordinator + new opcodes.
5. Extend the consumer to honour read_committed + LSO.
6. Once all above land, the in-memory PID table from §1 becomes a short-circuit: producers with transactional_id == nil keep the session-scoped behaviour; producers with a txn-id go through the coordinator.

References

• Kafka KIP-98: Exactly Once Delivery and Transactional Messaging
• Kafka KIP-129: Streams EOS
• Confluent's Transactions in Apache Kafka blog post (2017)