natsio-schedules

JetStream message schedules (ADR-51) for natsio: schedule-expression builders, a scheduled-publish helper, and read/cancel operations over the definitions a stream holds. Distribution natsio-schedules, imported as natsio.schedules. Zero runtime dependencies beyond natsio. Pre-1.0, no API-stability promises.

Requires nats-server 2.12+ for the feature itself, 2.14+ for cron time zones and Nats-Schedule-Rollup. Verified against the pinned 2.14.3.

What it is

A stream created with allow_msg_schedules can hold schedule definitions: ordinary stored messages carrying a Nats-Schedule header. The server republishes each definition's body to its Nats-Schedule-Target on the schedule — a one-shot delayed publish, a repeating interval, a cron expression, or a periodic sample of another subject's latest message.

A definition's subject is its identity. The server stores it as a Nats-Rollup: sub message, so there is exactly one schedule per subject and re-publishing replaces it.

Generated messages are published inside the stream. A plain core-NATS subscription on the target subject will not see them (unless the stream has republish configured). Consume them with a JetStream consumer.

Usage

from datetime import timedelta

import natsio
from natsio.schedules import (
    HOURLY,
    ScheduleStreamConfig,
    after,
    create_schedule_stream,
    cron,
    delivery_info,
    every,
)

nc = await natsio.connect("nats://localhost")
js = nc.jetstream()

# `subjects` must cover the schedule subjects AND every target subject.
sched = await create_schedule_stream(
    js, ScheduleStreamConfig(name="SCHED", subjects=["schedules.>", "orders.>"])
)

# One-shot: publish `orders.reminder` five minutes from now, then self-destruct.
await sched.create(
    "schedules.orders.r1", after(timedelta(minutes=5)), target="orders.reminder", payload=b"ping"
)

# Repeating, with a TTL on each generated message.
await sched.create(
    "schedules.heartbeat", every(timedelta(seconds=30)), target="orders.tick", ttl="5m"
)

# Cron (6 fields: sec min hour dom month dow), evaluated in a named zone.
await sched.create(
    "schedules.report", cron("0 0 5 * * *"), target="orders.report", time_zone="Europe/Amsterdam"
)
await sched.create("schedules.hourly", HOURLY, target="orders.hourly")

# Subject sampling: republish the latest `sensors.raw` reading every minute.
await sched.create(
    "schedules.sample", every("1m"), target="orders.sampled", source="sensors.raw"
)

# Inspect / enumerate / cancel.
entry = await sched.get("schedules.heartbeat")
print(entry.schedule, entry.target, entry.interval)

async for entry in sched.list("schedules.>"):
    print(entry.subject, entry.schedule)

await sched.cancel("schedules.heartbeat")

On the consuming side:

consumer = await sched.stream.create_consumer(ConsumerConfig(filter_subject="orders.>"))
msg = await consumer.next()

info = delivery_info(msg)     # None if this wasn't produced by a schedule
if info is not None:
    print(info.scheduler)     # "schedules.heartbeat" — the definition's subject
    print(info.next_run)      # next firing, or None ...
    print(info.final)         # ... when this was the last one (`Nats-Schedule-Next: purge`)

Cancelling atomically

ADR-51's "advanced" stop: remove the schedule only if a message on a different subject is persisted — the way to fire a delayed publish early without letting the schedule also fire it, or to record a cancellation where consumers can see it. CAS-gated on the definition's sequence, so a schedule that already fired is never "cancelled" twice.

entry = await sched.get("schedules.orders.r1")
await sched.stop_and_publish(
    "schedules.orders.r1",
    publish_to="orders.reminder",        # or any other subject except the schedule's own
    payload=b"sent early",
    expected_schedule_seq=entry.sequence,
)

Without an explicit expected_schedule_seq the definition is read first and its sequence used as the gate, so stopping a schedule that already fired (or never existed) raises ScheduleNotFoundError instead of publishing the message and cancelling nothing — the server accepts that no-op happily. Pass require_existing=False for a one-round-trip, fire-and-forget stop.

cancel() is the plain version: it reads the definition, refuses subjects that hold something other than a schedule, and purges only up to the message it read. cancel_many() is not guarded — it is a raw wildcard purge and will happily delete generated messages and ordinary traffic under the filter.

API

Errors

Everything roots at ScheduleError (a natsio.jetstream.JetStreamError), in two branches:

• Local, pre-flight — ScheduleExpressionError, ScheduleTargetError, ScheduleSourceError, ScheduleTimeZoneError, ScheduleTTLError, SchedulesNotEnabledError. All also subclass natsio.errors.ConfigError, so they are ValueErrors. (ScheduleNotFoundError covers "no definition there".)
• Server-reported — every ADR-51 err_code, bound to a typed error through the core's register_error hook (registration happens on import natsio.schedules). They are also plain APIErrors, so existing except APIError code keeps working.

Wire contract

Two things the table cannot show:

• Nats-Schedule-Next is stamped in the schedule's own time zone when Nats-Schedule-Time-Zone is set (2026-07-23T02:56:46+05:30), not in UTC. delivery_info keeps the offset, so ScheduleDelivery.next_run is an aware datetime in that zone — compare instants, not wall clocks.
• ScheduleEntry.headers holds the stored block only. Direct Get adds Nats-Stream / Nats-Subject / Nats-Sequence / Nats-Time-Stamp (and, on batch replies, Nats-Num-Pending / Nats-Last-Sequence); those are stripped (TRANSPORT_HEADERS), so an entry reads the same from get() and list() and its headers can be handed straight back to create().

Pinned against nats.go jetstream/message.go (header constants), jetstream/jetstream_options.go (WithScheduleAt / WithScheduleEvery / WithScheduleCron / WithScheduleTTL / WithScheduleTimeZone value formatting), jetstream/errors.go (err_codes) and ADR-51 itself; every accept/reject boundary is additionally probed against the pinned 2.14.3 server in TestGrammarParity. Note that Nats-Schedule-Rollup (ADR-51 rev 4) and the @annually / @midnight aliases exist on the 2.14.3 server but not yet in nats.go — for those the ADR and the server are the oracle.

Scope limits

• No local next-fire calculation. Cron/interval evaluation lives in the server; this package never predicts when a schedule will run. Read the server's own answer from Nats-Schedule-Next on a delivered message.
• The cron validator is a pre-flight, not a re-implementation. It matches the server on everything the parity suite covers (field count, ranges, names, steps, ?, @ aliases, 0-6 day-of-week, a leading */? swallowing the rest of a range, and day-of-month/month pairs that can never occur) and is stricter in exactly one place: an empty Nats-Schedule is rejected here, while the server accepts it and simply stores a normal message. Two known gaps in the other direction — cases the server rejects and this validator lets through, so they come back as an APIError rather than a local one: a cron expression whose only firing is further out than the server's search horizon (e.g. 0 0 0 29 2 * evaluated in the late 2090s), and anything else its evaluator dislikes. The server stays the authority.
• No stream-lifecycle management beyond creation. Enabling allow_msg_schedules on an existing stream is a core update_stream call; this package will not do it implicitly.
• list() requires allow_direct (set by create_schedule_stream). It is one batch Direct Get over the stream's subjects, filtered to messages that actually carry Nats-Schedule. A single request cannot answer more than MAX_SUBJECTS_PER_BATCH (1024) matching subjects — the server refuses the whole thing with 413 Too Many Results — so above that list() reads the matching subjects from STREAM.INFO and fetches them page by page. That is transparent but not free: narrow the filter (list("schedules.>")) on streams that also carry high-cardinality target subjects to stay on the single-request path. Enumeration is complete or it raises; a truncated read is never returned as if it were the whole set.
• Retention interactions are the operator's problem. ADR-51's WorkQueue / Interest caveats (a consumer ack can silently delete a schedule) are documented in the ADR, not enforced here; ScheduleStreamConfig defaults to Limits, which is the recommended policy.

Example

A runnable script is at examples/basic.py — start a server with just server, then:

python extensions/natsio-schedules/examples/basic.py