v0.7.2 feat(jobworkerdb): sanitize user provided job data PostgreSQL cannot store

CHANGELOG.md

@@ -5,6 +5,72 @@ All notable changes to this project are documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [v0.7.2] - 2026-07-30
+
+Jobs carrying user provided data that PostgreSQL cannot store are now persisted
+with the unstorable characters removed, instead of failing the write and losing
+the job.
+
+### Changed
+
+- **Writes that used to fail now succeed with characters removed.** `jobworkerdb`
+  strips what PostgreSQL cannot store from the job origin, payload, error message,
+  error data and result, and from the job bundle origin: zero bytes, invalid UTF-8,
+  and two escape sequences that jsonb rejects even though they are valid JSON —
+  the zero byte escape (what `encoding/json` emits for a zero byte inside a Go
+  string) and unpaired surrogates (what a browser's `JSON.stringify` emits for a
+  truncated emoji). Because both are accepted by `json.Valid`, nothing upstream
+  refused such a payload before and only the insert failed.
+
+  This is a behaviour change from fail-closed to silently transformed: a value read
+  back from the database may be missing characters that were present in the `Job`
+  passed to `Add`. The in-memory `Job` is never modified. Sanitizing deletes rather
+  than replaces, so it joins whatever surrounded a removed character — validate
+  values *after* sanitizing, not before, and see the `jobworkerdb` package docs for
+  the full list of consequences, including that sanitizing is not injective.
+
+- Sanitizing applies only to values being written, never to a value used to look
+  rows up. `DeleteJobsFromOrigin` and `DeleteJobBundlesFromOrigin` match their
+  argument verbatim, so a job whose origin had to be sanitized on write is
+  reachable by the sanitized string or by ID, but not by the original. Normalizing
+  a destructive unbounded key would be worse: sanitizing is not injective, so it
+  could collapse one origin onto another and delete jobs the caller never named.
+
+- A payload with nothing storable left now fails the write instead of being stored
+  as an empty object. `notnull.JSON` binds a nil value as `{}`, so such a job would
+  otherwise have been queued and dispatched to a worker as an all-zero payload.
+
+- `Job.Type` and `JobBundle.Type` are deliberately **not** sanitized, so an
+  unstorable type still fails the write loudly: `Job.Type` is the dispatch key a
+  worker registers for, and storing a rewritten one would queue a job no worker
+  could ever claim.
+
+- An error message that ends up empty is replaced by a placeholder rather than
+  stored as `''`, which would read back as SQL NULL through
+  `nullable.NonEmptyString` and make a failed job report `Succeeded() == true`.
+  The two ways it can be empty get distinct texts, so the row does not claim
+  characters were removed when the caller simply had nothing to say.
+
+- Removing characters can join a backslash to the text behind it and form one of
+  the rejected escapes that was not in the input, so stripping repeats — a bounded
+  number of times rather than until stable. Crafted malformed JSON can chain those
+  seams so that each round exposes exactly one more, which would make the work
+  quadratic on the write path. A value still unstorable after the cap is malformed
+  JSON that PostgreSQL rejects on its own, so the write fails loudly instead.
+
+### Added
+
+- A `WARN` log line whenever sanitizing actually removes characters, naming the
+  job or bundle id, the field, and the number of bytes dropped, so that a stored
+  value differing from what a producer sent is traceable instead of invisible.
+
+### Performance
+
+- Sanitizing runs on every job write and payloads can be large, so the
+  already-storable case — which is essentially every real payload — scans in place
+  and allocates nothing: 0 allocations for a 1MB payload, one allocation when
+  something is actually removed.
+
 ## [v0.7.1] - 2026-07-02
 
 Quieter shutdowns: a job dispatch cancelled during shutdown is no longer reported

VERSION

@@ -1 +1 @@
-v0.7.1
+v0.7.2

jobworkerdb/doc.go

@@ -71,6 +71,86 @@ other abandoned job, subject to its deadFor grace period:
 	set worker_alive_at=started_at, updated_at=now()
 	where started_at is not null and stopped_at is null and worker_alive_at is null;
 
+# String Sanitization
+
+Jobs routinely carry user provided data that PostgreSQL cannot store. It accepts
+zero bytes in neither text nor jsonb columns and rejects invalid UTF-8. On top of
+that, jsonb rejects two escape sequences that are perfectly valid JSON, which
+matters because encoding/json and json.Valid both accept them, so nothing
+upstream refuses such a payload and only the insert fails:
+
+  - the escape for a zero byte, which is what encoding/json emits for a zero byte
+    inside a Go string, so a marshalled payload carries no raw zero byte at all
+  - unpaired surrogates, meaning a high surrogate not followed by a low one or a
+    low surrogate not preceded by a high one. Complete surrogate pairs are valid
+    and are kept. Browsers' JSON.stringify emits an unpaired surrogate for a
+    truncated emoji.
+
+Rather than failing the write and losing the job, this package silently removes
+those characters as it writes. This applies to the job origin, payload, error
+message, error data and result, and to the job bundle origin. Values read back
+from the database therefore may be missing characters that were present in the
+Job passed to AddJob — the in-memory Job itself is never modified.
+
+Sanitizing applies to values being WRITTEN, never to a value used to look rows up.
+[jobworkerDB.DeleteJobsFromOrigin] and [jobworkerDB.DeleteJobBundlesFromOrigin]
+match their argument verbatim, so a job whose origin had to be sanitized on write
+is reachable by the sanitized string or by ID, but not by the original. Normalizing
+a destructive unbounded key would be worse: sanitizing is not injective, so it
+could collapse one origin onto another and delete jobs the caller never named.
+
+Three cases are not silently repaired, and fail the write instead. An origin
+consisting entirely of unstorable characters sanitizes to empty and violates the
+column's non-empty constraint. A payload with nothing storable left is rejected
+rather than stored as an empty object, which would otherwise be dispatched to a
+worker as an all-zero payload. And an error message that sanitizes to empty is
+replaced by a placeholder rather than stored empty, because an empty error_msg
+reads back as SQL NULL through nullable.NonEmptyString and would make a failed job
+look successful.
+
+Consequences worth knowing before relying on this:
+
+Validating the value a producer sends does not constrain what a worker reads.
+Sanitizing deletes characters rather than replacing them, so it joins whatever
+surrounded a removed character: "a<zero byte>b" passes a producer-side check as two
+tokens and is stored as "ab". A check that has to hold for the value a worker acts
+on belongs in the worker, against what it read back, not in the producer. Before
+this behaviour existed such a write failed instead, so this is a change from
+fail-closed to silently transformed.
+
+Sanitizing is not injective. Distinct inputs can collapse onto the same stored
+value, so two origins differing only in unstorable characters become one origin in
+the database, and an origin-keyed delete removes the jobs of both. Inside a
+payload this can go further than losing characters: two JSON object keys that
+differ only in unstorable characters become one duplicated key, and jsonb keeps
+the last of them, so one field's value is replaced by another's rather than merely
+shortened.
+
+Raw control bytes other than the zero byte are not removed. PostgreSQL accepts them
+in a text column, and in jsonb it tolerates only tab, newline and carriage return as
+whitespace between tokens; any other raw control byte between tokens is rejected, as
+is any raw control byte inside a jsonb string ("Character with value 0x01 must be
+escaped"). Reaching either state requires JSON that was already malformed, since a
+raw control byte inside a string is invalid JSON that json.Valid rejects, so such a
+value is not repaired and the write still fails. Detecting them would mean another
+full scan of every payload on the write path, on top of the two cheap vectorized
+scans already done for zero bytes and UTF-8 validity.
+
+Synchronous jobs are not sanitized. ContextWithSynchronousJobs executes jobs
+without touching the database, so the worker receives the payload and origin
+exactly as passed, while the same job persisted normally would be given the
+sanitized values. Tests using that helper cannot reproduce what a production
+worker reads.
+
+The job and job bundle Type is deliberately not sanitized, so an unstorable Type
+fails the write loudly instead. Job.Type is a dispatch key: jobworker looks a
+worker up by it, so rewriting it would store a job no worker could ever claim,
+which is worse than not storing it. JobBundle.Type is not a dispatch key — it is
+only reported to bundle listeners — but it is kept verbatim for consistency with
+Job.Type. jobqueue.NewJob checks Job.Type only for non-emptiness and NewJobBundle
+does not check JobBundle.Type at all, so a caller that derives a Type from user data
+has to keep it storable itself.
+
 # LISTEN/NOTIFY
 
 The service uses PostgreSQL LISTEN/NOTIFY for real-time job notifications:

jobworkerdb/jobworkerdb.go

@@ -131,9 +131,30 @@ func (*jobworkerDB) unlisten(ctx context.Context) (err error) {
 	return errors.Join(err1, err2)
 }
 
+// insertJob inserts the job row, sanitizing the origin and payload because they
+// can carry user provided data that PostgreSQL cannot store. job.Type is
+// deliberately left verbatim. See the String Sanitization section of the package
+// documentation for what is removed and why Type is exempt.
+//
+// Two outcomes specific to this statement: an origin whose every character is
+// unstorable sanitizes to "" and fails check(length(origin) > 0), while a payload
+// with nothing storable left does NOT fail, because notnull.JSON binds a nil value
+// as the empty object and the job is stored with an empty payload.
 func insertJob(ctx context.Context, job *jobqueue.Job) (err error) {
 	defer errs.WrapWithFuncParams(&err, ctx, job)
 
+	payload := sanitizedJSON(ctx, "jobID", job.ID, "payload", job.Payload)
+	origin := sanitizedString(ctx, "jobID", job.ID, "origin", job.Origin)
+
+	// A payload with nothing storable left must not be stored. notnull.JSON binds a
+	// nil value as the empty object, so without this the row would be written with
+	// payload {} and a worker would claim the job and act on an all-zero payload
+	// struct. Fail loudly instead, as an emptied origin and an unstorable Type
+	// already do — the caller passed data this queue cannot carry.
+	if len(job.Payload) > 0 && len(payload) == 0 {
+		return errs.New("job payload contained no characters PostgreSQL can store")
+	}
+
 	return db.Exec(ctx,
 		/*sql*/ `
 			INSERT INTO worker.job
@@ -160,9 +181,9 @@ func insertJob(ctx context.Context, job *jobqueue.Job) (err error) {
 		job.ID,            // $1
 		job.BundleID,      // $2
 		job.Type,          // $3
-		job.Payload,       // $4
+		payload,           // $4
 		job.Priority,      // $5
-		job.Origin,        // $6
+		origin,            // $6
 		job.MaxRetryCount, // $7
 		job.StartAt,       // $8
 	)
@@ -230,6 +251,8 @@ func (j *jobworkerDB) AddJobBundle(ctx context.Context, jobBundle *jobqueue.JobB
 		return nil
 	}
 
+	bundleOrigin := sanitizedString(ctx, "jobBundleID", jobBundle.ID, "origin", jobBundle.Origin)
+
 	return db.Transaction(ctx, func(ctx context.Context) error {
 		err = db.Exec(ctx,
 			/*sql*/ `
@@ -238,7 +261,7 @@ func (j *jobworkerDB) AddJobBundle(ctx context.Context, jobBundle *jobqueue.JobB
 			`,
 			jobBundle.ID,      // $1
 			jobBundle.Type,    // $2
-			jobBundle.Origin,  // $3
+			bundleOrigin,      // $3
 			jobBundle.NumJobs, // $4
 		)
 		if err != nil {
@@ -563,6 +586,31 @@ func (j *jobworkerDB) SetJobError(ctx context.Context, jobID uu.ID, errorMsg str
 		return jobqueue.ErrClosed
 	}
 
+	// Sanitize outside the transaction, so no scanning work is done while the
+	// bundle row below is locked.
+	//
+	// An error message must never end up empty: error_msg is a plain text column,
+	// so SQL would still read '' as errored (see GetAllJobsWithErrors and
+	// DeleteFinishedJobs), while Go reads it back through
+	// nullable.NonEmptyString, whose IsNull is `== ""`. Job.HasError would then
+	// report false and Job.Succeeded true for a job that actually failed, and
+	// DeleteFinishedJobs would never reap the row. Substitute a placeholder
+	// instead of letting sanitizing change the NULL-ness of the value.
+	// Distinguish the two ways the message can be empty here, because they mean
+	// different things to whoever reads the row: sanitizing removed everything, or
+	// the caller had nothing to say. Claiming the former for the latter would
+	// misreport the cause of the failure.
+	hadMessage := errorMsg != ""
+	errorMsg = sanitizedString(ctx, "jobID", jobID, "error message", errorMsg)
+	if errorMsg == "" {
+		if hadMessage {
+			errorMsg = "error message contained no characters PostgreSQL can store"
+		} else {
+			errorMsg = "job failed without an error message"
+		}
+	}
+	errorData = sanitizedJSON(ctx, "jobID", jobID, "error data", errorData)
+
 	return db.Transaction(ctx, func(ctx context.Context) error {
 		// SetJobError records a TERMINAL failure: the job has stopped and will
 		// not be retried. current_retry_count is clamped up to max_retry_count so
@@ -746,6 +794,11 @@ func (j *jobworkerDB) SetJobResult(ctx context.Context, jobID uu.ID, result null
 		return jobqueue.ErrClosed
 	}
 
+	// Sanitize before the empty check below, so that a result consisting only of
+	// characters PostgreSQL can't store falls back to the empty object instead of
+	// failing the update with empty JSON.
+	result = sanitizedJSON(ctx, "jobID", jobID, "result", result)
+
 	// if the result is `nil`, set an empty object so that the bundle knows the job existed correctly
 	if len(result) == 0 {
 		result = []byte("{}")
@@ -957,6 +1010,17 @@ func (j *jobworkerDB) DeleteJob(ctx context.Context, jobID uu.ID) (err error) {
 	)
 }
 
+// DeleteJobsFromOrigin deletes every job stored under exactly the given origin.
+//
+// The origin is matched verbatim, NOT sanitized the way insertJob sanitizes it on
+// write. That asymmetry is deliberate: sanitizing is not injective, so normalizing
+// a destructive unbounded key would let an origin containing unstorable characters
+// collapse onto a different, legitimate origin and delete its jobs — and it would
+// do so after any authorization the caller performed on the raw string. Deleting
+// too little is recoverable, deleting someone else's jobs is not.
+//
+// A job whose origin had to be sanitized on write is therefore reachable here only
+// by the sanitized string (or by ID).
 func (j *jobworkerDB) DeleteJobsFromOrigin(ctx context.Context, origin string) (err error) {
 	defer errs.WrapWithFuncParams(&err, ctx, origin)
 
@@ -1048,6 +1112,11 @@ func (j *jobworkerDB) DeleteJobBundle(ctx context.Context, jobBundleID uu.ID) (e
 	)
 }
 
+// DeleteJobBundlesFromOrigin deletes every job bundle stored under exactly the
+// given origin, cascading to the bundle's jobs.
+//
+// The origin is matched verbatim and not sanitized, for the reason given on
+// DeleteJobsFromOrigin — more so here, because this delete cascades.
 func (j *jobworkerDB) DeleteJobBundlesFromOrigin(ctx context.Context, origin string) (err error) {
 	defer errs.WrapWithFuncParams(&err, ctx, origin)