feat(14-02): conga.Describe reports a registered job's kind, queue, attempts and timeout
- lets a plugin test assert what its Jobs() registers, such as the CSV match job's 240 s timeout, without reaching into conga internals - README API table, jobs docs page and an example
This commit is contained in:
@@ -51,6 +51,8 @@ The options set the job's defaults:
|
|||||||
| `conga.MaxAttempts` | How many times the job is tried before its row is marked as an error. | `queue.max_attempts` (3) |
|
| `conga.MaxAttempts` | How many times the job is tried before its row is marked as an error. | `queue.max_attempts` (3) |
|
||||||
| `conga.Timeout` | The deadline of each attempt. | `queue.job_timeout` (300 seconds) |
|
| `conga.Timeout` | The deadline of each attempt. | `queue.job_timeout` (300 seconds) |
|
||||||
|
|
||||||
|
`conga.Describe` reports a job's kind and these defaults, so a plugin test can assert what its `Jobs()` registers, for example that a long job keeps its `conga.Timeout`.
|
||||||
|
|
||||||
`conga.JobID` returns the `summer_jobs` row of the running job. It reports `false` when the job has no row, as in the example above, where the function is called directly rather than by a worker.
|
`conga.JobID` returns the `summer_jobs` row of the running job. It reports `false` when the job has no row, as in the example above, where the function is called directly rather than by a worker.
|
||||||
|
|
||||||
## Registering jobs
|
## Registering jobs
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ The scheduler is the Go form of WinterCMS `registerSchedule`. Plugins declare re
|
|||||||
|
|
||||||
## Features
|
## Features
|
||||||
|
|
||||||
- River-free job declarations: `conga.Job` turns `func(ctx context.Context, args T) error` into a `pact.Job`; `conga.OnQueue`, `conga.MaxAttempts` and `conga.Timeout` set per-job defaults. A `pact.Job` not built by `conga.Job` is rejected with `conga.ErrNotCongaJob`.
|
- River-free job declarations: `conga.Job` turns `func(ctx context.Context, args T) error` into a `pact.Job`; `conga.OnQueue`, `conga.MaxAttempts` and `conga.Timeout` set per-job defaults, which `conga.Describe` reports back for tests. A `pact.Job` not built by `conga.Job` is rejected with `conga.ErrNotCongaJob`.
|
||||||
- Transactional dispatch: `conga.Manager.Dispatch` inserts the `summer_jobs` row with `conga.StatusInProgress`, the principal's user id and admin flag, `progress_max` from `conga.DispatchOpts.Count` and JSON metadata, then enqueues the River job in the same transaction. It opens a transaction itself when the caller has none.
|
- Transactional dispatch: `conga.Manager.Dispatch` inserts the `summer_jobs` row with `conga.StatusInProgress`, the principal's user id and admin flag, `progress_max` from `conga.DispatchOpts.Count` and JSON metadata, then enqueues the River job in the same transaction. It opens a transaction itself when the caller has none.
|
||||||
- Plain enqueue: `conga.Manager.Enqueue` inserts a River job without a record row, inside the caller's transaction when there is one.
|
- Plain enqueue: `conga.Manager.Enqueue` inserts a River job without a record row, inside the caller's transaction when there is one.
|
||||||
- Jobs whose worker ships later: a kind that no plugin registers is always inserted through the insert-only River client, so `conga.Manager.Dispatch` and `conga.Manager.Enqueue` succeed while a worker runs and the job waits on its queue. While a worker runs, such a kind must name a queue no worker serves; an empty queue, `default`, `conga.QueueScheduled`, a configured queue or a registered job's queue is `conga.ErrUnregisteredKindQueue`, because a worker would fetch the job, find no worker for its kind and discard it.
|
- Jobs whose worker ships later: a kind that no plugin registers is always inserted through the insert-only River client, so `conga.Manager.Dispatch` and `conga.Manager.Enqueue` succeed while a worker runs and the job waits on its queue. While a worker runs, such a kind must name a queue no worker serves; an empty queue, `default`, `conga.QueueScheduled`, a configured queue or a registered job's queue is `conga.ErrUnregisteredKindQueue`, because a worker would fetch the job, find no worker for its kind and discard it.
|
||||||
@@ -164,6 +164,8 @@ defer w.Stop(context.Background())
|
|||||||
| `conga.Job` | Wraps a typed job function as a `pact.Job` that conga can run on River. |
|
| `conga.Job` | Wraps a typed job function as a `pact.Job` that conga can run on River. |
|
||||||
| `conga.JobOption` | Per-job option: `conga.OnQueue`, `conga.MaxAttempts`, `conga.Timeout`. |
|
| `conga.JobOption` | Per-job option: `conga.OnQueue`, `conga.MaxAttempts`, `conga.Timeout`. |
|
||||||
| `conga.JobID` | Returns the record row id of the job running in a context. |
|
| `conga.JobID` | Returns the record row id of the job running in a context. |
|
||||||
|
| `conga.Describe` | Returns a job's kind, queue, attempt limit and timeout as set by its options (`conga.JobInfo`); false for a job not built by `conga.Job`. |
|
||||||
|
| `conga.JobInfo` | Kind and option defaults of a job built by `conga.Job`. |
|
||||||
| `conga.StartWorker` | Registers plugin jobs and starts a River worker client carrying the plugins' periodic schedule jobs. |
|
| `conga.StartWorker` | Registers plugin jobs and starts a River worker client carrying the plugins' periodic schedule jobs. |
|
||||||
| `conga.StartServeWorker` | The worker of the `serve` command; nil when `queue.work_in_serve` is false. |
|
| `conga.StartServeWorker` | The worker of the `serve` command; nil when `queue.work_in_serve` is false. |
|
||||||
| `conga.WorkerOptions` | Selects the queues a worker runs. |
|
| `conga.WorkerOptions` | Selects the queues a worker runs. |
|
||||||
|
|||||||
@@ -174,3 +174,11 @@ func TestDocsDispatch(t *testing.T) {
|
|||||||
time.Sleep(25 * time.Millisecond)
|
time.Sleep(25 * time.Millisecond)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func ExampleDescribe() {
|
||||||
|
job := conga.Job(func(ctx context.Context, args ImportPostsArgs) error { return nil },
|
||||||
|
conga.OnQueue("imports"), conga.Timeout(4*time.Minute))
|
||||||
|
info, ok := conga.Describe(job)
|
||||||
|
fmt.Println(ok, info.Kind, info.Queue, info.MaxAttempts, info.Timeout)
|
||||||
|
// Output: true acme_blog_import_posts imports 0 4m0s
|
||||||
|
}
|
||||||
|
|||||||
@@ -107,6 +107,27 @@ func (w *riverWorker[T]) Timeout(*river.Job[T]) time.Duration {
|
|||||||
return w.job.cfg.timeout
|
return w.job.cfg.timeout
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// JobInfo describes a job built by Job: its kind and the defaults its
|
||||||
|
// options set (zero values mean the queue config defaults apply).
|
||||||
|
type JobInfo struct {
|
||||||
|
Kind string
|
||||||
|
Queue string
|
||||||
|
MaxAttempts int
|
||||||
|
Timeout time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
|
// Describe returns the kind and option defaults of a job built by Job, so
|
||||||
|
// a plugin's tests can assert what Jobs() registers (for example a job's
|
||||||
|
// Timeout). ok is false for a pact.Job that was not built by Job.
|
||||||
|
func Describe(j pact.Job) (info JobInfo, ok bool) {
|
||||||
|
cj, ok := j.(congaJob)
|
||||||
|
if !ok || cj == nil {
|
||||||
|
return JobInfo{}, false
|
||||||
|
}
|
||||||
|
cfg := cj.config()
|
||||||
|
return JobInfo{Kind: cj.kind(), Queue: cfg.queue, MaxAttempts: cfg.maxAttempts, Timeout: cfg.timeout}, true
|
||||||
|
}
|
||||||
|
|
||||||
type jobIDKey struct{}
|
type jobIDKey struct{}
|
||||||
|
|
||||||
// JobID returns the summer_jobs id of the dispatched job running in ctx.
|
// JobID returns the summer_jobs id of the dispatched job running in ctx.
|
||||||
|
|||||||
Reference in New Issue
Block a user