# Events Store (Persistent Pub/Sub) (/integrations/faststream/how-to/events-store)



## Overview [#overview]

Events Store is KubeMQ's persistent pub/sub pattern → see
[Events Store](/learn/events-store) for how the broker persists and replays event streams.
Unlike plain [Events](/integrations/faststream/how-to/events), which are fire-and-forget,
messages published with `events_store=` are persisted on the broker and a subscriber can
replay the stored stream from any position.

In `kubemq-faststream`, you opt into Events Store by using the `events_store=` keyword on
`@broker.subscriber(...)` and `broker.publish(...)`, and you control where replay begins
with the `start_position` argument (and, where required, `start_value`).

```python
from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")


@broker.subscriber(
    events_store="audit-log",
    start_position=StartPosition.START_FROM_FIRST,
)
async def on_audit(msg: dict) -> None:
    print(f"Audit event: {msg}")
```

`StartPosition` is exported from the `kubemq_faststream` package alongside `KubeMQBroker`.
Publishing to an events-store channel uses the same `broker.publish(...)` call you use for
plain events, with `events_store=` in place of `events=`:

```python
await broker.publish({"action": "login", "user": "alice"}, events_store="audit-log")
```

## Start Positions [#start-positions]

`StartPosition` is a string enum with six values. Three of them (`START_AT_SEQUENCE`, `START_AT_TIME`, `START_AT_TIME_DELTA`) need a companion `start_value` to say *where* to start; the other three are positional and ignore `start_value`.

<TypeTable
  type="{
  START_FROM_NEW: {
    description: 'Only messages published after the subscription starts. This is the default — any historical messages in the store are skipped.',
    type: 'no start_value',
  },
  START_FROM_FIRST: {
    description: 'Replay every stored message from the beginning of the stream.',
    type: 'no start_value',
  },
  START_FROM_LAST: {
    description: 'Start from the last stored message.',
    type: 'no start_value',
  },
  START_AT_SEQUENCE: {
    description: 'Start at a specific sequence number.',
    type: 'requires start_value',
  },
  START_AT_TIME: {
    description: 'Start at an absolute Unix timestamp (seconds since epoch).',
    type: 'requires start_value',
  },
  START_AT_TIME_DELTA: {
    description: 'Start from N seconds ago.',
    type: 'requires start_value',
  },
}"
/>

If you omit `start_position`, the subscriber behaves as `START_FROM_NEW` and only sees messages published after it connects.

## START\_FROM\_NEW vs FIRST and LAST [#start_from_new-vs-first-and-last]

The three positional values cover the common cases: ignore history, replay all of it, or jump to the latest.

`START_FROM_NEW` skips everything already in the store and only delivers messages published after the subscription is active — it is the persistent-channel equivalent of plain Events behavior:

```python title="start_from_new.py"
import asyncio
import os

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)

CHANNEL = "example.events_store.start_from_new"


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_FROM_NEW,
    group="new-group",
)
async def handle_new_event(msg: dict) -> None:
    """Only receives events published after subscription starts."""
    print(f"[NEW] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 4):
        await broker.publish({"seq": i}, events_store=CHANNEL)
        print(f"Pre-subscription event {i} published (will NOT be received)")

    # Small delay to simulate a gap between historical and new events.
    await asyncio.sleep(1)

    for i in range(4, 6):
        await broker.publish({"seq": i}, events_store=CHANNEL)
        print(f"Post-subscription event {i} published")

    await asyncio.sleep(3)
    await app.stop()


if __name__ == "__main__":
    asyncio.run(app.run())
```

`START_FROM_FIRST` replays the entire stored stream, and `START_FROM_LAST` delivers only the most recent stored message. The example below registers both on the same channel — each subscriber uses its own `group` so they replay independently — then publishes a few events and lets them replay:

```python title="basic_positions.py"
import asyncio

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.events_store.positions"


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_FROM_FIRST,
    group="first-group",
)
async def from_first(msg: dict) -> None:
    """Replay every stored event from the beginning."""
    print(f"[FIRST] Received: {msg}")


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_FROM_LAST,
    group="last-group",
)
async def from_last(msg: dict) -> None:
    """Receive only the most recent stored event."""
    print(f"[LAST] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 4):
        await broker.publish(
            {"order_id": i, "item": f"widget-{i}"},
            events_store=CHANNEL,
        )
        print(f"Published event {i}")

    await asyncio.sleep(3)
    await app.stop()


if __name__ == "__main__":
    asyncio.run(app.run())
```

`START_FROM_FIRST` receives all three events; `START_FROM_LAST` receives only the latest.

## Resume from a Sequence Number [#resume-from-a-sequence-number]

Every message in an events-store channel has a monotonically increasing sequence number. `START_AT_SEQUENCE` with a `start_value` resumes delivery from that exact position — the canonical way to pick up where a consumer left off after a restart. Track the last sequence you processed, persist it, and pass it back as `start_value` on reconnect.

```python title="start_at_sequence.py"
import asyncio

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.events_store.at_sequence"
RESUME_AT = 3


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_AT_SEQUENCE,
    start_value=RESUME_AT,
)
async def handle_from_seq(msg: dict) -> None:
    """Process events starting from sequence 3."""
    print(f"[SEQ>={RESUME_AT}] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 6):
        await broker.publish(
            {"seq_demo": i, "data": f"payload-{i}"},
            events_store=CHANNEL,
        )
        print(f"Published event {i}")

    await asyncio.sleep(3)
    await app.stop()


if __name__ == "__main__":
    asyncio.run(app.run())
```

After publishing 5 events, the subscriber starting at sequence `3` receives only events 3, 4, and 5.

## Time-Based Replay [#time-based-replay]

Two start positions replay by time instead of by sequence:

* **`START_AT_TIME`** — `start_value` is an absolute **Unix timestamp** (seconds since epoch). Delivery begins at the first message stored at or after that moment.
* **`START_AT_TIME_DELTA`** — `start_value` is a **number of seconds to look back**. The broker translates "N seconds ago" into a starting point relative to now.

```python title="start_at_time.py"
import asyncio
import time

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.events_store.at_time"


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_AT_TIME_DELTA,
    start_value=60,
    group="delta-group",
)
async def from_last_60s(msg: dict) -> None:
    """Receive events published in the last 60 seconds."""
    print(f"[DELTA 60s] Received: {msg}")


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_AT_TIME,
    start_value=int(time.time()) - 30,
    group="abs-time-group",
)
async def from_absolute_time(msg: dict) -> None:
    """Receive events published since 30 seconds ago (absolute timestamp)."""
    print(f"[ABS TIME] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 4):
        await broker.publish(
            {"event_id": i, "ts": time.time()},
            events_store=CHANNEL,
        )
        print(f"Published event {i}")

    await asyncio.sleep(3)
    await app.stop()


if __name__ == "__main__":
    asyncio.run(app.run())
```

`START_AT_TIME_DELTA` with `start_value=60` replays the last minute; `START_AT_TIME` with `start_value=int(time.time()) - 30` replays from an absolute timestamp 30 seconds in the past.

## Consumer-Group Replay [#consumer-group-replay]

Start positions combine with consumer **groups** the same way they do for plain Events. When multiple subscribers join the same `group` on an events-store channel, messages are load-balanced across the group members rather than fanned out to all of them — while each member still honors the configured `start_position`. This lets you scale a replay across several workers that share the stored backlog.

```python title="consumer_group_replay.py"
import asyncio

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, StartPosition

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.events_store.group_replay"
GROUP = "workers"


@broker.subscriber(
    events_store=CHANNEL,
    group=GROUP,
    start_position=StartPosition.START_FROM_FIRST,
)
async def worker_a(msg: dict) -> None:
    """Worker A in the consumer group."""
    print(f"[Worker-A] Received: {msg}")


@broker.subscriber(
    events_store=CHANNEL,
    group=GROUP,
    start_position=StartPosition.START_FROM_FIRST,
)
async def worker_b(msg: dict) -> None:
    """Worker B in the consumer group."""
    print(f"[Worker-B] Received: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 7):
        await broker.publish(
            {"task_id": i, "action": f"process-{i}"},
            events_store=CHANNEL,
        )
        print(f"Published task {i}")

    await asyncio.sleep(3)
    await app.stop()


if __name__ == "__main__":
    asyncio.run(app.run())
```

The six published tasks are distributed across `Worker-A` and `Worker-B` instead of each worker receiving all six.

<Callout type="info">
  Like plain Events, Events Store is fire-and-forget delivery: `ack` / `nack` operations are **no-ops** for this pattern. Acknowledgement and requeue semantics apply only to [Queues](/integrations/faststream/how-to/queues). Replay position is controlled entirely by `start_position`, not by settlement.
</Callout>

## Related [#related]

<Cards>
  <Card title="Events (Pub/Sub)" href="/integrations/faststream/how-to/events" description="The non-persistent counterpart — fire-and-forget broadcast with optional group load balancing." />

  <Card title="Events Store concept" href="/learn/events-store" description="How KubeMQ persists and replays event streams at the broker level." />
</Cards>