Idempotent jobs and cancel#220
Merged
Merged
Conversation
Large audiobook episodes (~900MB concatenated) were exceeding the 30-minute upload context, so every S3 PUT failed with "context deadline exceeded" and the flow retried indefinitely. Raise the upload timeout to 2h and the concat timeout to 1h to fit large jobs. Also raise the per-call job-state context (GetJob / SaveJob status updates) from 1s to 10s. With busy_timeout now at 5s, a contended SQLite write waits for the lock up to 5s, which the old 1s deadline cut short — surfacing as "failed to save job state ... context deadline exceeded". 10s leaves headroom above busy_timeout. Applies to both the concatenate and upload-original flows. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The per-stage upload context is derived from the job context (context.WithTimeout(jobCtx, 2h)), and jobCtx carries the queue's WithDefaultJobTimeout deadline. With that default at 2h, time spent in download + concat ate into the upload budget, so a job that took e.g. 45m before uploading would still have its S3 PUT cancelled ~75m in, regardless of the 2h per-stage value — and keep retrying. Raise the default job timeout to 5h, above the worst-case sum of the stage budgets (download 1h + concat 1h + upload 2h = 4h), so each stage can use its full per-stage timeout. Applies to both the concatenate and upload-original flows. Reported-by: chatgpt-codex-connector Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Two changes that together stop a retry storm from re-running expensive
download/convert/upload work:
1. calculateJobId now normalizes the destination "uploadUrl" to just its path
before hashing. That URL is a presigned S3 URL whose signature and expiry
change on every request, so hashing it verbatim gave identical work a
different job id each time — defeating the duplicate check in CreateJob and
causing the same job to run again on every caller retry. The path still
encodes bucket + user + key, so distinct destinations stay distinct.
2. handleCreateJob now treats ErrJobAlreadyExists as success (202 with the
existing job id) instead of 500, so callers attach to the existing job
rather than erroring and retrying into a fresh job.
Also add a terminal "cancelled" status, a CancelJob method, and DELETE
/jobs/{id} so a caller can stop treating a stuck job as in-progress.
ErrJobAlreadyExists is now exported for callers to detect.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 822d8bfcf5
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Comment on lines
+225
to
+228
| if k == "uploadUrl" { | ||
| if s, ok := v.(string); ok { | ||
| v = stripURLQuery(s) | ||
| } |
There was a problem hiding this comment.
When a retry submits the same work with a fresh presigned uploadUrl, stripping the query makes it collide with the existing job id, but CreateJob then returns the already-persisted job without updating its params. If the original job is still queued/downloading/concatenating and its first presigned URL expires before upload, all later retries attach to that same job while the flow still uses the stale URL from storage, so the eventual upload fails instead of using the fresh signature.
Useful? React with 👍 / 👎.
Comment on lines
+274
to
+275
| job.DisplayStatus = JobStatusCancelled | ||
| if err := svc.storage.SaveJob(ctx, job); err != nil { |
There was a problem hiding this comment.
When DELETE /jobs/{id} lands while the job is still queued, this only saves a cancelled status; the queued process message remains, and onPublishedJob constructs and runs the flow without checking DisplayStatus, whose later status saves overwrite this value and still perform the download/upload. That makes a successful cancel response ineffective for the common pre-execution race.
Useful? React with 👍 / 👎.
Replace the hand-rolled method dispatcher (handleJobByID) and
strings.TrimPrefix path parsing with method+wildcard patterns
(POST /jobs, GET|DELETE /jobs/{id}, GET|POST /metadata[/long-polling])
and req.PathValue("id"). ServeMux now enforces methods, so the
per-handler req.Method guards are gone.
Span naming now reuses req.Pattern (the matched route template, e.g.
"GET /jobs/{id}") instead of manually rewriting /jobs/<uuid> -> /jobs/:id.
The template is inherently low-cardinality, which is what a span name
must be so SigNoz per-endpoint aggregation doesn't shatter.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Set go 1.26.5 + toolchain go1.26.5, and upgrade dependencies to latest minor/patch (otel 1.40->1.44, aws-sdk-go-v2, testcontainers 0.40->0.43, modernc/sqlite, golang.org/x/*, grpc, etc.). Deliberately excludes the anacrolix/torrent + RoaringBitmap/roaring + pion cluster: a blanket `go get -u ./...` drags anacrolix's transitive deps past what torrent v1.61.0 (already latest) supports, pulling in roaring/v2, which collides with the roaring v1 that torrent uses. That cluster is pinned; see CLAUDE.md. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Documents the go get -u trap (anacrolix/torrent vs roaring/v2), the targeted-upgrade workaround, the CGO/testcontainers test requirements, and the method-based routing convention. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…sts)
fake_s3.go pinned localstack/localstack:latest, which has rolled forward
to a Pro-gated build that exits with code 55 ("No credentials were found
... LOCALSTACK_AUTH_TOKEN") and refuses to start; the s3-latest image is
discontinued as of v2026.03. With no S3 container, GetS3Client errored and
every S3-dependent subtest t.Skip'd -- which also silently truncated the
autogenerated README (the /jobs sections vanished).
Pin localstack/localstack:3.8 (last widely-used token-free community line),
scope it to SERVICES=s3, and wait on /_localstack/health.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add a "cancelling a job" subtest to the gen_docs suite that POSTs a job and then DELETEs it, showing the cancelled-status response. Regenerate README.md so the new "DELETE /jobs/:id" example appears (and all existing sections are restored now that S3 comes up). Document the README-is- autogenerated flow and the localstack :latest trap in CLAUDE.md. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
service/jobs_queue: cut the whole-job budget from 5h to 2h (still the
parent of every per-stage context, with stage ceilings kept well under
it); add finite retries (5) with a capped, jittered backoff (~10s..30m)
instead of sqlq's uncapped 2^n default.
service/{concatenate,encode}_flow: tighten the short GetJob/SaveJob
timeouts (10s -> 5s) and the concat ceiling (1h -> 30m), and always
os.Remove the ~1GB concatenated temp file so a failed/aborted upload
can't leak it into /tmp.
uploader: give the uploader its own http.Client with granular transport
timeouts (no overall Client.Timeout, so a legitimate long ~1GB PUT isn't
cut off, but a stalled socket fails fast), serialize uploads via a
concurrency semaphore (concurrent large PUTs starve each other into S3
"RequestTimeout"), and retry a single upload up to 4x with jittered
backoff -- transport errors and transient statuses (incl. S3's 400
"RequestTimeout") are retryable; 403/expired-URL are not.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add storage.DeleteJob + service.DeleteJob, exposed as
DELETE /jobs/{id}?purge=true (default DELETE stays a soft cancel). Purge
removes the content-addressed record so an identical create re-runs the
work — the basis for retrying a failed/cancelled job.
Make cancel and purge actually interrupt an in-flight flow: each flow now
runs under a cancelable context registered by job id (Service.runningCancels),
and CancelJob/DeleteJob cancel it. A cancelled flow's status writes then fail
on the cancelled context instead of resurrecting a purged row (SaveJob is an
upsert). Also guard onPublishedJob against a job purged after being enqueued
but before processing — it previously dereferenced a nil job and panicked,
crashing the worker goroutine.
Tests: DeleteJob table test + TestCancelStopsRunningFlow (a mid-download flow
stops on cancel).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add a DELETE /jobs/{id}?purge=true example to the gen_docs suite and
regenerate README. Document the cancel/purge lifecycle in CLAUDE.md (both
stop a running flow; purge frees the content-addressed id for retry), and
drop the tg-podcastotron references — mediary is a separate product.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Address two review findings: - Cancel-before-execution (P2): a DELETE that lands while the job is still queued only set the cancelled status; onPublishedJob then ran the flow anyway and its status writes overwrote "cancelled". onPublishedJob now skips a job whose stored status is already cancelled, so a pre-execution cancel sticks. - Stale upload URL on duplicates (P1): the presigned uploadUrl is hashed path-only, so a retry with a fresh signature resolves to the existing job — but CreateJob returned it without updating params, leaving the flow to upload with the original (possibly expired) URL. CreateJob now refreshes the stored uploadUrl from the duplicate submit. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Dependabot groups default to applies-to: version-updates, so vulnerability fixes still arrived as separate PRs. Add a parallel security-updates group per ecosystem (gomod/docker/github-actions) so security bumps are bundled into one PR as well. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…-cancel # Conflicts: # service/concatenate_flow.go # service/encode_flow.go # service/jobs_queue/jobs_queue.go
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
The go 1.26.5 directive broke lint and test: CI's actions/setup-go with
go-version 1.26.x resolves to 1.26.4 and runs GOTOOLCHAIN=local, so it can't
satisfy or fetch 1.26.5 ("go.mod requires go >= 1.26.5"). Drop to 1.26.4 (the
installed patch everywhere) and remove the toolchain directive.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The concatenated result path is derived (via its extension) from user-provided variant names, and CodeQL flagged os.Remove on it as an uncontrolled path expression. Guard the removal so it only deletes a file under os.TempDir() (where Concatenate's os.CreateTemp placed it), so a crafted path can't escape and delete something else. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ided) mediary uploads job results to a caller-provided presigned URL, so the "uncontrolled data used in network request" (SSRF) finding is by design: the destination is meant to be caller-controlled and there is no fixed host to allowlist. Add a CodeQL config excluding go/request-forgery and wire it into the analysis workflow. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.