Java/Guides/Tasks & execution/

Topic Pub/Sub

Fan a published message out to N independent subscribers — each delivered as its own ordinary job, so retries, DLQ, and middleware apply per subscriber.

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, rate limits, whatever hooks your binding has — 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.

Subscribe with the @queue.subscriber() decorator.Subscribe with queue.subscriber().Subscribe with flexiq.subscribe().The Rust SDK does not wrap pub/sub: subscribe with queue.storage().register_subscription() and publish with flexiq::pubsub::publish_to_topic().

Quick start

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

python

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

java

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
}

The Rust SDK does not wrap pub/sub. The registry lives on the Storage trait and the fan-out itself is flexiq::pubsub, both reached through queue.storage():

rust

use flexiq::pubsub::{publish_to_topic, DeliveryDefaults, PublishRequest};
use flexiq::storage::records::SubscriptionMode;
use flexiq::wire::{encode_call, WireValue};
use flexiq::{now_millis, FlexiQ, NewSubscription, Storage};

#[flexiq::task]
fn send_confirmation_email(order_id: i64) -> flexiq::Outcome<()> { Ok(()) }

#[flexiq::task]
fn track_order(order_id: i64) -> flexiq::Outcome<()> { Ok(()) }

let queue = FlexiQ::open("orders.db")?;

// A subscription is a registry row: (topic, name) -> (task, queue).
for (name, task) in [
    ("email", "send_confirmation_email"),
    ("analytics", "track_order"),
] {
    queue.storage().register_subscription(&NewSubscription {
        topic: "orders".into(),
        subscription_name: name.into(),
        task_name: task.into(),
        queue: "default".into(),
        active: true,
        durable: true,
        owner_worker_id: None,
        created_at: now_millis(),
        priority: None,
        max_retries: None,
        timeout_ms: None,
        mode: SubscriptionMode::Fanout,
    })?;
}

// The payload is one call envelope, shared by every delivery.
let jobs = publish_to_topic(
    queue.storage(),
    &PublishRequest {
        topic: "orders".into(),
        payload: encode_call(&[WireValue::Integer(42)], &[]),
        idempotency_key: None,
        metadata: None,
        notes: None,
        priority: None,
        scheduled_at: now_millis(),
        max_retries: None,
        timeout_ms: None,
        expires_at: None,
        result_ttl_ms: None,
        namespace: queue.namespace().map(str::to_string),
        queue_defaults: DeliveryDefaults {
            priority: 0,
            max_retries: 3,
            timeout_ms: 300_000,
        },
    },
)?; // -> Vec<Job> — one per subscriber

register_subscription writes the row immediately, so a producer-only process needs no separate "declare" step. The subscriber tasks themselves are ordinary tasks: register them on a worker with register::<send_confirmation_email>() like any other.

Every field of PublishRequest is required because it has no Default — build it once in a helper of your own rather than at each call site.

subscribe() only wires the routing — it doesn't take a handler. Register the handler itself on the worker the same way you would for any other task:

java

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

declare_subscriptions() writes durable subscriptions to storage immediately — call it in a producer-only process (one that imports subscriber modules but never runs a worker) so publish() sees them. run_worker() does this automatically at startup, so a process that both subscribes and runs a worker doesn't need the explicit call.

declareSubscriptions() writes durable subscriptions to storage immediately — call it in a producer-only process (one that registers subscribers but never runs a worker) so publish() sees them. runWorker() does this automatically at startup, so a process that both subscribes and runs a worker doesn't need the explicit call.

register_subscription is idempotent on (topic, subscription_name), so calling it again with the same pair updates that row rather than adding a second subscriber. The pair is the identity; task_name only says where deliveries go, and changing it re-points an existing subscription at a different task.

priority, max_retries and timeout_ms on the row are Options — None takes the publisher's queue_defaults, and a value overrides it for this subscriber alone, so two subscribers on one topic can have different retry budgets. durable: false together with an owner_worker_id is the ephemeral form: the row is reaped once that worker stops heartbeating.

NewSubscription derives no Default, so all twelve fields are spelled out at every call site. Write one constructor of your own and register through it.

A durable subscribe() call registers immediately — there's no separate "declare" step. A producer-only process just needs to call subscribe() before it calls publish().

There is no declare step here either: the row is written when register_subscription returns, so a producer-only process just registers before it publishes.

What such a process cannot do is serve the subscription. A fan-out delivery is an ordinary job, so it sits pending until a worker that registered the subscriber task claims it. A worker serving that queue without the task registered is worse than none at all — it claims the delivery and fails it fatally with task not registered on this worker.

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.

python

@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

java

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()) {
    ...
}

Ephemeral is two fields on the row, not a flag with machinery behind it: an owner, and durable: false. Name the worker so the owner matches the process that will be reaped.

rust

queue.storage().register_subscription(&NewSubscription {
    topic: "orders".into(),
    subscription_name: "debug-tail".into(),
    task_name: "tail".into(),
    queue: "default".into(),
    active: true,
    durable: false,
    owner_worker_id: Some("w-1".into()), // reaped when w-1 stops heartbeating
    created_at: now_millis(),
    priority: None,
    max_retries: None,
    timeout_ms: None,
    mode: SubscriptionMode::Fanout,
})?;

let worker = queue.worker().register::<tail>().worker_id("w-1").spawn()?;

Nothing registers the row for you here, so write it before the worker starts. shutdown() sweeps this worker's ephemeral rows immediately rather than waiting for a peer's heartbeat tick.

Reaping is automatic

Every SDK reaps ephemeral subscriptions on the worker heartbeat cadence, so a crashed or stopped worker's ephemeral rows disappear without operator intervention.

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.

python

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

java

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

rust

let store = queue.storage();

store.set_subscription_active("orders", "email", false)?; // stop deliveries
publish_to_topic(store, &request("orders", 1))?;          // only "analytics" gets one
store.set_subscription_active("orders", "email", true)?;

store.unsubscribe("orders", "email")?;          // remove it entirely; false if unknown
store.list_subscriptions_for_topic("orders")?;  // active subscriptions for a topic
store.list_subscriptions()?;                    // every row, active or paused, all topics

There is no "list the topics that have a subscriber" call: derive it from list_subscriptions(). list_declared_topics() is a different thing — the registry of topics someone declared, covered under Declared topics.

Late-join boundary

No replay

There's no history to replay. A subscription only receives messages published after it was registered — publishing before a subscriber exists, or while it's unsubscribed, permanently misses that subscriber. If you need durable event history, that's a different problem than pub/sub fan-out; this feature is about delivering live messages to every current subscriber, each with full task semantics.

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:

An explicit option passed to publish().
The subscriber task's own registered default.
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.

python

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

java

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

rust

let keyed = PublishRequest {
    idempotency_key: Some("evt-42".into()),
    ..request("orders", 1)  // a helper of your own; PublishRequest has no Default
};

let first = publish_to_topic(queue.storage(), &keyed)?;  // 2 subscribers -> 2 jobs
let second = publish_to_topic(queue.storage(), &keyed)?; // same jobs, no new rows
assert_eq!(first[0].id, second[0].id);

// Add a third subscription named "audit", then republish the same key.
queue.storage().register_subscription(&audit)?;
let third = publish_to_topic(queue.storage(), &keyed)?;  // "audit" gets its own delivery

The salting is core's, not the caller's: the key becomes evt-42::6:5:ordersemail per subscription before it reaches the job row. The jobs unique index is global, so an unsalted key would let the first insert win and silently drop every other delivery — which is why this lives below the shells rather than in one.

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.

python

(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" }

java

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}

rust

let jobs = publish_to_topic(
    queue.storage(),
    &PublishRequest {
        notes: Some(r#"{"tenant":"acme"}"#.into()),
        ..request("orders", 5)
    },
)?;

let notes: serde_json::Value = serde_json::from_str(jobs[0].notes.as_deref().unwrap())?;
// notes["topic"] == "orders", notes["subscription"] == "email", notes["tenant"] == "acme"

notes is an already-encoded JSON object string on the way in and on the way out — Job::notes is Option<String>, so parse it yourself. The two stamped keys are merged into whatever object you supplied.

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.

Cross-SDK payload contract

When a topic's publisher and its subscribers aren't all the same language, every process that touches the topic has to be on CborSerializer — configure it wherever the binding lets you choose. It's the wire format the cross-SDK contract is built on, and it round-trips the types JSON can't (big integers, datetime/Date, bytes, decimals) losslessly. A single object/dict argument maps most cleanly onto every language's handler signature.

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.

Register a log subscription with queue.subscribe_log(), pull with queue.read_topic(), and advance its cursor with queue.ack_topic().Register a log subscription with queue.subscribeLog(), pull with queue.readTopic(), and advance its cursor with queue.ackTopic().Register a log subscription with flexiq.subscribeLog(), pull with flexiq.readTopic(), and advance its cursor with flexiq.ackTopic().Register a log subscription with register_subscription() and SubscriptionMode::Log, pull with read_topic_messages(), and advance its cursor with ack_topic_cursor() — all on queue.storage().

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.

python

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);
}

java

// 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);
    }
}

A log subscription is the same registry row with mode: SubscriptionMode::Log. It routes to no task, so task_name is empty.

rust

queue.storage().register_subscription(&NewSubscription {
    topic: "orders".into(),
    subscription_name: "audit-log".into(),
    task_name: String::new(), // a log subscription has no handler
    queue: "default".into(),
    active: true,
    durable: true,
    owner_worker_id: None,
    created_at: now_millis(),
    priority: None,
    max_retries: None,
    timeout_ms: None,
    mode: SubscriptionMode::Log,
})?;

let jobs = publish_to_topic(queue.storage(), &request("orders", 42))?;
assert!(jobs.is_empty()); // no fan-out jobs; one message stored

// Read, process, then ack — the cursor only moves on the explicit ack.
for msg in queue.storage().read_topic_messages("orders", "audit-log", 100)? {
    audit_sink.write(&msg.payload)?;
    queue.storage().ack_topic_cursor("orders", "audit-log", &msg.id)?;
}

TopicMessage::payload is the raw envelope bytes the publisher wrote, not a decoded value, and the Rust SDK exposes no reader for it — the one it uses for task arguments is private to the crate. Read it yourself: drop the leading flexiq::wire::TAG_CBOR byte, then decode the remaining two-element [args, kwargs] array with any CBOR implementation.

TopicMessage.payload is raw byte[], not a decoded value — Java is statically typed, so there's no target type to infer it into the way readTopic() infers one for Python/Node. Decode it yourself with the same serializer the publisher used. metadata and notes are still decoded to Map<String, Object>, same as the fan-out Job type.

A TopicMessage is id, topic, payload, metadata, notes, created_at and expires_at. Only payload is opaque; metadata and notes stay the JSON strings the publisher wrote, undecoded, exactly as they are on a Job.

id is opaque and backend-specific — a UUIDv7 on the Diesel backends, a stream id on Redis. It is time-ordered, and it doubles as the cursor token, so pass it back rather than parsing it.

read_topic_messages(topic, subscription_name, limit) resolves the cursor server-side from the subscription row and reads exclusive of it, so an empty Vec means this consumer is caught up.

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.

python

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

java

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

rust

// Retention is milliseconds, and `Log` is the only declarable mode.
queue
    .storage()
    .declare_topic("orders", SubscriptionMode::Log, Some(3_600_000))?;

publish_to_topic(queue.storage(), &request("orders", 42))?; // retained, no subscriber yet
queue.storage().register_subscription(&audit_log)?;               // the Log row from above
queue.storage().read_topic_messages("orders", "audit-log", 100)?; // sees the earlier publish

Declaring is an idempotent upsert on the name: re-declaring updates the retention window and keeps the original created_at. Passing SubscriptionMode::Fanout, or a negative retention, is rejected by every backend.

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.

python

@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

java

// 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

The Rust SDK has no managed consumer. Nothing on FlexiQ or WorkerBuilder registers one, and spawn() starts no poll loop for a log subscription — the read/ack loop under Log topics is the whole surface, and driving it is yours: a thread, a tokio task, or a call on whatever interval suits the consumer.

The delivery guarantee is unchanged either way, because it is the cursor's: read, process, then ack_topic_cursor.

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.

python

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
  }
}

java

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
    }
}

rust

let store = queue.storage();

// limit, visibility window in ms, and the `now` the lease is measured from.
for msg in store.lease_topic_messages("orders", "audit-log", 100, 30_000, now_millis())? {
    match audit_sink.write(&msg.payload) {
        Ok(()) => store.ack_message("orders", "audit-log", &msg.id)?,
        Err(_) => store.nack_message("orders", "audit-log", &msg.id)?, // redeliver now
    };
}

now is a parameter rather than a clock read inside the call, so a test can drive the visibility window without sleeping. Both ack_message and nack_message return false when there was no un-acked delivery to act on.

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.

Redis note

On the Redis backend, log topics are backed by a Redis Stream — an implementation detail, not a different API. read_topic() / readTopic() and ack_topic() / ackTopic() behave identically across SQLite, Postgres, and Redis.

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.)