fix(schedule): address review of lastJobId

proxy: scheduleSchema gains createdOn, updatedOn and lastJobId, which
types.Schedule now requires. Without them the workspace failed to
typecheck and the new fields were missing from the OpenAPI contract.

timekeeper: reduce a batch to one occurrence per (name, key) before the
UPDATE. Postgres leaves the choice of source row unspecified when
several join one target, so a worker that picked up two minute buckets
for one schedule could store the older job id.

timekeeper: report send-it failures through a helper that cannot throw.
An 'error' event with no listener is a throw by Node's contract, and
index.ts re-promotes this one onto the PgBoss instance, so the report
could escape the handler, fail the send-it job and replay the batch.
The annotation failure now names the schedules it covers, which a bare
driver error did not distinguish from a forwarding failure.

timekeeper: the send-it singleton key is JSON.stringify([name, key])
rather than `${name}__${key}`. Underscores are legal in both a queue
name and a schedule key, so ('report_', 'daily') and ('report',
'_daily') collided and the 60s window dropped one of them. A
mixed-version deployment can fire a schedule twice in the minute a
rollout straddles.

plans, timekeeper: one named FiredSchedule type, with a camelCase jobId
quoted in the recordset like every other JSON payload crossing to SQL.

docs, types: null means "no run recorded", not "never fired". The
annotation is a separate statement from the send, and the column is new,
so a long-running schedule reads null until its first fire after an
upgrade.

tests: the database tests poll instead of sleeping 4s and check
attribution through each job's payload, so an every-minute schedule
firing twice inside the window no longer fails them. The handler unit
tests move beside the existing onSendIt coverage in scheduleTest.ts and
share its fixture, and gain a partly-failed batch, a batch repeating one
schedule, and a missing error listener.

package.json: version 12.30.0, matching the release the migration
declares.

Author: kibertoad

docs/api/scheduling.md

@@ -101,7 +101,7 @@ Each schedule carries the following properties.
 | `options` | `send()` options applied to each job |
 | `createdOn` | When the schedule was first stored |
 | `updatedOn` | When the definition was last changed |
-| `lastJobId` | Id of the job the schedule most recently created, or `null` if it has not fired yet |
+| `lastJobId` | Id of the job the schedule most recently created |
 
 `lastJobId` connects a schedule to its last run, so a queue's history can be inspected from the
 schedule that produced it.
@@ -121,6 +121,12 @@ if (schedule.lastJobId) {
 eventually deleted, so it may no longer exist. Re-running `schedule()` for the same `(name, key)`
 updates the definition and leaves `lastJobId` alone; `unschedule()` removes the row entirely.
 
+It is recorded on a best-effort basis, in a separate statement once the job has been created, so
+`null` does not prove a schedule never fired. A schedule that last fired before the upgrade adding
+the column reads `null` until its next run, and so does one whose annotating statement failed after
+its job was already created. Treat it as a pointer to the last run pg-boss observed, not as a
+complete firing record: the queue's job history is the authority on what actually ran.
+
 ### `getSchedules(name)`
 
 Returns all scheduled jobs by queue name.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pg-boss",
-  "version": "12.29.0",
+  "version": "12.30.0",
   "description": "Queueing jobs in Postgres from Node.js like a boss",
   "type": "module",
   "main": "./dist/index.js",

packages/proxy/src/contracts.ts

@@ -249,6 +249,9 @@ export const scheduleSchema = z.object({
   timezone: z.string(),
   data: jsonRecordSchema.optional(),
   options: sendOptionsSchema.optional(),
+  createdOn: z.iso.datetime().transform((val) => new Date(val)),
+  updatedOn: z.iso.datetime().transform((val) => new Date(val)),
+  lastJobId: z.string().nullable(),
 }) satisfies z.ZodType<types.HttpSchedule>
 
 export const bamStatusSummarySchema = z.object({

src/plans.ts

@@ -948,13 +948,17 @@ export function getSchedulesByQueueAndKey (schema: string) {
 // job exists, one statement per batch rather than one per schedule, with the (name, key, job id)
 // triples carried in a JSON recordset.
 //
+// The caller must supply at most one record per (name, key): postgres leaves it unspecified which
+// source row an UPDATE ... FROM uses when several join the same target, so duplicates would make
+// the resulting last_job_id arbitrary rather than latest.
+//
 // updated_on is deliberately left alone: it tracks edits to the definition, and a firing schedule
 // has not been edited.
 export function setScheduleLastJobIds (schema: string) {
   return `
     UPDATE ${schema}.schedule s
-    SET last_job_id = x.job_id
-    FROM json_to_recordset($1::json) AS x (name text, key text, job_id uuid)
+    SET last_job_id = x."jobId"
+    FROM json_to_recordset($1::json) AS x (name text, key text, "jobId" uuid)
     WHERE s.name = x.name
       AND COALESCE(s.key, '') = x.key
   `

src/timekeeper.ts

@@ -34,6 +34,11 @@ const WARNING_TYPES = {
 // instance older than 12.30.0, which is why the handler treats it as optional rather than required.
 type ScheduledRequest = types.Request & { key?: string }
 
+// One schedule occurrence that produced a job, as handed to plans.setScheduleLastJobIds. camelCase
+// to match the recordset column list the plan quotes, which is how every other JSON payload crossing
+// into SQL is shaped.
+type FiredSchedule = { name: string, key: string, jobId: string }
+
 /**
  * Asserts that `tz` is a time zone cron evaluation can actually use.
  *
@@ -248,7 +253,12 @@ class Timekeeper extends EventEmitter implements types.EventsMixin {
       }
 
       if (due) {
-        scheduled.push({ data: { name, key, data, options }, singletonKey: `${name}__${key}`, singletonSeconds: 60 })
+        // JSON rather than `${name}__${key}`: underscores are legal in both a queue name and a
+        // schedule key, so the concatenation collapsed ('report_', 'daily') and ('report', '_daily')
+        // onto one key and the 60s singleton then dropped whichever occurrence lost the race. An
+        // instance still on the old format writes the old key, so a mixed-version deployment can
+        // fire a schedule twice in the minute the rollout straddles.
+        scheduled.push({ data: { name, key, data, options }, singletonKey: JSON.stringify([name, key]), singletonSeconds: 60 })
       }
     }
 
@@ -271,6 +281,18 @@ class Timekeeper extends EventEmitter implements types.EventsMixin {
     return prevDiff < 60
   }
 
+  // Reports a problem the send-it handler must survive. Node treats an `error` event with no
+  // listener as a throw, and index.ts re-promotes this one onto the PgBoss instance, so a plain
+  // emit() here could escape the handler, fail the send-it job and replay the whole batch, sending
+  // every occurrence in it a second time.
+  private reportSendItError (err: unknown): void {
+    try {
+      this.emit(this.events.error, err)
+    } catch {
+      // nothing left to report it to
+    }
+  }
+
   private async onSendIt (jobs: types.Job<ScheduledRequest>[]): Promise<void> {
     // async so a malformed payload rejects its own settlement rather than throwing synchronously
     // out of map() and taking the whole batch with it
@@ -279,12 +301,15 @@ class Timekeeper extends EventEmitter implements types.EventsMixin {
       return await this.manager.send(request)
     }))
 
-    const fired: { name: string, key: string, job_id: string }[] = []
+    // Keyed on (name, key) so a batch that spans two minute buckets for the same schedule resolves
+    // to its latest occurrence. Feeding both to the UPDATE would let postgres pick either source
+    // row, and last_job_id could end up naming the older job.
+    const fired = new Map<string, FiredSchedule>()
 
     // Surface any failed forward so a lost cron tick isn't silent
     for (const [index, result] of results.entries()) {
       if (result.status === 'rejected') {
-        this.emit(this.events.error, result.reason)
+        this.reportSendItError(result.reason)
         continue
       }
 
@@ -293,23 +318,29 @@ class Timekeeper extends EventEmitter implements types.EventsMixin {
       // send() resolves null when a throttle or queue policy dropped the job, so there is nothing
       // to point last_job_id at. `key` is absent on a payload written by an older instance.
       if (result.value && key !== undefined) {
-        fired.push({ name, key, job_id: result.value })
+        fired.set(JSON.stringify([name, key]), { name, key, jobId: result.value })
       }
     }
 
-    if (fired.length > 0) {
-      await this.setLastJobIds(fired)
+    if (fired.size > 0) {
+      await this.setLastJobIds([...fired.values()])
     }
   }
 
   // Best effort: the schedule fired and the job exists, so failing to annotate the schedule row
   // must not fail the send-it job and replay the occurrence. Reported through `error` instead.
-  private async setLastJobIds (fired: { name: string, key: string, job_id: string }[]): Promise<void> {
+  private async setLastJobIds (fired: FiredSchedule[]): Promise<void> {
     try {
       const sql = plans.setScheduleLastJobIds(this.config.schema)
       await this.db.executeSql(sql, [JSON.stringify(fired)])
     } catch (err) {
-      this.emit(this.events.error, err)
+      // Named, because a bare driver error here is indistinguishable from the forwarding failures
+      // emitted above, and the two call for different responses: this one leaves the jobs created
+      // and only the bookkeeping behind.
+      const schedules = fired.map(({ name, key }) => `"${name}" (key "${key}")`).join(', ')
+      const message = `Warning: schedules fired but their last job id could not be recorded for ${schedules}: ${(err as Error).message}`
+
+      this.reportSendItError(new Error(message, { cause: err }))
     }
   }
 

src/types.ts

@@ -903,8 +903,12 @@ export interface Schedule {
   createdOn: Date;
   updatedOn: Date;
   /**
-   * Id of the job this schedule most recently created, or `null` if it has not fired since the
-   * column was added.
+   * Id of the job this schedule most recently created.
+   *
+   * Recorded on a best-effort basis, in a separate statement once the job exists, so `null` does
+   * not mean the schedule never fired. It also reads `null` for a schedule that last fired before
+   * the upgrade that added the column, and for one whose annotating statement lost its connection
+   * between creating the job and recording it.
    *
    * Not a foreign key: the job is subject to the queue's retention policy and will eventually be
    * deleted, so an id here does not guarantee the job still exists.