fix: address review of pluggable recurrence kinds

Occurrence planning

- Send every occurrence a pass arrived in time for rather than one per pass, so
  a kind finer than the monitor interval is no longer throttled back to it. That
  is what the scheduling docs already promised.
- Treat a schedule's first occurrence like any later one. The null-priorRunAt
  exemption sent occurrences of any age under a policy documented as sending
  nothing for the ones that were missed.
- Warn (missed_occurrences_skipped) when the default policy writes a run off, so
  the drop is visible instead of silent.
- Cap a whole pass rather than only each schedule, so neither the insert nor the
  synchronous parser loop scales with the number of schedules owed a catch-up.
  Every schedule still gets the occurrence it is actually due.

Duplicate protection

- Give each forwarded job an id derived from the schedule and the occurrence, so
  a forward retried after a lost acknowledgement collapses on ON CONFLICT rather
  than becoming a second job.
- File a cron occurrence on a minute boundary in the one-per-minute slot a
  pre-40 release uses. Such an instance keeps running against a 40 schema, so
  without the slot a rolling upgrade double-sends every cron occurrence.

Pass mechanics

- Release the deployment-wide pass slot when a pass finds a due schedule of a
  kind it has no parser for, so an instance that has one reaches the occurrence
  while it is still on time instead of roughly one interval per instance later.
- Run every phase of a pass and always keep the warning bookkeeping. A failed
  insert used to discard it, re-emitting and re-persisting the pass's warnings
  every pass for as long as the failure lasted.
- Keep the repair write out of the parser try, so a lock timeout is no longer
  recorded as invalid_schedule and then suppressed for the life of the process.
- Measure repair staleness on updated_on, which the claim now bumps. last_run_at
  holds the occurrence, which after any outage is already outside the window.
- Write a whole pass's next occurrences in one statement instead of a round trip
  per row. The first pass after the migration has to anchor every schedule.
- Frame a first-occurrence parser failure as a problem with the expression.

Configuration

- missedGraceSeconds, maxCatchupOccurrences and scheduleRepairSeconds are
  constructor options; the former module constants are their defaults.

Also

- Migration 40 drops its columns with IF EXISTS, mirroring the install.
- Drop the unused lastRunAt projection from the repair query.
- Remove the stale schedule() heading and the cron argument name from the docs.
- Proxy contracts carry kind, expression, nextRunAt, lastRunAt and the missed
  policy, and accept the object recurrence form. This is what the Proxy CI
  typecheck was failing on.
- Replace the wall-clock races in the missed-occurrence tests with an hourly
  expression pinned half an hour from its boundary, which is what made
  "missed: once" fail intermittently on the pglite leg.

Author: kibertoad

docs/api/constructor.md

@@ -111,6 +111,18 @@ The following options can be set as properties in an object for additional confi
   })
   ```
 
+* **missedGraceSeconds**, int, default 60
+
+  How long after coming due an occurrence still counts as on time, in seconds. Anything older came due while no instance was claiming, and the schedule's `missed` policy decides its fate. Defaults to 60 seconds, or twice `cronMonitorIntervalSeconds` when that is longer. Raise it in a deployment whose passes routinely run late, so healthy occurrences are not written off as missed. See [Missed occurrences](./scheduling.md#missed-occurrences).
+
+* **maxCatchupOccurrences**, int, default 1000
+
+  The most occurrences one scheduling pass will send, per schedule and in total. `missed: 'all'` is unbounded by construction, so the remainder is dropped and reported as a `missed_occurrences_capped` warning.
+
+* **scheduleRepairSeconds**, int, default 300
+
+  How long a schedule may sit with no pending occurrence before a pass re-anchors it, in seconds. Covers rows carried over from a schema that stored no occurrence, and rows whose claiming process died before it could write the following occurrence back.
+
 * **migrate**, bool, default true
 
   If this is set to false, this instance will skip attempts to run schema migrations during `start()`. If schema migrations exist, `start()` will throw and error and block usage. This is an advanced use case when the configured user account does not have schema mutation privileges.

docs/api/events.md

@@ -35,7 +35,8 @@ boss.on('warning', ({ message, data }) => {
 | `index_bloat` | A job index is holding far more pages than its live entries need and was not rebuilt — because rebuilds are disabled, the connected role does not own the index, the index exceeds `maxIndexBytes`, or `REINDEX CONCURRENTLY` failed. The message names the reason. Emitted once per index rather than on every pass, and again if the condition returns after being cleared | `name`, `table`, `pages`, `entries`, `bytes`, `owned` |
 | `invalid_schedule` | A stored schedule could not be evaluated (for example an unusable `timezone` written by an older release, or a recurrence parser that answered with something other than a date) and was skipped for this scheduling pass; the remaining schedules are unaffected. Emitted once per broken schedule rather than on every pass, and again if the schedule is edited or the instance restarts | `queue`, `key`, `kind`, `expression`, `timezone` |
 | `unsupported_recurrence` | A schedule came due whose recurrence `kind` this instance has no parser for. The row is left untouched for an instance that has one. Emitted once per schedule rather than on every pass | `queue`, `key`, `kind`, `expression` |
-| `missed_occurrences_capped` | A `missed: 'all'` schedule was owed more than 1000 occurrences; the remainder were dropped | `queue`, `key`, `kind`, `expression`, `timezone`, `cap` |
+| `missed_occurrences_capped` | A schedule was owed more occurrences than one pass may send (`maxCatchupOccurrences`, 1000 by default); the remainder were dropped | `queue`, `key`, `kind`, `expression`, `timezone`, `cap` |
+| `missed_occurrences_skipped` | An occurrence came due more than `missedGraceSeconds` before any instance claimed it and the schedule's default `missed: 'skip'` policy sent nothing for it. Emitted once per schedule rather than on every pass | `queue`, `key`, `kind`, `expression`, `timezone`, `dueAt`, `graceSeconds` |
 
 ### Warning Persistence
 

docs/api/scheduling.md

@@ -4,6 +4,8 @@ Jobs may be created automatically on a recurring expression. As with other cron-
 
 Each schedule stores the moment its next occurrence is due. Every 30 seconds, one instance claims the occurrences that have come due and sends their jobs. Claiming the row is what keeps a multi-instance deployment from sending the same occurrence twice, so no throttling is involved and a recurrence kind with finer resolution than a minute is sent as often as its expression says.
 
+Each forwarded job carries an id derived from the schedule and the occurrence, so a forward that has to be retried collapses instead of becoming a second job. A cron occurrence on a minute boundary additionally carries the one-per-minute slot used by releases before schema 40, which is what keeps a rolling upgrade from double-sending: an instance still running the older code evaluates the same expression itself and lands in the same slot. There is exactly one such occurrence per minute, so that slot can never collapse two occurrences of the same schedule.
+
 To change how often occurrences are claimed, set `cronMonitorIntervalSeconds`. To change how often the claimed jobs are forwarded to their queues, set `cronWorkerIntervalSeconds`.
 
 To mitigate clock skew and drift, every 10 minutes the clock of each instance is compared to the database server's clock. The skew, if any, is stored and used as an offset when occurrences are computed, so all instances agree on when a schedule is due. The default clock monitoring interval can be adjusted with `clockMonitorIntervalSeconds`.
@@ -44,6 +46,8 @@ await boss.schedule('run-workflow',
 
 Parsers are pure functions that pg-boss calls; they are never stored or serialized. Only the kind and the expression reach the database, so an instance that has no parser for a stored kind leaves those schedules alone and emits a [`warning`](./events.md#warning) of type `unsupported_recurrence`, exactly as a queue with no `work()` handler is simply never fetched. Register the parser on at least one running instance and the schedule resumes.
 
+Only one instance runs a scheduling pass per `cronMonitorIntervalSeconds`, so an instance without the parser would otherwise spend the pass on a row it cannot evaluate and leave the occurrence to age out of the grace window. When a pass finds a due schedule of a kind it cannot evaluate, it releases the pass so the next instance to tick can try, and an instance that does have the parser gets there while the occurrence is still on time.
+
 The expression is stored in the `cron` column whatever the kind, and `getSchedules()` reports it as both `cron` and `expression`.
 
 ## Cron expressions
@@ -66,31 +70,32 @@ For more cron documentation and examples see the docs for the [cron-parser packa
 
 ## Missed occurrences
 
-If no instance is running when an occurrence comes due, the occurrence is missed. Because each schedule stores the occurrence it is waiting on, pg-boss knows exactly which ones were skipped, and the `missed` option decides what to do about them.
+An occurrence claimed within `missedGraceSeconds` of coming due was not missed, and is sent whatever the policy below says. That window defaults to 60 seconds, or twice `cronMonitorIntervalSeconds` when that is longer, and it is what lets a pass send every occurrence it arrived in time for rather than one per pass: a kind with second-level resolution gets all of them.
+
+Anything older came due while no instance was claiming, which is what the `missed` option decides the fate of. Because each schedule stores the occurrence it is waiting on, pg-boss knows exactly which ones those were.
 
 * **skip** (default)
 
-  Send nothing for them and resume at the next occurrence. This is how scheduling has always behaved.
+  Send nothing for them and resume at the next occurrence. A [`warning`](./events.md#warning) of type `missed_occurrences_skipped` names the schedule and the occurrence, so a drop is not silent.
 
 * **once**
 
   Send a single job, no matter how many occurrences were missed. Useful for a job that reconciles state: running it once brings everything up to date.
 
 * **all**
 
-  Send one job per missed occurrence, oldest first. Capped at 1000 per schedule, after which the remainder is dropped and a [`warning`](./events.md#warning) of type `missed_occurrences_capped` is emitted.
+  Send one job per missed occurrence, oldest first. Capped by `maxCatchupOccurrences` (1000 by default), after which the remainder is dropped and a [`warning`](./events.md#warning) of type `missed_occurrences_capped` is emitted. The cap applies to the pass as a whole as well as to each schedule, so a long catch-up cannot turn one pass into an unbounded insert; every schedule due in that pass still gets the occurrence it is actually due.
 
-An occurrence claimed within 60 seconds of coming due (or twice `cronMonitorIntervalSeconds`, whichever is longer) counts as on time and is sent whatever the policy. A schedule's first occurrence is also always sent, since `schedule()` applies the same window when it anchors the row.
+A schedule's first occurrence is treated exactly like any later one. `schedule()` anchors it one grace window back, so `0 3 * * *` created at 03:00:30 still sends immediately, but a schedule created while nothing was claiming gets whatever its policy says rather than an exemption.
 
-### `schedule(name, cron, data, options)`
 ### `schedule(name, recurrence, data, options)`
 
 Schedules a job to be sent to the specified queue on a recurring expression. If the schedule already exists, it's updated to the new expression.
 
 **Arguments**
 
 - `name`: string, *required*
-- `cron`: string, *required*. A cron expression, or `{ kind, expression }` for a registered recurrence kind
+- `recurrence`: string or object, *required*. A cron expression, or `{ kind, expression }` for a registered recurrence kind
 - `data`: object
 - `options`: object
 

docs/sql/warning-table.md

@@ -15,7 +15,7 @@ CREATE TABLE pgboss.warning (
 | Column | Description |
 |--------|-------------|
 | `id` | Auto-incrementing primary key |
-| `type` | Warning type: `slow_query`, `queue_backlog`, `clock_skew`, `listen_notify_unavailable`, `index_bloat`, `invalid_schedule`, `unsupported_recurrence`, or `missed_occurrences_capped` |
+| `type` | Warning type: `slow_query`, `queue_backlog`, `clock_skew`, `listen_notify_unavailable`, `index_bloat`, `invalid_schedule`, `unsupported_recurrence`, `missed_occurrences_capped`, or `missed_occurrences_skipped` |
 | `message` | Human-readable warning message |
 | `data` | JSON object with warning-specific details |
 | `created_on` | Timestamp when the warning was recorded |

packages/proxy/src/contracts.ts

@@ -88,6 +88,7 @@ export const updateOptionsSchema = z.object({
 export const scheduleOptionsSchema = sendOptionsSchemaBase.extend({
   tz: z.string().optional(),
   key: z.string().optional(),
+  missed: z.enum(['skip', 'once', 'all']).optional(),
 }) satisfies z.ZodType<types.HttpScheduleOptions>
 
 export const fetchOptionsSchema = z.object({
@@ -245,10 +246,15 @@ export const queueResultSchema = z.object({
 export const scheduleSchema = z.object({
   name: z.string(),
   key: z.string(),
+  kind: z.string(),
+  // the expression lives in the cron column whatever the kind, so both names are reported
+  expression: z.string(),
   cron: z.string(),
   timezone: z.string(),
   data: jsonRecordSchema.optional(),
-  options: sendOptionsSchema.optional(),
+  options: scheduleOptionsSchema.optional(),
+  nextRunAt: z.iso.datetime().nullable().transform((val) => val ? new Date(val) : null),
+  lastRunAt: z.iso.datetime().nullable().transform((val) => val ? new Date(val) : null),
 }) satisfies z.ZodType<types.HttpSchedule>
 
 export const bamStatusSummarySchema = z.object({
@@ -611,9 +617,16 @@ export const schemaVersionResponseSchema: z.ZodType<types.HttpSchemaVersionRespo
   result: z.number().nullable()
 })
 
+export const recurrenceSchema = z.object({
+  kind: z.string(),
+  expression: z.string(),
+}) satisfies z.ZodType<types.HttpRecurrence>
+
 export const scheduleRequestSchema: z.ZodType<types.HttpScheduleRequest> = z.object({
   name: queueNameSchema,
-  cron: z.string(),
+  // A bare string is cron, which is what every caller written before kinds existed sends, so the
+  // field keeps its name. The object form names a kind registered on the server's constructor.
+  cron: z.union([z.string(), recurrenceSchema]),
   data: nullableJsonRecordSchema.optional(),
   options: scheduleOptionsSchema.optional()
 })

packages/proxy/src/types.ts

@@ -76,7 +76,7 @@ export type HttpQueueResult = types.QueueResult
 
 export type HttpSchedule = Omit<types.Schedule, 'data' | 'options'> & {
   data?: HttpJsonRecord
-  options?: HttpSendOptions
+  options?: HttpScheduleOptions
 }
 
 export type HttpBamStatusSummary = types.BamStatusSummary
@@ -358,9 +358,11 @@ export type HttpSchemaVersionResponse = {
   result: number | null
 }
 
+export type HttpRecurrence = types.Recurrence
+
 export type HttpScheduleRequest = {
   name: HttpQueueName
-  cron: string
+  cron: string | HttpRecurrence
   data?: HttpNullableJsonRecord
   options?: HttpScheduleOptions
 }

src/attorney.ts

@@ -785,6 +785,23 @@ function applyScheduleConfig (config: any) {
 
   config.cronWorkerIntervalSeconds = config.cronWorkerIntervalSeconds || 5
 
+  assert(!('missedGraceSeconds' in config) || config.missedGraceSeconds >= 1,
+    'configuration assert: missedGraceSeconds must be at least 1 second')
+
+  // Derived rather than fixed: a pass that cannot run more often than the monitor interval cannot
+  // be expected to claim inside a window narrower than two of them.
+  config.missedGraceSeconds = config.missedGraceSeconds || Math.max(60, config.cronMonitorIntervalSeconds * 2)
+
+  assert(!('maxCatchupOccurrences' in config) || config.maxCatchupOccurrences >= 1,
+    'configuration assert: maxCatchupOccurrences must be at least 1')
+
+  config.maxCatchupOccurrences = config.maxCatchupOccurrences || 1000
+
+  assert(!('scheduleRepairSeconds' in config) || config.scheduleRepairSeconds >= 1,
+    'configuration assert: scheduleRepairSeconds must be at least 1 second')
+
+  config.scheduleRepairSeconds = config.scheduleRepairSeconds || 300
+
   if ('recurrences' in config) {
     assertRecurrenceConfig(config.recurrences)
   }

src/migrationStore.ts

@@ -1486,10 +1486,14 @@ function getAll (schema: string, noPartitioning = false, noCovering = false): ty
         `ALTER TABLE ${schema}.schedule ADD COLUMN IF NOT EXISTS next_run_at timestamp with time zone`,
         `ALTER TABLE ${schema}.schedule ADD COLUMN IF NOT EXISTS last_run_at timestamp with time zone`
       ],
+      // IF EXISTS mirrors the install's IF NOT EXISTS: the adds are idempotent because the install
+      // is expected to survive being re-run over a partial state, and a rollback of that same
+      // partial state has to survive a column that never got added. Without it, a process that died
+      // between the adds leaves a schema that can be neither completed nor rolled back.
       uninstall: [
-        `ALTER TABLE ${schema}.schedule DROP COLUMN kind`,
-        `ALTER TABLE ${schema}.schedule DROP COLUMN next_run_at`,
-        `ALTER TABLE ${schema}.schedule DROP COLUMN last_run_at`
+        `ALTER TABLE ${schema}.schedule DROP COLUMN IF EXISTS kind`,
+        `ALTER TABLE ${schema}.schedule DROP COLUMN IF EXISTS next_run_at`,
+        `ALTER TABLE ${schema}.schedule DROP COLUMN IF EXISTS last_run_at`
       ]
     }
   ]