|
| 1 | +import { randomUUID } from 'node:crypto' |
| 2 | +import { afterAll, describe, expect, it } from 'vitest' |
| 3 | +import { ctx } from './hooks.ts' |
| 4 | +import * as helper from './testHelper.ts' |
| 5 | +import * as plans from '../src/plans.ts' |
| 6 | +import type { IDatabase } from '../src/types.ts' |
| 7 | + |
| 8 | +// Some drivers infer a bind parameter's type from the cast in front of it, and that inference cuts |
| 9 | +// both ways depending on what pg-boss binds. |
| 10 | +// |
| 11 | +// Where pg-boss binds an already-stringified payload, `$N::json` reads as "this is JSON" and the |
| 12 | +// string gets encoded a second time: Postgres receives a JSON scalar and json_to_recordset() fails |
| 13 | +// with 22023. That is #880 (Bun.SQL, oven-sh/bun#28819). `$N::text::json` pins the inference to |
| 14 | +// text and changes nothing for the drivers that send the value as text either way. |
| 15 | +// |
| 16 | +// Where pg-boss binds a plain JS object, that same cast is what makes such a driver serialize it as |
| 17 | +// JSON at all - through text it falls back to String(value) and sends `[object Object]`. PGlite |
| 18 | +// does this, so both directions need pinning, which is what this file is for. |
| 19 | +// |
| 20 | +// No Bun in CI: the runtime half models the double-encoding driver with a wrapper, the way |
| 21 | +// distributedDatabaseTest wraps a connection for CockroachDB's INT8-as-string behaviour. |
| 22 | + |
| 23 | +const DIRECT_JSON_CAST = /\$\d+::jsonb?\b/ |
| 24 | +const TEXT_ROUTED_JSON_CAST = /\$\d+::text::jsonb?\b/ |
| 25 | + |
| 26 | +// 22023 invalid_parameter_value - what json_to_recordset() raises when handed a JSON scalar. |
| 27 | +const INVALID_PARAMETER_VALUE = '22023' |
| 28 | + |
| 29 | +const schema = 'pgboss' |
| 30 | +const table = plans.COMMON_JOB_TABLE |
| 31 | + |
| 32 | +// Statements whose JSON parameter is bound as a string. |
| 33 | +const textRouted: Array<[string, string]> = [ |
| 34 | + ['completeJobsWithOutputs', plans.completeJobsWithOutputs(schema, table)], |
| 35 | + ['completeJobsWithOutputsDistributed', plans.completeJobsWithOutputsDistributed(schema, table)], |
| 36 | + ['insertJobs', plans.insertJobs(schema, { table, name: 'q' })], |
| 37 | + ['failJobsByIdWithOutputs', plans.failJobsByIdWithOutputs(schema, table)], |
| 38 | + ['deadLetterJobsByIdWithOutputs', plans.deadLetterJobsByIdWithOutputs(schema, table)], |
| 39 | + ['updateJob', plans.updateJob(schema, table, 'q', 'id', 'newest')], |
| 40 | + ['insertDependencies', plans.insertDependencies(schema)], |
| 41 | + // buildFetchParams renders the tier parameter only with groupConcurrency.tiers set, and builds it |
| 42 | + // by concatenation - which is why a grep for `$N::jsonb` does not turn it up. |
| 43 | + ['fetchNextJob (group concurrency tiers)', plans.fetchNextJob({ |
| 44 | + schema, |
| 45 | + table, |
| 46 | + name: 'q', |
| 47 | + policy: undefined, |
| 48 | + limit: 1, |
| 49 | + ignoreSingletons: null, |
| 50 | + groupConcurrency: { default: 1, tiers: { enterprise: 3 } } |
| 51 | + }).text] |
| 52 | +] |
| 53 | + |
| 54 | +// Statements whose JSON parameter is bound as a plain JS object, left for the driver to serialize. |
| 55 | +const driverSerialized: Array<[string, string]> = [ |
| 56 | + ['updateQueue', plans.updateQueue(schema)], |
| 57 | + ['completeJobs', plans.completeJobs(schema, table)], |
| 58 | + ['completeJobs (includeQueued)', plans.completeJobs(schema, table, true)], |
| 59 | + // Wraps the same completeJobsUpdate body; this is the one the CockroachDB leg actually runs. |
| 60 | + ['completeJobsDistributed', plans.completeJobsDistributed(schema, table)], |
| 61 | + ['failJobsById', plans.failJobsById(schema, table)] |
| 62 | +] |
| 63 | + |
| 64 | +describe('json bind parameter casts', function () { |
| 65 | + for (const [name, sql] of textRouted) { |
| 66 | + it(`${name} routes its json parameter through text`, function () { |
| 67 | + expect(sql).not.toMatch(DIRECT_JSON_CAST) |
| 68 | + expect(sql).toMatch(TEXT_ROUTED_JSON_CAST) |
| 69 | + }) |
| 70 | + } |
| 71 | + |
| 72 | + for (const [name, sql] of driverSerialized) { |
| 73 | + it(`${name} keeps its json parameter cast direct`, function () { |
| 74 | + expect(sql).toMatch(DIRECT_JSON_CAST) |
| 75 | + expect(sql).not.toMatch(TEXT_ROUTED_JSON_CAST) |
| 76 | + }) |
| 77 | + } |
| 78 | + |
| 79 | + // Guards the guards: a typo in either pattern leaves every assertion above trivially true. |
| 80 | + it('the patterns recognise the casts they are looking for', function () { |
| 81 | + expect('SELECT $1::json').toMatch(DIRECT_JSON_CAST) |
| 82 | + expect('SELECT $2::jsonb').toMatch(DIRECT_JSON_CAST) |
| 83 | + expect('SELECT $1::text::json').not.toMatch(DIRECT_JSON_CAST) |
| 84 | + expect('SELECT $2::text::jsonb').not.toMatch(DIRECT_JSON_CAST) |
| 85 | + |
| 86 | + expect('SELECT $1::text::json').toMatch(TEXT_ROUTED_JSON_CAST) |
| 87 | + expect('SELECT $2::text::jsonb').toMatch(TEXT_ROUTED_JSON_CAST) |
| 88 | + expect('SELECT $1::json').not.toMatch(TEXT_ROUTED_JSON_CAST) |
| 89 | + }) |
| 90 | +}) |
| 91 | + |
| 92 | +// Re-encodes a string bound in front of a json cast, and leaves objects alone - a driver of this |
| 93 | +// kind serializes those correctly. |
| 94 | +// |
| 95 | +// `textRoutedHits` counts the strings it saw bound in front of a `::text::json` cast: every one of |
| 96 | +// those is a parameter that would have been double-encoded before this change. Asserting it is |
| 97 | +// non-zero is what stops the lifecycle test passing vacuously - without it, a boss that quietly |
| 98 | +// ignored the `db` option would look just as green. |
| 99 | +function doubleEncodesJsonParams (db: IDatabase): { db: IDatabase, textRoutedHits: () => number } { |
| 100 | + let hits = 0 |
| 101 | + |
| 102 | + const wrapped: IDatabase = { |
| 103 | + executeSql (text: string, values?: unknown[]) { |
| 104 | + const encoded = values?.map((value, index) => { |
| 105 | + if (typeof value !== 'string') return value |
| 106 | + |
| 107 | + if (new RegExp(`\\$${index + 1}::text::jsonb?\\b`).test(text)) hits++ |
| 108 | + |
| 109 | + return new RegExp(`\\$${index + 1}::jsonb?\\b`).test(text) ? JSON.stringify(value) : value |
| 110 | + }) |
| 111 | + |
| 112 | + return db.executeSql(text, encoded) |
| 113 | + } |
| 114 | + } |
| 115 | + |
| 116 | + return { db: wrapped, textRoutedHits: () => hits } |
| 117 | +} |
| 118 | + |
| 119 | +// Owned by the file rather than the test: hooks.ts stops the boss in afterEach, and the boss is |
| 120 | +// still talking to this connection when it does. |
| 121 | +let rawDb: Awaited<ReturnType<typeof helper.getDb>> | undefined |
| 122 | + |
| 123 | +afterAll(async () => { |
| 124 | + if (rawDb) await rawDb.close() |
| 125 | +}) |
| 126 | + |
| 127 | +describe('json bind parameters under a type-inferring driver', function () { |
| 128 | + it('runs the job lifecycle through a double-encoding driver', async function () { |
| 129 | + rawDb ??= await helper.getDb() |
| 130 | + const driver = doubleEncodesJsonParams(rawDb) |
| 131 | + ctx.boss = await helper.start({ ...ctx.bossConfig, db: driver.db }) |
| 132 | + |
| 133 | + const queue = ctx.schema |
| 134 | + |
| 135 | + // insertJobs, the plan #880 reported, through both of its entry points |
| 136 | + const jobId = await ctx.boss.send(queue, { hello: 'world' }) |
| 137 | + helper.assertTruthy(jobId) |
| 138 | + await ctx.boss.insert(queue, [{ data: { hello: 'insert' } }]) |
| 139 | + |
| 140 | + // updateJob, the other text-routed plan reachable from a public method. The id is new, so |
| 141 | + // this misses the update and inserts - three jobs now, and the fetch below has to take all |
| 142 | + // of them: fetch orders on (priority, created_on) with no tiebreaker, and PGlite's now() is |
| 143 | + // coarse enough that the three can share a created_on. Taking a subset would pick an |
| 144 | + // arbitrary one. (separateTimestamps only wraps send/insert, not upsert.) |
| 145 | + await ctx.boss.upsert(queue, { hello: 'upserted' }, { id: randomUUID() }) |
| 146 | + |
| 147 | + const jobs = await ctx.boss.fetch(queue, { batchSize: 3 }) |
| 148 | + expect(jobs).toHaveLength(3) |
| 149 | + |
| 150 | + const sent = jobs.find(job => job.id === jobId) |
| 151 | + helper.assertTruthy(sent) |
| 152 | + expect(sent.data).toEqual({ hello: 'world' }) |
| 153 | + |
| 154 | + // the object-bound outputs, which have to keep round-tripping |
| 155 | + await ctx.boss.complete(queue, jobs[0].id, { done: true }) |
| 156 | + await ctx.boss.fail(queue, jobs[1].id, { because: 'test' }) |
| 157 | + |
| 158 | + const completed = await ctx.boss.getJobById(queue, jobs[0].id) |
| 159 | + helper.assertTruthy(completed) |
| 160 | + expect(completed.output).toEqual({ done: true }) |
| 161 | + |
| 162 | + await ctx.boss.updateQueue(queue, { retryLimit: 7 }) |
| 163 | + const updated = await ctx.boss.getQueue(queue) |
| 164 | + helper.assertTruthy(updated) |
| 165 | + expect(updated.retryLimit).toBe(7) |
| 166 | + |
| 167 | + // The driver above only matters if pg-boss actually went through it. Every hit is a string |
| 168 | + // bound in front of a `::text::json` cast - the shape that used to be double-encoded. |
| 169 | + expect(driver.textRoutedHits()).toBeGreaterThan(0) |
| 170 | + }) |
| 171 | + |
| 172 | + it('breaks on a direct $N::json cast, and not on the text-routed one', async function () { |
| 173 | + rawDb ??= await helper.getDb() |
| 174 | + const { db } = doubleEncodesJsonParams(rawDb) |
| 175 | + const payload = JSON.stringify([{ id: 1 }, { id: 2 }]) |
| 176 | + |
| 177 | + // Negative control: the statement pg-boss used to emit. This is what establishes that the |
| 178 | + // wrapper can break a direct cast at all - it drives executeSql itself, so it says nothing |
| 179 | + // about pg-boss's own path. That part is textRoutedHits() in the test above. |
| 180 | + await expect(db.executeSql('SELECT * FROM json_to_recordset($1::json) AS x (id int)', [payload])) |
| 181 | + .rejects.toMatchObject({ code: INVALID_PARAMETER_VALUE }) |
| 182 | + |
| 183 | + const { rows } = await db.executeSql('SELECT * FROM json_to_recordset($1::text::json) AS x (id int)', [payload]) |
| 184 | + expect(rows).toEqual([{ id: 1 }, { id: 2 }]) |
| 185 | + }) |
| 186 | +}) |
0 commit comments