Topics

A topic keeps every message it is given, in the order the messages were committed, and hands each one a position: 1, 2, 3 and on, with no gaps. It is for the job a message broker does in a project that would rather not run one — events one part of an application publishes and several others read, each at its own pace.

It is slower than a dedicated broker and will stay slower. What it gives in exchange is one server fewer, and one guarantee a broker beside a database cannot give: a reader's position and the effect it writes commit together.

DEFINE NAMESPACE shop;
USE NAMESPACE shop;
DEFINE DATABASE orders;
USE DATABASE orders;
DEFINE TOPIC events;

Appending

Appending is CREATE or INSERT — the store already has the words:

CREATE events = { kind: 'paid', order: 'o-17' };
CREATE events = { kind: 'shipped', order: 'o-17' };
CREATE events:'evt-7f3a' = { kind: 'refunded', order: 'o-9' };
READ FROM events;

Three messages, each answering under its own identity as { position, value }, positions 1, 2 and 3.

A message is never changed. A correction is a new message, because a reader that already acted on the old one cannot be told it changed:

DELETE events:'evt-7f3a';

TopicIsAppendOnly, and the same for UPDATE and UPSERT. Naming the identity yourself, as evt-7f3a above, makes appending idempotent: a second CREATE of that identity is refused as it is for any table, so a producer that retries after a lost answer does not append twice.

Positions are dense, and decided at commit

The position is decided when the message commits, not when the statement runs. Two transactions appending at once get positions in the order they committed, and a reader never sees position 8 before position 7 exists: a read answers a prefix and stops at the first position its transaction cannot yet see.

Dense makes two questions arithmetic. How far behind is a reader: the last position minus its own. Did it miss anything: whether the next position it was given is one more than the last.

READ FROM events AFTER 1 LIMIT 1;

Position 2. AFTER n starts after position n, and without a LIMIT a read answers at most 100 messages.

A reader keeps its place in the store

DEFINE COLLECTION invoices;
BEGIN;
READ FROM events FOR CONSUMER 'billing' LIMIT 2;
CREATE invoices = { order: 'o-17', total: 40 };
COMMIT;

FOR CONSUMER 'billing' answers the messages after the position stored for billing and moves that position to the last message it answered, in the reader's own transaction. If the transaction commits, the invoice and the new position commit together; if it fails or is cancelled, neither does, and the next read is given the same messages again.

So an effect the reader writes into this store happens exactly once per message — not because anything deduplicates, but because the position and the effect are one commit. An effect outside the store — an email, a call to another service — happens at least once: do it, then commit, and a crash between the two repeats it. Make those effects idempotent.

READ FROM events FOR CONSUMER 'billing';

Only position 3 — billing already has the first two. A lone read with no transaction around it commits its move at once, which is at most once: a reader that crashes after the answer and before acting has lost those messages. Use the transaction form for anything that must not be lost.

Several readers under one name take turns. They write one stored position, so two reading at once conflict and the store runs the loser again, after the winner's position. Between them they are given every message once. Different names each have their own position and each see everything. There is no partition assignment: a name reads in one order, one batch at a time — which is the ordering promise and also the throughput ceiling. To share the work instead, declare a group (below).

AFTER n with FOR CONSUMER also moves the stored position there, which is how a reader is rewound or skipped forward:

READ FROM events FOR CONSUMER 'billing' AFTER 0 LIMIT 1;

Groups: several workers, each message acknowledged

A reader that commits its position can only say everything up to here is done. A group lets several workers share one topic and settle each message on its own — the shape of a consumer group in a broker, with one difference worth knowing: the group's state is one record in this store, so there is no coordinator to run and nothing to rebalance.

DEFINE TOPIC mail;
CREATE mail = { to: 'ada@example.com' };
CREATE mail = { to: 'grace@example.com' };
CREATE mail = { to: 'alan@example.com' };
DEFINE GROUP 'mailer' ON TOPIC mail ACK DEADLINE 30s;
READ FROM mail FOR CONSUMER 'mailer' LIMIT 10;

Once a group is declared under a name, READ FROM … FOR CONSUMER that name hands messages out rather than moving a position. Each one is held in flight with a deadline and answers as { position, value, deliveries }. The read above answers only position 1: a group holds one message at a time unless it says otherwise, which is what keeps the topic's order.

ACK mail FOR CONSUMER 'mailer' AT 1;
READ FROM mail FOR CONSUMER 'mailer';
NACK mail FOR CONSUMER 'mailer' AT 2 DELAY 10s;

• ACK … AT p, … — those messages are done and are never handed out to the group again. It answers how many were in flight; acknowledging one that is not counts nothing, so an acknowledgement retried after a lost answer is harmless.
• NACK … AT p, … [DELAY d] — hand them out again, now or after the delay.
• No answer at all — once ACK DEADLINE passes, the next read hands the message out again. Nothing sweeps: the deadline is written in the group's record and the next read compares it with the clock.

ACK DEADLINE is required and has no default, because a deadline the store guessed would decide how long a crashed worker holds a message. The rest is optional:

DEFINE TOPIC mail_dead;
DEFINE GROUP 'bulk' ON TOPIC mail ACK DEADLINE 1m DELIVERIES 5 IN FLIGHT 100 DEAD LETTER TO mail_dead;

• IN FLIGHT n (default 1, at most 10 000) — how many messages the group holds unacknowledged at once. One keeps the topic's order: the next message waits until the previous one is settled. More lets several workers go at once, and a message handed out again may then arrive after later ones.
• DELIVERIES n — a message due to be handed out an n+1th time is given up on instead, and appended to the DEAD LETTER TO topic as { topic, group, position, deliveries, value } when the group names one. A group that names a dead letter must also say DELIVERIES, since otherwise nothing would ever go there.

Several workers reading under one group name take turns on its record, as readers under one name take turns on a position; a read or an acknowledgement that meets another is run again by the node.

ALTER GROUP 'mailer' ON TOPIC mail START AT 0;
DROP GROUP 'bulk' ON TOPIC mail;

START AT n makes the group hand out position n + 1 next and forget what it held in flight — a replay after a fix, or a skip past messages nobody needs.

What each way of reading promises:

A group is at least once: a worker that did the work and crashed before its ACK sees the message again. Make the effect idempotent — the topic, the group and the position together are a stable key for it.

Four refusals belong to groups. AFTER on a group's read is AfterOnGroup, because a member cannot skip the group's work; ACK or NACK under a name with no group is NotAGroup, because nothing is held to settle; a topic cannot be its own group's dead letter (DeadLetterIsTheTopic); and declaring a group twice is GroupExists. A group declared under a name that already has a stored reader position takes that position over.

The five clients each carry a consumer that runs this loop for you — a handler per message, acknowledged when it returns or settled by what it answers. See Rust, Python, JavaScript, Go and Kotlin.

Retention, and a reader it left behind

DEFINE TOPIC audit RETAIN 7d MAX BYTES 4096;

RETAIN 7d keeps a message seven days after it was appended; the node's housekeeping then removes it through the log like any other delete. A topic without RETAIN keeps everything. There is no compaction — a topic that keeps only the newest value per key is a space, which already exists.

From 0.22.0-beta a topic can also be bounded by size:

DEFINE TOPIC clicks RETAIN BYTES 1073741824;
DEFINE TOPIC trail RETAIN 30d RETAIN BYTES 10737418240;

RETAIN BYTES n keeps at most n bytes of messages — the encoded size MAX BYTES measures, summed. An append that would take the topic past it removes the oldest messages in the same commit, as ordinary deletes in its log record, so a follower applies the removal rather than deciding it and the total is exact however many writers append at once. With both limits a message goes at whichever it reaches first. The commit's own messages are never the ones removed: a commit whose own messages add up to more than n is refused as TopicRetainExceeded rather than trimmed, because a message that vanished on commit is a loss nobody would be told about.

Removing messages never reuses a position. A reader whose next position was removed is told: the read carries a lapsed note naming how many messages it passed over, and moves a named reader past them so the note is given once. Nothing is skipped silently.

MAX BYTES n refuses a message whose encoded value is larger, with TopicMessageTooLarge.

What INFO FOR TOPIC reports

INFO FOR TOPIC events;

first is the first position still held, last the last one given, retain_bytes and the bytes held for a topic bounded by size, each reader's position and lag, and each group's figures: position (the last message handed out), committed (the last one below which everything is acknowledged), lag counted from committed, in_flight, redelivered, dead_lettered, its deadline and its width. Watch lag: a reader whose lag keeps growing will one day be passed by retention, and the lapsed note will be the first thing that says so. ingested_by names the topic consumers reading the topic into a table — each with its group, its table and whether it is running here.

A topic's messages can be written into a table by a declaration rather than by a worker you run: DEFINE TOPIC CONSUMER reads the topic as a member of a group and writes each message into a table in the transaction that acknowledges it, so each one lands exactly once — see Stream ingestion.

Publishing once

DEFINE TOPIC payments DEDUPLICATE 5m ON payment_id;
CREATE payments = { payment_id: 'p1', amount: 10 };
CREATE payments = { payment_id: 'p1', amount: 10 };
READ FROM payments;

A message whose payment_id was published less than the window ago is not appended — no position is spent on it — and answers no record; an INSERT appends its other rows (from 0.33.4-beta). A message without the field is appended. A key that is not a string, an integer, a uuid or bytes is refused as DeduplicationKey. A topic declared PUBLIC cannot deduplicate, because its answer would tell a stranger which keys were published.

Letting anyone append

DEFINE TOPIC signups RETAIN 1d MAX BYTES 1024 PUBLIC RATE 60 PER 1m;

PUBLIC RATE n PER d lets a caller that has not signed in append to this topic and do nothing else on a store that otherwise refuses them — a sign-up form, a telemetry beacon. It is refused unless the declaration also says MAX BYTES, because an open door with no bound on what comes through it is a way to fill the store from anywhere.

An anonymous caller may run USE NAMESPACE and USE DATABASE, and CREATE or INSERT into the topic with an identity the store generates. Naming one is refused, because the refusal of an identity already taken would tell a stranger which messages exist. Nothing that reads is allowed — not a subquery in the value, not a read of the topic itself. Everything else answers that the store requires a signed-in user.

The rate is counted in messages, per node, in memory: a quiet topic takes n at once and then one every d / n, and an append past it is refused with TopicRateExceeded. Two things follow and are worth sizing for: a cluster of three nodes that all accept writes admits three times the rate, and a restarted node has forgotten what it counted. A signed-in caller is never rated.

What a topic is not

A topic cannot be split into shards in this build: one topic is one order. Messages are read by asking; a subscriber that wants them pushed follows the change feed of the topic's table. And a topic is not a queue: a queue hands each record to one worker and forgets it when the work is done, while a topic keeps every message for every reader.

DROP TOPIC audit;