Schedules

A schedule is a producer on a cron expression. You register it with a name, an expression, the stream it produces to, and the message it produces; the system produces that message onto that stream every time the expression comes due. Consuming it is a plain consumer group on that stream, and there can be any number of them - the schedule knows nothing about who listens.

type InvoiceRun struct {
	Region string `json:"region"`
}

func (InvoiceRun) SchemaVersion() int { return 1 }

nightly, err := client.Scheduler("invoices.nightly").Register(ctx,
	"invoices", "0 2 * * *",
	&InvoiceRun{Region: "eu"}, nil)
if err != nil {
	return err
}

return nightly.Schedule(ctx)

And in whichever service does the invoicing, nothing about cron:

invoices, err := client.Stream[InvoiceRun]("invoices").Consumer("invoice-runner").Register(ctx, nil)
if err != nil {
	return err
}

return invoices.Consume(ctx, func(ctx context.Context, run *InvoiceRun) error {
	meta, _ := sqlstreams.MetaFromContext(ctx)
	fmt.Printf("invoicing %s for %s\n", run.Region, meta.ScheduledAt.Format("2006-01-02"))
	return nil
}, nil)

This is the same shape as the other two register verbs: declare a named resource on the client, run what comes back. For a producer - Producer().Register resolves the stream and Produce writes. For a consumer - Consumer(name).Register declares the group and Consume runs it. For a schedule - Scheduler(name).Register declares it and returns a SchedulerInstance; its Schedule keeps the system producing it. The Scheduler handle administers the stored schedule.

What Register stores

Register is a declaration, and the newest one wins: calling it again with a different expression or payload updates the row, calling it with the same values changes nothing. The row is one schedule_config entry:

column	from
name	the first argument; also the message key every produce carries
expression	the cron expression, UTC unless it carries TZ=
stream_id	the target stream, resolved by name at Register
payload	the message, marshaled once
schema_version	InvoiceRun.SchemaVersion(), read from the type argument
concurrency, timeout_ns	the config’s Concurrency and Timeout

The type argument is why the payload is safe to store as JSON: the schema version is captured at declaration, and every produce carries it on the row, so a consumer registered with InvoiceRun accepts the message the same way it accepts one from a live producer. Bump SchemaVersion and re-Register and the next produce is version 2; consumers still on version 1 skip it, as on any stream (schema versions).

The payload is the same document every time. A handler that needs to know which scheduled time a message is for reads meta.ScheduledAt; it is not in the payload because the payload never changes.

What SchedulerInstance.Schedule does

SchedulerInstance.Schedule runs the system manager, the same thing sqlstreams manager run does. The schedule producer is a system worker; it is what actually produces, and it produces every registered schedule, not just the one whose instance called Schedule. So SchedulerInstance.Schedule exists for one reason: a program that registers a schedule and exits has registered one that never fires unless a manager is running somewhere. Calling SchedulerInstance.Schedule makes the registering program that somewhere — and so does any Consume in the deployment, since a consumer runs the manager too. SchedulerInstance.Schedule is one more caller of it, and the manager row’s target_instances decides which caller holds the claim.

The produce itself is what a hand-written producer would do, in one transaction with the schedule_cursor advance:

• MessageKey is the schedule name, with compaction on, so a due message nobody claimed yet is superseded by the next one instead of piling up.
• Concurrency and Timeout come from the schedule’s config, so exclusive means a message waits for the previous one to finish.
• The idempotency key is derived from the scheduled time and the schedule id: a replay after an ambiguous commit dedupes, every other produce lands.
• After downtime the schedule producer produces the newest due time only; older ones are dropped, not produced late.

Worked case

invoices is stream id 4. Scheduler("invoices.nightly").Register[InvoiceRun] writes:

SELECT id, name, stream_id, expression, payload, schema_version
FROM sqlstreams.schedule_config;
--  id | name             | stream_id | expression | payload          | schema_version
--  3  | invoices.nightly | 4        | 0 2 * * *  | {"region":"eu"}  | 1

At 02:00 UTC the schedule producer claims the row, produces onto message_log_4, and advances schedule_cursor to the next 02:00:

SELECT id, routing_key, message_key, schema_version, payload
FROM sqlstreams.message_log_4
ORDER BY id DESC LIMIT 1;
--  id     | routing_key | message_key      | schema_version | payload
--  88012  |             | invoices.nightly | 1              | {"region":"eu"}

invoice-runner claims id 88012 like any other message; a second group invoice-audit on the same stream claims it too. Neither group was named anywhere in the schedule.

Admin verbs

Everything an operator does to a schedule needs no type argument and no running process, so it stays on the Scheduler handle and the CLI, one noun on both:

client	CLI	what it does
Scheduler(name).Run	sqlstreams scheduler run	produces one message now, off-schedule, from the stored payload
Suspend / Unsuspend	sqlstreams scheduler suspend / unsuspend	stops and resumes the schedule; a time that came due while suspended is dropped
Status, Messages	sqlstreams scheduler status / messages	per-group outcomes, read from the target stream’s delivery_log by message key; a succeeded count needs the target stream’s DeliveryLogMode: all (the default keeps failures only) - Register warns SQL0058 when it isn’t
Destroy	sqlstreams scheduler destroy	deletes the row; gated by AllowDestroy

Temporal and EventBridge Scheduler use the same noun for the same thing; “cron job” stays the name of K8s’s resource, where what fires is a batch job and not a message.

The message key on a shared stream

The schedule’s name is its message key on the target stream. On a stream where your producers also key messages, a schedule named the same as one of your keys would supersede it. Give a schedule its own stream, or key your own messages so the two can’t collide - I’m not sure a config knob here is worth its weight until someone hits it.

What this is not

A one-off “run this in ten minutes” is not a schedule; it is a delayed delivery, and a schedule does not do it. A schedule repeats, and its unit is a message.