Topic Pub/Sub

A topic has any number of independent subscribers. A subscriber is nothing special — it's an ordinary task plus a registry row mapping (topic, subscription name) → (task, queue). Publishing to a topic fans the message out as one ordinary job per active subscription, so every task feature — retries, the dead-letter queue, middleware, rate limits — applies independently per subscriber. A failing subscriber never affects its siblings, and publishing to a topic with zero subscribers is a valid no-op that returns an empty list.

Quick start

Register a subscriber like any other task, then publish a message to the topic — every active subscription gets its own delivery:

from flexiq import Queue

queue = Queue()

@queue.subscriber("orders", name="email")
def send_confirmation_email(order_id: int) -> None:
    ...

@queue.subscriber("orders", name="analytics")
def track_order(order_id: int) -> None:
    ...

queue.declare_subscriptions()  # only needed before run_worker() has started
queue.publish("orders", 42)    # -> [JobResult, JobResult] — one per subscriber

import { Queue } from "@byteveda/flexiq";

const queue = new Queue({ dbPath: "orders.db" });

queue.subscriber("orders", "sendConfirmationEmail", (orderId: number) => {
  // ...
}, { subscriptionName: "email" });

queue.subscriber("orders", "trackOrder", (orderId: number) => {
  // ...
}, { subscriptionName: "analytics" });

await queue.declareSubscriptions(); // only needed before runWorker() has started
await queue.publish("orders", [42]); // -> Job[] — one per subscriber

import org.byteveda.flexiq.FlexiQ;
import org.byteveda.flexiq.pubsub.SubscriptionOptions;
import org.byteveda.flexiq.task.Task;

Task<Integer> sendConfirmationEmail = Task.of("send_confirmation_email", Integer.class);
Task<Integer> trackOrder = Task.of("track_order", Integer.class);

try (FlexiQ flexiq = FlexiQ.builder().sqlite("orders.db").open()) {
    flexiq.subscribe("orders", sendConfirmationEmail,
            SubscriptionOptions.builder().name("email").build());
    flexiq.subscribe("orders", trackOrder,
            SubscriptionOptions.builder().name("analytics").build());

    flexiq.publish("orders", 42); // -> List<Job> — one per subscriber
}

flexiq.worker()
        .handle(sendConfirmationEmail, orderId -> mailer.sendConfirmation(orderId))
        .handle(trackOrder, orderId -> analytics.record(orderId))
        .start();

Subscription lifecycle

Durable vs. ephemeral

Subscriptions are durable by default: the registry row persists until you unsubscribe, and a worker that's down just lets its deliveries pile up as ordinary pending jobs. An ephemeral subscription (durable=False / durable: false / SubscriptionOptions.durable(false)) ties the subscription to one worker process instead — it's only registered while that worker is running, and it's automatically reaped once that worker stops heartbeating. Use ephemeral subscriptions for throwaway consumers (a debug tail, a short-lived fan-out watcher) that shouldn't leave a stale registration behind if the process never comes back.

@queue.subscriber("orders", name="debug-tail", durable=False)
def tail(order_id: int) -> None:
    print(f"order {order_id}")

queue.run_worker()  # the subscription registers here, owned by this worker

queue.subscriber("orders", "tail", (orderId: number) => {
  console.log(`order ${orderId}`);
}, { durable: false });

queue.runWorker(); // the subscription registers here, owned by this worker

Task<Integer> tail = Task.of("tail", Integer.class);
flexiq.subscribe("orders", tail, SubscriptionOptions.builder().durable(false).build());

// The subscription registers once a worker actually starts and handles it.
try (Worker worker = flexiq.worker().handle(tail, orderId -> log(orderId)).start()) {
    ...
}

Re-registering updates instead of duplicating

Registering the same (topic, subscription name) again updates its routing target (task, queue) in place instead of creating a second row — safe to call on every process restart.

Pause, resume, unsubscribe

Pausing stops deliveries without forgetting the subscription — useful for a maintenance window on one consumer while the others keep receiving messages. Unsubscribing removes the registry row entirely.

queue.pause_subscription("orders", "email")   # stop deliveries to "email"
queue.publish("orders", 1)                    # only "analytics" is delivered
queue.resume_subscription("orders", "email")

queue.unsubscribe("orders", "email")          # remove it entirely
queue.list_subscriptions("orders")            # active subscriptions for a topic
queue.list_topics()                           # distinct topics with a subscriber

await queue.pauseSubscription("orders", "email");
await queue.publish("orders", [1]);            // only "analytics" is delivered
await queue.resumeSubscription("orders", "email");

await queue.unsubscribe("orders", "email");    // remove it entirely
await queue.listSubscriptions("orders");       // active subscriptions for a topic
await queue.listTopics();                      // distinct topics with a subscriber

flexiq.pauseSubscription("orders", "email");
flexiq.publish("orders", 1);                  // only "analytics" is delivered
flexiq.resumeSubscription("orders", "email");

flexiq.unsubscribe("orders", "email");        // remove it entirely
flexiq.listSubscriptions("orders");           // active subscriptions for a topic
flexiq.listTopics();                          // distinct topics with a subscriber

Late-join boundary

Delivery semantics

At-least-once, per subscriber

Each delivery is an ordinary job, so it inherits the same at-least-once delivery guarantee as everything else: a delivery is never silently lost, but a crash between running the handler and recording the result can run it twice. A subscriber that exhausts its retries dead-letters independently — the send_confirmation_email delivery failing and dead-lettering has no effect on the track_order delivery for the same message.

Delivery settings resolve per subscriber

An unset delivery setting (priority, max retries, timeout) resolves in this order:

1. An explicit option passed to publish().
2. The subscriber task's own registered default.
3. The queue's default.

So one high-priority subscriber and one best-effort subscriber can consume the exact same published message, each retried and prioritized according to its own task registration — unless the publisher explicitly overrides a setting for every delivery.

Idempotency-key dedup is per subscriber

idempotency_key / idempotencyKey dedupes per subscriber, not per publish call — internally it's salted with the subscription name before becoming the delivery's dedup key. Republishing the same key creates no new deliveries for subscribers that already received it, but a subscription added after the first publish still gets its own copy the next time that key (or any key) is published, because its salted key hasn't been used yet.

first = queue.publish("orders", 1, idempotency_key="evt-42")   # 2 subscribers -> 2 jobs
second = queue.publish("orders", 1, idempotency_key="evt-42")  # same jobs, no new rows

@queue.subscriber("orders", name="audit")
def audit_order(order_id: int) -> None: ...
queue.declare_subscriptions()

third = queue.publish("orders", 1, idempotency_key="evt-42")   # "audit" gets its own delivery

const first = await queue.publish("orders", [1], { idempotencyKey: "evt-42" });  // 2 jobs
const second = await queue.publish("orders", [1], { idempotencyKey: "evt-42" }); // same jobs

queue.subscriber("orders", "auditOrder", (orderId: number) => {}, { subscriptionName: "audit" });
await queue.declareSubscriptions();

const third = await queue.publish("orders", [1], { idempotencyKey: "evt-42" }); // "audit" delivered

PublishOptions keyed = PublishOptions.builder().idempotencyKey("evt-42").build();
List<Job> first = flexiq.publish("orders", 1, keyed);   // 2 subscribers -> 2 jobs
List<Job> second = flexiq.publish("orders", 1, keyed);  // same jobs, no new rows

Task<Integer> auditOrder = Task.of("audit_order", Integer.class);
flexiq.subscribe("orders", auditOrder, SubscriptionOptions.builder().name("audit").build());
List<Job> third = flexiq.publish("orders", 1, keyed);   // "audit" gets its own delivery

Every delivery is stamped with topic and subscription

Every delivery's structured notes carry the topic and the subscription name it was routed through, merged with any notes the publisher supplied — filter or group deliveries on the dashboard, or in list_jobs()/listJobs() queries, by which subscriber received them.

(result,) = queue.publish("orders", 5, notes={"tenant": "acme"})
job = queue.get_job(result.id)
job.notes  # {"topic": "orders", "subscription": "email", "tenant": "acme"}

const [job] = await queue.publish("orders", [5], { notes: { tenant: "acme" } });
JSON.parse(job.notes ?? "{}");
// { topic: "orders", subscription: "email", tenant: "acme" }

List<Job> jobs = flexiq.publish("orders", 5,
        PublishOptions.builder().notes(Map.of("tenant", "acme")).build());
Map<String, Object> notes = jobs.get(0).notesMap().orElseThrow();
// {topic=orders, subscription=email, tenant=acme}

Cross-SDK topics

A publish writes exactly one serialized payload, shared by every delivery — it's encoded once with the queue-level serializer, not per-subscriber, so per-task serializer overrides don't apply to topic deliveries.

Log topics

Every subscription above is fan-out: one job per subscriber, each with its own retries and DLQ. A log subscription is the other shape — a named, durable cursor with no handler, reading a single stored copy of every published message at its own pace. Publishing to a topic with log subscribers writes one message no matter how many are reading it (O(1) publish, versus fan-out's one job write per subscriber), and a topic can mix both kinds: the same publish stores the log message once and fans out jobs to any fan-out subscribers on it.

Reach for fan-out when each subscriber has its own independent side effect that needs its own retries and DLQ. Reach for a log subscription for a replayable stream one consumer works through at its own pace, or to publish once to many — or unknown — consumers without multiplying writes by subscriber count.

queue.subscribe_log("orders", "audit-log")

queue.publish("orders", 42)  # -> [] no fan-out jobs; one message stored

for msg in queue.read_topic("orders", "audit-log"):
    audit_sink.write(msg.args, msg.kwargs)
    queue.ack_topic("orders", "audit-log", msg.id)

await queue.subscribeLog("orders", "audit-log");

await queue.publish("orders", [42]); // -> [] no fan-out jobs; one message stored

for (const msg of await queue.readTopic("orders", "audit-log")) {
  auditSink.write(msg.args);
  await queue.ackTopic("orders", "audit-log", msg.id);
}

// Keep the Serializer you configure on the builder — readTopic() returns a
// raw byte[] payload, and FlexiQ has no getter to recover it later.
Serializer serializer = new JsonSerializer();
try (FlexiQ flexiq = FlexiQ.builder().sqlite("orders.db").serializer(serializer).open()) {
    flexiq.subscribeLog("orders", "audit-log");

    flexiq.publish("orders", 42); // -> [] no fan-out jobs; one message stored

    for (TopicMessage msg : flexiq.readTopic("orders", "audit-log")) {
        int orderId = serializer.deserialize(msg.payload, Integer.class);
        auditSink.write(orderId);
        flexiq.ackTopic("orders", "audit-log", msg.id);
    }
}

The cursor is a high-water mark

read_topic() / readTopic() returns messages oldest-first, exclusive of the subscription's cursor — each message's id doubles as the cursor token you pass back. ack_topic() / ackTopic() advances the cursor to a given id: acking id X acks everything up to and including it, and it's monotonic, so acking an id older than what's already acked is a no-op that returns false.

At-least-once, same contract as fan-out

A log subscription gets the same at-least-once delivery guarantee as everything else: read, process, then ack. A consumer that crashes (or otherwise fails) between reading and acking re-reads the same messages the next time it calls read_topic() / readTopic() — the cursor only moves forward on an explicit ack. A log subscription also shares the same late-join boundary as fan-out subscriptions above: subscribe_log() / subscribeLog() only sees messages published after it registered — unless the topic is declared (below).

Declared topics

By default a log message is stored only when a durable log subscription already exists at publish time. Declaring a topic lifts that late-join boundary: its publishes are retained even with zero subscribers, so a consumer that subscribes later reads the backlog that's still within the topic's retention window (anything already expired is gone). Declaring is idempotent.

queue.declare_topic("orders", retention=3600)  # keep up to 1 hour

queue.publish("orders", 42)          # retained even with no subscriber yet
queue.subscribe_log("orders", "audit-log")
queue.read_topic("orders", "audit-log")  # -> sees the earlier publish

await queue.declareTopic("orders", { retention: 3600 }); // seconds

await queue.publish("orders", [42]); // retained even with no subscriber yet
await queue.subscribeLog("orders", "audit-log");
await queue.readTopic("orders", "audit-log"); // -> sees the earlier publish

flexiq.declareTopic("orders", Duration.ofHours(1));

flexiq.publish("orders", 42); // retained even with no subscriber yet
flexiq.subscribeLog("orders", "audit-log");
flexiq.readTopic("orders", "audit-log"); // -> sees the earlier publish

retention bounds a sub-less backlog: with no live log subscriber, each stored message expires that long after it was published so the retention sweep can reclaim it. Once a log subscriber exists, cursor compaction takes over (next section). Omit retention to keep messages until a subscriber consumes them.

Managed consumer

Instead of writing the read/ack loop yourself, register a managed consumer: the SDK creates the log subscription and, once a worker runs, drives a background poll loop that pulls each message, invokes your handler, and advances the cursor for you. The handler runs in-SDK (not as a fan-out job), so there's no per-message retry/DLQ — the delivery guarantee is the same at-least-once cursor contract as the manual loop. The Python and Node handlers may be synchronous or async (a returned promise/coroutine is awaited); the Java handler is a synchronous Consumer<T>. Run a given (topic, name) consumer in one worker process — its cursor is a single high-water mark.

@queue.log_consumer("orders", "audit-log")
def audit(order_id: int) -> None:
    audit_sink.write(order_id)

queue.run_worker()  # spawns the consumer's poll loop

queue.logConsumer("orders", "audit-log", (orderId: number) => {
  auditSink.write(orderId);
});

queue.runWorker(); // spawns the consumer's poll loop

// The payload type is explicit — Java can't infer it from the lambda.
flexiq.logConsumer("orders", "audit-log", Integer.class, orderId -> {
    auditSink.write(orderId);
});

flexiq.worker().start(); // spawns the consumer's poll loop

Tune the loop with poll_interval / pollIntervalMs (wait after an empty poll), batch_size / batchSize (messages per poll), and on_error / onError: "retry" (default) leaves a failed message un-acked so its batch re-reads, while "skip" acks past it and moves on. A non-empty batch drains immediately; only an empty one waits the poll interval.

Per-message ack

The cursor is a single high-water mark, so one message a consumer can't process blocks everything behind it. Per-message consumption is the alternative: instead of the cursor read, lease messages and ack or nack each one individually. A leased message is skipped by later lease reads on that same subscription until its visibility window elapses (a separate subscription consuming the same topic is unaffected); ack it when done, nack it to redeliver immediately, or let the lease time out to have it redelivered — all without blocking its siblings. Delivery is at-least-once: a timed-out lease redelivers, and two concurrent leasers on one subscription can briefly claim the same message, so handlers must be idempotent. It's a consumption choice on the same log subscription, not a separate registration.

for msg in queue.lease_topic("orders", "audit-log", visibility=30):
    try:
        audit_sink.write(msg.args)
        queue.ack_message("orders", "audit-log", msg.id)
    except TransientError:
        queue.nack_message("orders", "audit-log", msg.id)  # redeliver now

for (const msg of await queue.leaseTopic("orders", "audit-log", { visibility: 30 })) {
  try {
    auditSink.write(msg.args);
    await queue.ackMessage("orders", "audit-log", msg.id);
  } catch {
    await queue.nackMessage("orders", "audit-log", msg.id); // redeliver now
  }
}

for (TopicMessage msg : flexiq.leaseTopic("orders", "audit-log", 100, Duration.ofSeconds(30))) {
    try {
        auditSink.write(serializer.deserialize(msg.payload, Order.class));
        flexiq.ackMessage("orders", "audit-log", msg.id);
    } catch (TransientException e) {
        flexiq.nackMessage("orders", "audit-log", msg.id); // redeliver now
    }
}

Pick one style per subscription — mixing the cursor read and leasing on the same subscription lets their positions diverge. On every backend, a message every per-message consumer has acked is compacted like the cursor case, and its delivery state is dropped with it. A topic that mixes a per-message consumer with a plain cursor reader falls back to expires / retention-window cleanup.

Retention compacts, it doesn't expire

A log message is deleted once every log subscriber on its topic has acked past it — the retention sweep compacts up to the minimum cursor across all of a topic's log subscribers, on the same background cadence as job retention. A topic with an unread (or never-yet-read) log subscriber keeps its entire backlog; nothing is dropped just because it's old.

Operational notes

• A down fan-out subscriber doesn't lose messages. Its deliveries just accumulate as ordinary pending jobs in its queue until a worker resumes consuming it — the same backlog you'd see from any other stalled queue, and visible the same way (stats(), the dashboard). (A down log subscriber is the same story from the other side: its cursor just stops advancing.)
• Fan-out retention is normal job retention. A fan-out delivery is purged, replayed, or expired exactly like any other job, via result_ttl, purge_completed(), and the dead-letter queue. Log topics are the exception — see "Retention compacts, it doesn't expire" under Log topics above.
• Fan-out publish multiplies writes by subscriber count. A fan-out topic with 20 active subscribers turns every publish() call into 20 job rows. For a high-fanout topic, set an aggressive result_ttl on the publish (or on the subscriber tasks themselves) so completed deliveries don't pile up in storage between cleanup passes. (A log topic writes one row per publish regardless of subscriber count — this multiplication is a fan-out property only.)

See also

• Delivery guarantees — the at-least-once contract every delivery inherits.
• Idempotency — how dedup keys are derived and prioritized for a single enqueue.
• Serializers — the CborSerializer cross-SDK wire format used for cross-language topics.