diff --git a/CHANGELOG.md b/CHANGELOG.md index 96e83e6..391e7dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,11 @@ versions may change the API. - `hoppersql`, a driver for `database/sql` (pgx's stdlib adapter or lib/pq) that passes the same conformance suite as `hopperpgx`. It polls instead of listening and inserts without COPY. +- Workflows: `NewWorkflow`, `Add` with `After`, `InsertWorkflow`/ + `InsertWorkflowTx`, `WorkflowGet` and `hopper workflows get`. Steps with + dependencies wait pending and are promoted by the statement that finalizes + the last of them; a failed step cancels its dependents unless they opt to + `DependencyIgnore`. Schema v4 adds `hopper_job_deps`. ### Changed diff --git a/cmd/hopper/main.go b/cmd/hopper/main.go index 5ce9e6f..397a8b5 100644 --- a/cmd/hopper/main.go +++ b/cmd/hopper/main.go @@ -23,6 +23,7 @@ import ( "os" "os/signal" "slices" + "strings" "text/tabwriter" "time" @@ -57,6 +58,7 @@ commands: queues pause|resume queues limit [-global N] [-rate R] [-burst B] [-partition P] [-aging D] (omitted limits are removed) clients list + workflows get stats bench (see the hopperbench command) @@ -111,6 +113,8 @@ func run(ctx context.Context, args []string, out io.Writer) error { return c.queues(ctx, rest[1:]) case "clients": return c.clients(ctx, rest[1:]) + case "workflows": + return c.workflows(ctx, rest[1:]) case "stats": return c.stats(ctx) default: @@ -339,6 +343,37 @@ func (c *cli) clients(ctx context.Context, args []string) error { }) } +func (c *cli) workflows(ctx context.Context, args []string) error { + if len(args) != 2 || args[0] != "get" { + return errors.New("workflows: get ") + } + id, err := hopper.ParseJobID(args[1]) + if err != nil { + return err + } + wf, err := c.client.WorkflowGet(ctx, id) + if err != nil { + return err + } + return c.print(wf, func(w io.Writer) { + b := wf.Batch + fmt.Fprintf(w, "id: %s\nname: %s\nprogress: %d of %d finished, %d failed\n", b.ID, b.Name, b.Total-b.Pending, b.Total, b.Failed) + if !b.CompletedAt.IsZero() { + fmt.Fprintf(w, "completed: %s\n", b.CompletedAt.Local().Format(time.RFC3339)) + } + deps := map[hopper.JobID][]string{} + for _, e := range wf.Edges { + deps[e.Job] = append(deps[e.Job], e.DependsOn.String()) + } + tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0) + fmt.Fprintln(tw, "ID\tKIND\tQUEUE\tSTATE\tATTEMPT\tAFTER") + for _, j := range wf.Jobs { + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%d/%d\t%s\n", j.ID, j.Kind, j.Queue, j.State, j.Attempt, j.MaxAttempts, strings.Join(deps[j.ID], ",")) + } + tw.Flush() + }) +} + func (c *cli) stats(ctx context.Context) error { stats, err := c.client.Stats(ctx) if err != nil { diff --git a/cmd/hopper/main_test.go b/cmd/hopper/main_test.go index 3214006..8adb5a2 100644 --- a/cmd/hopper/main_test.go +++ b/cmd/hopper/main_test.go @@ -108,6 +108,17 @@ func TestCLI(t *testing.T) { if got := hopperCmd("stats"); !strings.Contains(got, "q1") { t.Errorf("stats = %q", got) } + wf := hopper.NewWorkflow("ingest", nil) + first := wf.Add(ping{}, nil) + wf.Add(ping{}, hopper.After(first)) + wres, err := client.InsertWorkflow(ctx, wf) + if err != nil { + t.Fatal(err) + } + if got := hopperCmd("workflows", "get", wres.ID.String()); !strings.Contains(got, "name: ingest") || + !strings.Contains(got, "0 of 2 finished") || !strings.Contains(got, "pending") || !strings.Contains(got, wres.Jobs[0].Job.ID.String()) { + t.Errorf("workflows get = %q", got) + } if got := hopperCmd("migrate", "down"); !strings.Contains(got, "schema version 0") { t.Errorf("migrate down = %q", got) } diff --git a/docs/PLAN.md b/docs/PLAN.md index aed3a52..8e86f0d 100644 --- a/docs/PLAN.md +++ b/docs/PLAN.md @@ -231,12 +231,13 @@ b.Add(ProcessShard{Shard: 0}, nil) b.Add(ProcessShard{Shard: 1}, nil) err = b.InsertTx(ctx, tx) -wf := hopper.NewWorkflow("ingest-9") -fetch := wf.Add(Fetch{URL: u}) +wf := hopper.NewWorkflow("ingest-9", &hopper.WorkflowOpts{OnFailure: AlertOps{RunID: 9}}) +fetch := wf.Add(Fetch{URL: u}, nil) parse := wf.Add(Parse{}, hopper.After(fetch)) wf.Add(Index{}, hopper.After(parse)) wf.Add(Notify{}, hopper.After(parse)) -err = client.InsertWorkflowTx(ctx, tx, wf) +res, err := client.InsertWorkflowTx(ctx, tx, wf) +row, err := client.WorkflowGet(ctx, res.ID) // progress, jobs and edges ``` ### 4.7 API conventions @@ -444,7 +445,8 @@ CREATE TABLE hopper_schema (version int PRIMARY KEY, applied_at timestamptz NOT Schema v2 (M6) adds `max_attempts` and `metadata` to `hopper_subscriptions`, the ordering-key indexes (§10) and the SQL contract functions. Schema v3 (M7) adds `partition_limit` and `aging_seconds` to `hopper_queues`, the partition-key running -index, `hopper_batches` and the batch index. M8 adds `hopper_job_deps` (§11). +index, `hopper_batches` and the batch index. Schema v4 (M8) adds `hopper_job_deps`, the +name and edges of a batch, and a `batch_id` index on history (§11). ### 6.1 Job IDs Job IDs are UUIDv7. They are safe to expose outside the application (in URLs, APIs @@ -919,11 +921,23 @@ observability without new machinery. failed, `on_failure` otherwise, `on_complete` either way, each with `batch_id` and `batch_failed` in its metadata, and notifies their queues. `client.NewBatch(opts)`, `Add`, `Insert`/`InsertTx` and `BatchGet` are the API. -- **Workflows (M8).** Jobs with dependencies are inserted as `pending`, with edges in - `hopper_job_deps`. When a job completes, the same finalize statement promotes - dependents whose dependencies are all complete to `available`. Failure policies are - `cancel dependents` (the default) and `ignore`. Workflows are inspectable as a DAG in - `hopperui`. +- **Workflows (M8).** A workflow is a batch whose jobs depend on each other, so it has + a name, the batch's progress and callbacks, and its graph (`hopper_batches.edges`). + `NewWorkflow` collects steps; a step can only depend on steps added before it, so the + graph is a DAG by construction. `InsertWorkflow` generates the IDs up front and + inserts the batch, the jobs (`pending` when they have dependencies, otherwise + available) and the edges in one transaction. `hopper_job_deps` is the working set of + unsatisfied edges, deleted as dependents finalize, and the statement that finalizes + a job is the one that acts on its dependents: it promotes those whose dependencies + have all left the live table (including the ones finalized in the same statement) + to `available` and notifies their queues, and it cancels, transitively, the + dependents of a cancelled or discarded job whose edge says `cancel` (the default; + `ignore` lets the dependent run once its dependencies have finished, whatever their + outcome). Cancelled dependents go to history with a "dependency failed" error and + count as batch failures. Every finalizing statement (the batched finalize, + cancelling a waiting job, expiring) ends with the same tail, so batches and + dependents behave the same however a job finishes. Workflows are inspectable with + `WorkflowGet`, `hopper workflows get` and, as a DAG, in `hopperui`. ## 12. Migrations @@ -960,8 +974,8 @@ observability without new machinery. gauges for queue depth by state, the oldest claimable job's age and live clients from `Stats`. Prometheus users export through the OTel exporter. - **CLI (`cmd/hopper`, core module):** `migrate up|down|version`, `jobs list|get|retry|cancel`, - `queues list|pause|resume`, `clients list` and `stats`, all with `-json`. `queues limit` - and `subscriptions list` arrive with M7 and M6. Benchmarks are the separate + `queues list|pause|resume|limit`, `clients list`, `workflows get` and `stats`, all with + `-json`. `subscriptions list` arrives with M6. Benchmarks are the separate `hopperbench` command. - **Web UI (`hopperui` module):** an embeddable `http.Handler` for browsing queues, jobs, history, subscriptions and workflows, with retry, cancel and pause actions @@ -1006,7 +1020,7 @@ The estimates assume one engineer. Each milestone is one or more PRs. | M5 | First adoption | Move an internal service's `internal/jobs` package to hopper; drain and drop its old queue tables | 1d | | M6 | Messaging | Subscriptions, AMQP topic patterns, typed `Message[T]`, PublishTx fan-out, dedup, ordering keys, request/reply, SQL publish contract, `ReplayDiscarded`, the upgrade test. **Done**; **v0.2.0** follows v0.1.0. | 5d | | M7 | Flow control and batches | Global limits, rate limits, partitioned limits, priority aging, batches with callbacks, `hoppersql` driver, **v0.3.0**. **Done.** | 5d | -| M8 | Workflows, streams, UI | Job dependencies and DAG workflows, streams with consumer groups, `hopperui` | 2–3w | +| M8 | Workflows, streams, UI | Job dependencies and DAG workflows (**done**), streams with consumer groups, `hopperui`. Each lands in its own PR. | 2–3w | | M9 | More engines (later) | `hoppersqlite`, then `hoppermongo`, each in its own module and passing `drivertest`. Not scheduled yet. | per engine | M0–M4 take roughly four weeks to a production-ready v0.1.0 that meets its performance diff --git a/docs/getting-started.md b/docs/getting-started.md index 4044d11..0cc6114 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -165,6 +165,29 @@ Periodic: []hopper.PeriodicJob{ The leader inserts each slot exactly once, whichever replica is leader. +## Batches and workflows + +```go +b := client.NewBatch(hopper.BatchOpts{OnSuccess: ReportDone{RunID: 9}, OnFailure: AlertOps{RunID: 9}}) +b.Add(ProcessShard{Shard: 0}, nil) +b.Add(ProcessShard{Shard: 1}, nil) +res, err := b.InsertTx(ctx, tx) + +wf := hopper.NewWorkflow("ingest-9", &hopper.WorkflowOpts{OnFailure: AlertOps{RunID: 9}}) +fetch := wf.Add(Fetch{URL: u}, nil) +parse := wf.Add(Parse{}, hopper.After(fetch)) +wf.Add(Index{}, hopper.After(parse)) +wf.Add(Notify{}, hopper.After(parse)) +res, err := client.InsertWorkflowTx(ctx, tx, wf) +``` + +A batch runs its callbacks when its last job finishes. A workflow's steps +run as their dependencies finish; a failed step cancels the steps that +depend on it (set `StepOpts.OnDependencyFailure` to `DependencyIgnore` to run +anyway), and the workflow is a batch, so it has the same callbacks. +`client.WorkflowGet(ctx, res.ID)` and `hopper workflows get ` show its +progress and graph. + ## Results and waiting ```go diff --git a/driver/driver.go b/driver/driver.go index aae9571..3aad550 100644 --- a/driver/driver.go +++ b/driver/driver.go @@ -189,6 +189,13 @@ type Executor interface { BatchInsert(ctx context.Context, params BatchInsertParams) (JobID, error) // BatchGet returns a batch's progress. BatchGet(ctx context.Context, id JobID) (*BatchRow, error) + // WorkflowInsert inserts a batch, its jobs and the dependencies between + // them atomically. Jobs with dependencies start pending. Results are in + // input order. + WorkflowInsert(ctx context.Context, params WorkflowInsertParams) (*WorkflowInsertResult, error) + // WorkflowGet returns a workflow's progress, its jobs (live and + // finished) and its edges. + WorkflowGet(ctx context.Context, id JobID) (*WorkflowRow, error) // PeriodicInsert inserts the job for a periodic slot if that slot has // not been inserted yet, atomically, and reports whether it did. A slot @@ -419,7 +426,9 @@ type BatchInsertParams struct { // BatchRow is a batch's progress. type BatchRow struct { - ID JobID + ID JobID + // Name is set for workflows. + Name string Pending int Failed int Total int @@ -429,6 +438,60 @@ type BatchRow struct { Metadata json.RawMessage } +// DependencyFailure says what happens to a dependent job when the job it +// depends on is cancelled or discarded. +type DependencyFailure string + +const ( + // DependencyCancel cancels the dependent, and so on down the graph. + DependencyCancel DependencyFailure = "cancel" + // DependencyIgnore lets the dependent run once its dependencies have + // all finished, whatever their outcome. + DependencyIgnore DependencyFailure = "ignore" +) + +// JobDependency is an edge of a workflow: Job waits for DependsOn. Both +// are indexes into WorkflowInsertParams.Jobs. +type JobDependency struct { + Job, DependsOn int + OnFailure DependencyFailure +} + +// WorkflowInsertParams describes a workflow: a named batch whose jobs +// depend on each other. +type WorkflowInsertParams struct { + Name string + // Jobs must have no unique keys; a skipped job would leave its + // dependents waiting forever. + Jobs []JobInsertParams + Deps []JobDependency + // Callbacks and Metadata are the batch's; see BatchInsertParams. + OnSuccess, OnFailure, OnComplete *JobInsertParams + Metadata json.RawMessage + // Notify sends insert notifications for the jobs that start + // available from inside the statement. + Notify bool +} + +// WorkflowInsertResult reports an inserted workflow. +type WorkflowInsertResult struct { + ID JobID + Jobs []JobInsertResult +} + +// JobEdge is an edge of an inserted workflow. +type JobEdge struct { + Job, DependsOn JobID +} + +// WorkflowRow is a workflow's progress, jobs and graph. +type WorkflowRow struct { + Batch BatchRow + // Jobs are in insertion order, live and finished alike. + Jobs []*JobRow + Edges []JobEdge +} + // PeriodicInsertParams identifies a periodic slot and the job to insert // for it. type PeriodicInsertParams struct { diff --git a/driver/internal/pgsql/flow.go b/driver/internal/pgsql/flow.go index db6d691..294b83a 100644 --- a/driver/internal/pgsql/flow.go +++ b/driver/internal/pgsql/flow.go @@ -12,33 +12,6 @@ import ( "github.com/parallelworks/hopper/driver" ) -// batchAccountingSQLWith returns the CTEs that count finalized jobs off -// their batches and insert the callbacks of batches that reached zero, all -// inside the finalizing statement. They expect a CTE named done whose rows -// carry batch_id and a text column final_state, and the insert channel as -// the given parameter. Rows without a batch cost one filtered scan of the -// CTE and nothing else. -func batchAccountingSQLWith(channelParam string) string { - return `batches AS ( - UPDATE hopper_batches b - SET pending = b.pending - c.n, failed = b.failed + c.f, - completed_at = CASE WHEN b.pending - c.n <= 0 THEN now() ELSE b.completed_at END - FROM (SELECT batch_id, count(*) AS n, count(*) FILTER (WHERE final_state IN ('cancelled', 'discarded')) AS f - FROM done WHERE batch_id IS NOT NULL GROUP BY batch_id) c - WHERE b.id = c.batch_id - RETURNING b.id, b.pending, b.failed, b.on_success, b.on_failure, b.on_complete -), -callbacks AS ( - INSERT INTO hopper_jobs (kind, queue, priority, max_attempts, args, metadata) - SELECT cb->>'kind', coalesce(cb->>'queue', 'default'), coalesce((cb->>'priority')::smallint, 2), - coalesce((cb->>'max_attempts')::smallint, 25), coalesce(cb->'args', '{}'), - coalesce(cb->'metadata', '{}') || jsonb_build_object('batch_id', b.id::text, 'batch_failed', b.failed) - FROM batches b, LATERAL (VALUES (CASE WHEN b.failed = 0 THEN b.on_success ELSE b.on_failure END), (b.on_complete)) AS v(cb) - WHERE b.pending <= 0 AND cb IS NOT NULL - RETURNING pg_notify(` + channelParam + `, queue) -)` -} - // claimLimited claims inside a transaction that locks the queue row, so that // every client's claims on a limited queue are serialized and the limits // hold exactly: the budget is the smallest of the free slots, what the diff --git a/driver/internal/pgsql/jobs.go b/driver/internal/pgsql/jobs.go index a925d36..476cf5f 100644 --- a/driver/internal/pgsql/jobs.go +++ b/driver/internal/pgsql/jobs.go @@ -312,11 +312,12 @@ func claimRows(ctx context.Context, conn Conn, query string, args ...any) ([]*dr // jobFinalizeSQL applies a batch of results in one statement. Every // transition checks the prior state (running) and the owning client, so a // stale result from a fenced client never overwrites a newer attempt. -// Retries and snoozes are updated in place; terminal outcomes are moved to -// history. Error timestamps come from database time, awaited jobs announce -// themselves on the done channel, and batches are counted down. +// Retries and snoozes are updated in place; terminal outcomes go through +// finalizeTailSQL: history, batches, and workflow dependents. Error +// timestamps come from database time and awaited jobs announce themselves +// on the done channel. var jobFinalizeSQL = ` -WITH r AS ( +WITH RECURSIVE r AS ( SELECT * FROM unnest( $1::uuid[], $2::bigint[], $3::text[], $4::float8[], $5::boolean[], $6::jsonb[], $7::jsonb[], $8::boolean[] ) AS r(id, attempted_by, state, delay, snooze, error, output, archive) @@ -334,32 +335,16 @@ retry AS ( AND r.state IN ('retryable', 'scheduled') RETURNING j.id ), -done AS ( +finished AS ( DELETE FROM hopper_jobs j USING r WHERE j.id = r.id AND j.state = 'running' AND j.attempted_by = r.attempted_by AND r.state IN ('completed', 'cancelled', 'discarded') - RETURNING j.id, j.seq, j.kind, j.queue, r.state, j.priority, j.attempt, j.max_attempts, - j.scheduled_at, j.attempted_at, j.attempted_by, j.args, j.metadata, - CASE WHEN r.error IS NULL THEN j.errors - ELSE j.errors || jsonb_set(r.error, '{at}', to_jsonb(now())) END AS errors, - j.unique_key, j.ordering_key, j.partition_key, j.batch_id, j.expires_at, - j.cancel_requested_at, j.await, j.created_at, r.output, r.archive, r.state AS final_state, - CASE WHEN j.await THEN pg_notify($9, j.id::text) END AS notified + RETURNING ` + finishedColumns(`CASE WHEN r.error IS NULL THEN j.errors + ELSE j.errors || jsonb_set(r.error, '{at}', to_jsonb(now())) END`, "r.output", "r.archive", "r.state", "$9") + ` ), -archived AS ( - INSERT INTO hopper_job_history ( - id, seq, kind, queue, state, priority, attempt, max_attempts, scheduled_at, attempted_at, - attempted_by, args, metadata, errors, unique_key, ordering_key, partition_key, batch_id, - expires_at, cancel_requested_at, await, created_at, finalized_at, output) - SELECT id, seq, kind, queue, state::hopper_job_state, priority, attempt, max_attempts, scheduled_at, attempted_at, - attempted_by, args, metadata, errors, unique_key, ordering_key, partition_key, batch_id, - expires_at, cancel_requested_at, await, created_at, now(), output - FROM done WHERE archive - RETURNING id -), -` + batchAccountingSQLWith("$10") + ` -SELECT id FROM retry UNION ALL SELECT id FROM done` +` + finalizeTailSQL("$9", "$10") + ` +SELECT id FROM retry UNION ALL SELECT id FROM finished` // JobFinalizeMany implements driver.Executor. func (e *Executor) JobFinalizeMany(ctx context.Context, params driver.JobFinalizeParams) ([]driver.JobID, error) { @@ -477,34 +462,20 @@ func (e *Executor) JobList(ctx context.Context, params driver.JobListParams) ([] return jobs, nil } -// historyInsertSQL archives rows from a CTE named done, with the given -// state and an error appended. The CTE must return hopper_jobs columns. -func historyInsertSQL(state, errorText string) string { - return fmt.Sprintf(` -INSERT INTO hopper_job_history ( - id, seq, kind, queue, state, priority, attempt, max_attempts, scheduled_at, attempted_at, - attempted_by, args, metadata, errors, unique_key, ordering_key, partition_key, batch_id, - expires_at, cancel_requested_at, await, created_at, finalized_at, output) -SELECT id, seq, kind, queue, '%s', priority, attempt, max_attempts, scheduled_at, attempted_at, - attempted_by, args, metadata, - errors || jsonb_build_object('at', now(), 'attempt', attempt, 'error', '%s'), - unique_key, ordering_key, partition_key, batch_id, - expires_at, cancel_requested_at, await, created_at, now(), NULL -FROM done`, state, errorText) -} - var ( jobCancelRunningSQL = fmt.Sprintf(` UPDATE hopper_jobs SET cancel_requested_at = coalesce(cancel_requested_at, now()) WHERE id = $1 AND state = 'running' RETURNING %s`, JobColumns("")) - jobCancelWaitingSQL = fmt.Sprintf(` -WITH done AS ( - DELETE FROM hopper_jobs WHERE id = $1 AND state <> 'running' RETURNING *, 'cancelled'::text AS final_state, - CASE WHEN await THEN pg_notify($2, id::text) END AS notified -), archived AS (%s RETURNING %s, finalized_at, output), -%s -SELECT * FROM archived`, historyInsertSQL("cancelled", "hopper: cancelled"), JobColumns(""), batchAccountingSQLWith("$3")) + // jobCancelWaitingSQL finalizes a waiting job as cancelled, with + // everything that entails for its batch and its dependents. + jobCancelWaitingSQL = ` +WITH RECURSIVE finished AS ( + DELETE FROM hopper_jobs j WHERE j.id = $1 AND j.state <> 'running' + RETURNING ` + finishedColumns(appendErrorSQL("hopper: cancelled"), "NULL::jsonb", "true", "'cancelled'::text", "$2") + ` +), +` + finalizeTailSQL("$2", "$3") + ` +SELECT ` + finishedSelectColumns + ` FROM finished` ) // JobCancel implements driver.Executor. @@ -602,18 +573,18 @@ func (e *Executor) JobRetry(ctx context.Context, id driver.JobID) (*driver.JobRo return job, nil } -var jobDiscardExpiredSQL = fmt.Sprintf(` -WITH done AS ( +var jobDiscardExpiredSQL = ` +WITH RECURSIVE finished AS ( DELETE FROM hopper_jobs j WHERE j.id IN ( SELECT id FROM hopper_jobs WHERE state IN ('available', 'scheduled', 'retryable') AND expires_at <= now() LIMIT $1 FOR UPDATE SKIP LOCKED ) - RETURNING j.*, 'discarded'::text AS final_state, CASE WHEN j.await THEN pg_notify($2, j.id::text) END AS notified -), archived AS (%s RETURNING %s, finalized_at, output), -%s -SELECT * FROM archived`, historyInsertSQL("discarded", "hopper: expired"), JobColumns(""), batchAccountingSQLWith("$3")) + RETURNING ` + finishedColumns(appendErrorSQL("hopper: expired"), "NULL::jsonb", "true", "'discarded'::text", "$2") + ` +), +` + finalizeTailSQL("$2", "$3") + ` +SELECT ` + finishedSelectColumns + ` FROM finished` // JobDiscardExpired implements driver.Executor. func (e *Executor) JobDiscardExpired(ctx context.Context, limit int) ([]*driver.JobRow, error) { diff --git a/driver/internal/pgsql/workflow.go b/driver/internal/pgsql/workflow.go new file mode 100644 index 0000000..54d7e77 --- /dev/null +++ b/driver/internal/pgsql/workflow.go @@ -0,0 +1,279 @@ +package pgsql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/parallelworks/hopper/driver" +) + +// Every statement that finalizes jobs (the batched finalize, cancelling a +// waiting job, expiring) produces a CTE named finished, in the shape +// finishedColumns builds, and ends with finalizeTailSQL, which does +// everything a finalized job entails: cancelling the dependents of a failed +// job, archiving, counting batches down, promoting dependents whose +// dependencies are all finished, and dropping satisfied edges. + +// finishedColumns lists the columns a finishing CTE returns from a +// hopper_jobs row aliased j: the row, with the given expressions for its +// errors, output, whether to archive it and its final state (text), and +// the done-channel notification for an awaited job. The notification sits +// in RETURNING because a data-modifying CTE always runs, while a plain +// SELECT CTE that nothing references does not. +func finishedColumns(errorsExpr, outputExpr, archiveExpr, stateExpr, doneChannelParam string) string { + return fmt.Sprintf(`j.id, j.seq, j.kind, j.queue, j.priority, j.attempt, j.max_attempts, j.scheduled_at, j.attempted_at, + j.attempted_by, j.args, j.metadata, %s AS errors, j.unique_key, j.ordering_key, j.partition_key, j.batch_id, + j.expires_at, j.cancel_requested_at, j.await, j.created_at, %s AS output, %s AS archive, %s AS final_state, + CASE WHEN j.await THEN pg_notify(%s, j.id::text) END AS notified`, + errorsExpr, outputExpr, archiveExpr, stateExpr, doneChannelParam) +} + +// appendErrorSQL is an errors expression that records a failure hopper +// itself decided, at database time. +func appendErrorSQL(message string) string { + return fmt.Sprintf(`j.errors || jsonb_build_object('at', now(), 'attempt', j.attempt, 'error', '%s')`, message) +} + +// finishedSelectColumns reads finished rows back in JobColumns order plus +// finalized_at and output, as history rows scan. +const finishedSelectColumns = `id, kind, queue, final_state, priority, attempt, max_attempts, scheduled_at, attempted_at, + attempted_by, args, metadata, errors, unique_key, ordering_key, partition_key, batch_id, expires_at, + cancel_requested_at, await, created_at, now() AS finalized_at, output` + +// finalizeTailSQL returns the CTEs that follow a finished CTE; the statement +// must start with WITH RECURSIVE. Cancelled dependents that were awaited +// are announced on the done channel, and promoted jobs and callbacks on +// the insert channel. Jobs without a batch or dependents cost index probes +// on small tables and nothing else. +func finalizeTailSQL(doneChannelParam, insertChannelParam string) string { + return `cascade AS ( + SELECT d.job_id FROM hopper_job_deps d JOIN finished f ON d.depends_on = f.id + WHERE f.final_state IN ('cancelled', 'discarded') AND d.on_failure = 'cancel' + UNION + SELECT d.job_id FROM hopper_job_deps d JOIN cascade c ON d.depends_on = c.job_id + WHERE d.on_failure = 'cancel' +), +cascaded AS ( + DELETE FROM hopper_jobs j USING (SELECT DISTINCT job_id FROM cascade) c + WHERE j.id = c.job_id AND j.state = 'pending' + RETURNING ` + finishedColumns(appendErrorSQL("hopper: dependency failed"), "NULL::jsonb", "true", "'cancelled'::text", doneChannelParam) + ` +), +done AS (SELECT * FROM finished UNION ALL SELECT * FROM cascaded), +archived AS ( + INSERT INTO hopper_job_history ( + id, seq, kind, queue, state, priority, attempt, max_attempts, scheduled_at, attempted_at, + attempted_by, args, metadata, errors, unique_key, ordering_key, partition_key, batch_id, + expires_at, cancel_requested_at, await, created_at, finalized_at, output) + SELECT id, seq, kind, queue, final_state::hopper_job_state, priority, attempt, max_attempts, scheduled_at, attempted_at, + attempted_by, args, metadata, errors, unique_key, ordering_key, partition_key, batch_id, + expires_at, cancel_requested_at, await, created_at, now(), output + FROM done WHERE archive + RETURNING id +), +batches AS ( + UPDATE hopper_batches b + SET pending = b.pending - c.n, failed = b.failed + c.f, + completed_at = CASE WHEN b.pending - c.n <= 0 THEN now() ELSE b.completed_at END + FROM (SELECT batch_id, count(*) AS n, count(*) FILTER (WHERE final_state IN ('cancelled', 'discarded')) AS f + FROM done WHERE batch_id IS NOT NULL GROUP BY batch_id) c + WHERE b.id = c.batch_id + RETURNING b.id, b.pending, b.failed, b.on_success, b.on_failure, b.on_complete +), +callbacks AS ( + INSERT INTO hopper_jobs (kind, queue, priority, max_attempts, args, metadata) + SELECT cb->>'kind', coalesce(cb->>'queue', 'default'), coalesce((cb->>'priority')::smallint, 2), + coalesce((cb->>'max_attempts')::smallint, 25), coalesce(cb->'args', '{}'), + coalesce(cb->'metadata', '{}') || jsonb_build_object('batch_id', b.id::text, 'batch_failed', b.failed) + FROM batches b, LATERAL (VALUES (CASE WHEN b.failed = 0 THEN b.on_success ELSE b.on_failure END), (b.on_complete)) AS v(cb) + WHERE b.pending <= 0 AND cb IS NOT NULL + RETURNING pg_notify(` + insertChannelParam + `, queue) +), +promoted AS ( + UPDATE hopper_jobs j + SET state = CASE WHEN j.scheduled_at > now() THEN 'scheduled' ELSE 'available' END::hopper_job_state + WHERE j.state = 'pending' + AND j.id IN (SELECT d.job_id FROM hopper_job_deps d JOIN done ON d.depends_on = done.id) + AND j.id NOT IN (SELECT id FROM cascaded) + AND NOT EXISTS ( + SELECT 1 FROM hopper_job_deps d JOIN hopper_jobs x ON x.id = d.depends_on + WHERE d.job_id = j.id AND x.id NOT IN (SELECT id FROM done)) + RETURNING pg_notify(` + insertChannelParam + `, j.queue) +), +edges AS ( + DELETE FROM hopper_job_deps d WHERE d.job_id IN (SELECT id FROM done) RETURNING d.job_id +)` +} + +// workflowJobInsertSQL inserts a workflow's jobs with the IDs generated up +// front, pending when they have dependencies. RETURNING follows input order. +var workflowJobInsertSQL = ` +WITH p AS ( + SELECT * FROM unnest( + $1::uuid[], $2::text[], $3::text[], $4::smallint[], $5::smallint[], $6::timestamptz[], $7::jsonb[], $8::jsonb[], + $9::float8[], $10::boolean[], $11::text[], $12::text[], $13::boolean[] + ) AS p(id, kind, queue, priority, max_attempts, scheduled_at, args, metadata, ttl, await, ordering_key, partition_key, pending) +) +INSERT INTO hopper_jobs (id, kind, queue, state, priority, max_attempts, scheduled_at, args, metadata, expires_at, await, ordering_key, partition_key, batch_id) +SELECT p.id, p.kind, p.queue, + CASE WHEN p.pending THEN 'pending' WHEN p.scheduled_at > now() THEN 'scheduled' ELSE 'available' END::hopper_job_state, + p.priority, p.max_attempts, coalesce(p.scheduled_at, now()), p.args, p.metadata, + CASE WHEN p.ttl > 0 THEN now() + make_interval(secs => p.ttl) END, p.await, p.ordering_key, p.partition_key, $14::uuid +FROM p +RETURNING ` + JobColumns("") + `, false AS duplicate` + +// WorkflowInsert implements driver.Executor. +func (e *Executor) WorkflowInsert(ctx context.Context, params driver.WorkflowInsertParams) (*driver.WorkflowInsertResult, error) { + n := len(params.Jobs) + if n == 0 { + return nil, errors.New("hopper: insert workflow: no jobs") + } + pending := make([]bool, n) + for _, d := range params.Deps { + if d.Job < 0 || d.Job >= n || d.DependsOn < 0 || d.DependsOn >= n || d.Job == d.DependsOn { + return nil, fmt.Errorf("hopper: insert workflow: dependency %d -> %d is out of range", d.Job, d.DependsOn) + } + pending[d.Job] = true + } + var ( + kinds, queues, ordering, partitions = make([]string, n), make([]string, n), make([]string, n), make([]string, n) + priorities, maxAttempts = make([]int, n), make([]int, n) + scheduledAt = make([]time.Time, n) + args, metadata = make([][]byte, n), make([][]byte, n) + ttls = make([]float64, n) + awaits = make([]bool, n) + queueSet = map[string]struct{}{} + notifyQueues []string + ) + for i, p := range params.Jobs { + if p.UniqueKey != "" { + return nil, fmt.Errorf("hopper: insert workflow: job %d has a unique key", i) + } + var err error + if priorities[i], err = smallint(p.Priority, "priority"); err != nil { + return nil, err + } + if maxAttempts[i], err = smallint(p.MaxAttempts, "max_attempts"); err != nil { + return nil, err + } + kinds[i], queues[i], scheduledAt[i] = p.Kind, p.Queue, p.ScheduledAt + args[i], metadata[i] = []byte(jsonOrEmptyObject(p.Args)), []byte(jsonOrEmptyObject(p.Metadata)) + ttls[i], awaits[i], ordering[i], partitions[i] = p.TTL.Seconds(), p.Await, p.OrderingKey, p.PartitionKey + if _, seen := queueSet[p.Queue]; !pending[i] && !seen { + queueSet[p.Queue] = struct{}{} + notifyQueues = append(notifyQueues, p.Queue) + } + } + onSuccess, onFailure, onComplete, err := callbacksJSON(params.OnSuccess, params.OnFailure, params.OnComplete) + if err != nil { + return nil, err + } + + res := &driver.WorkflowInsertResult{} + err = e.withTx(ctx, func(tx Tx) error { + ids, err := scanIDs(tx.Query(ctx, "SELECT hopper_uuidv7() FROM generate_series(1, $1)", n)) + if err != nil { + return fmt.Errorf("generate ids: %w", err) + } + // The graph, for inspection, and the working set of edges. + edges := make([][2]string, len(params.Deps)) + jobIDs, depIDs, policies := make([]driver.JobID, len(params.Deps)), make([]driver.JobID, len(params.Deps)), make([]string, len(params.Deps)) + for i, d := range params.Deps { + jobIDs[i], depIDs[i] = ids[d.Job], ids[d.DependsOn] + edges[i] = [2]string{ids[d.Job].String(), ids[d.DependsOn].String()} + policies[i] = string(d.OnFailure) + if d.OnFailure == "" { + policies[i] = string(driver.DependencyCancel) + } + } + edgesJSON, err := json.Marshal(edges) + if err != nil { + return err + } + err = tx.QueryRow(ctx, ` + INSERT INTO hopper_batches (name, pending, total, on_success, on_failure, on_complete, metadata, edges) + VALUES ($1, $2, $2, $3::jsonb, $4::jsonb, $5::jsonb, $6::jsonb, $7::jsonb) RETURNING id`, + nullable(params.Name), n, onSuccess, onFailure, onComplete, jsonOrEmptyObject(params.Metadata), string(edgesJSON)).Scan(&res.ID) + if err != nil { + return fmt.Errorf("insert batch: %w", err) + } + res.Jobs, err = scanInsertResults(tx.Query(ctx, workflowJobInsertSQL, + uuidArray(ids), textArray(kinds), textArray(queues), intArray(priorities), intArray(maxAttempts), timeArray(scheduledAt), + jsonArray(args), jsonArray(metadata), floatArray(ttls), boolArray(awaits), + nullableTextArray(ordering), nullableTextArray(partitions), boolArray(pending), uuidParam(res.ID))) + if err != nil { + return fmt.Errorf("insert jobs: %w", err) + } + if len(params.Deps) > 0 { + if _, err := tx.Exec(ctx, `INSERT INTO hopper_job_deps (job_id, depends_on, on_failure) + SELECT * FROM unnest($1::uuid[], $2::uuid[], $3::text[]) ON CONFLICT DO NOTHING`, + uuidArray(jobIDs), uuidArray(depIDs), textArray(policies)); err != nil { + return fmt.Errorf("insert dependencies: %w", err) + } + } + if params.Notify && len(notifyQueues) > 0 { + if _, err := tx.Exec(ctx, NotifySQL, driver.ChannelInsert, textArray(notifyQueues)); err != nil { + return err + } + } + return nil + }) + if err != nil { + return nil, fmt.Errorf("hopper: insert workflow: %w", err) + } + return res, nil +} + +func callbacksJSON(cbs ...*driver.JobInsertParams) (onSuccess, onFailure, onComplete *string, err error) { + out := make([]*string, len(cbs)) + for i, cb := range cbs { + if out[i], err = callbackJSON(cb); err != nil { + return nil, nil, nil, err + } + } + return out[0], out[1], out[2], nil +} + +var workflowJobsSQL = fmt.Sprintf(` +SELECT %s, finalized_at, output FROM ( + SELECT %s, seq, NULL::timestamptz AS finalized_at, NULL::jsonb AS output FROM hopper_jobs WHERE batch_id = $1 + UNION ALL + SELECT %s, seq, finalized_at, output FROM hopper_job_history WHERE batch_id = $1 +) u ORDER BY seq`, JobColumns(""), JobColumns(""), JobColumns("")) + +// WorkflowGet implements driver.Executor. +func (e *Executor) WorkflowGet(ctx context.Context, id driver.JobID) (*driver.WorkflowRow, error) { + var ( + w driver.WorkflowRow + name sql.NullString + completed sql.NullTime + metadata jsonText + edges jsonText + ) + err := e.Conn.QueryRow(ctx, `SELECT id, name, pending, failed, total, created_at, completed_at, metadata, edges FROM hopper_batches WHERE id = $1`, uuidParam(id)). + Scan(&w.Batch.ID, &name, &w.Batch.Pending, &w.Batch.Failed, &w.Batch.Total, &w.Batch.CreatedAt, &completed, &metadata, &edges) + if errors.Is(err, ErrNoRows) { + return nil, driver.ErrNotFound + } + if err != nil { + return nil, fmt.Errorf("hopper: get workflow: %w", err) + } + w.Batch.Name, w.Batch.CompletedAt, w.Batch.Metadata = name.String, completed.Time, metadata.raw() + if edges != "" { + var pairs [][2]driver.JobID + if err := json.Unmarshal([]byte(edges), &pairs); err != nil { + return nil, fmt.Errorf("hopper: get workflow: edges: %w", err) + } + w.Edges = make([]driver.JobEdge, len(pairs)) + for i, p := range pairs { + w.Edges[i] = driver.JobEdge{Job: p[0], DependsOn: p[1]} + } + } + if w.Jobs, err = historyJobs(e.Conn.Query(ctx, workflowJobsSQL, uuidParam(id))); err != nil { + return nil, fmt.Errorf("hopper: get workflow: jobs: %w", err) + } + return &w, nil +} diff --git a/drivertest/drivertest.go b/drivertest/drivertest.go index 67a7803..06e3356 100644 --- a/drivertest/drivertest.go +++ b/drivertest/drivertest.go @@ -88,6 +88,7 @@ func Run[TTx any](t *testing.T, f Fixture[TTx]) { {"PartitionLimit", testPartitionLimit[TTx]}, {"Aging", testAging[TTx]}, {"Batches", testBatches[TTx]}, + {"Workflows", testWorkflows[TTx]}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { diff --git a/drivertest/workflow.go b/drivertest/workflow.go new file mode 100644 index 0000000..2c527de --- /dev/null +++ b/drivertest/workflow.go @@ -0,0 +1,208 @@ +package drivertest + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + + "github.com/parallelworks/hopper/driver" +) + +func testWorkflows[TTx any](t *testing.T, f Fixture[TTx]) { + ctx := context.Background() + d := f.NewDriver(t) + exec := d.Executor() + clientID := register(ctx, t, exec) + cb := func(kind string) *driver.JobInsertParams { + return &driver.JobInsertParams{Kind: kind, Queue: "callbacks", Priority: 2, MaxAttempts: 3, Args: json.RawMessage(`{}`)} + } + state := func(id driver.JobID) driver.JobState { + t.Helper() + j, err := exec.JobGet(ctx, id) + if err != nil { + t.Fatal(err) + } + return j.State + } + complete := func(jobs ...*driver.JobRow) { + t.Helper() + fin := make([]driver.JobFinalize, len(jobs)) + for i, j := range jobs { + fin[i] = driver.JobFinalize{ID: j.ID, AttemptedBy: clientID, State: driver.JobStateCompleted, Archive: true} + } + if applied := finalize(ctx, t, exec, fin...); len(applied) != len(jobs) { + t.Fatalf("applied %d of %d", len(applied), len(jobs)) + } + } + callbacks := func() map[string]*driver.JobRow { + t.Helper() + out, err := exec.JobList(ctx, driver.JobListParams{Queue: "callbacks", Limit: 10}) + if err != nil { + t.Fatal(err) + } + m := map[string]*driver.JobRow{} + for _, j := range out { + m[j.Kind] = j + } + return m + } + + // A diamond: b and c after a, d after b and c. + wf, err := exec.WorkflowInsert(ctx, driver.WorkflowInsertParams{ + Name: "diamond", Jobs: params("step", 4), + Deps: []driver.JobDependency{{Job: 1, DependsOn: 0}, {Job: 2, DependsOn: 0}, {Job: 3, DependsOn: 1}, {Job: 3, DependsOn: 2}}, + OnSuccess: cb("wf-done"), OnFailure: cb("wf-alert"), Metadata: json.RawMessage(`{"run":1}`), + }) + if err != nil { + t.Fatal(err) + } + if len(wf.Jobs) != 4 || wf.ID.IsZero() { + t.Fatalf("workflow result = %+v", wf) + } + a, b, c, dd := wf.Jobs[0].Job, wf.Jobs[1].Job, wf.Jobs[2].Job, wf.Jobs[3].Job + for i, j := range wf.Jobs { + if argsN(j.Job) != i || j.Job.BatchID != wf.ID || j.Duplicate { + t.Errorf("job %d = %+v", i, j.Job) + } + } + if a.State != driver.JobStateAvailable || b.State != driver.JobStatePending || c.State != driver.JobStatePending || dd.State != driver.JobStatePending { + t.Errorf("states = %s %s %s %s", a.State, b.State, c.State, dd.State) + } + row, err := exec.WorkflowGet(ctx, wf.ID) + if err != nil { + t.Fatal(err) + } + if row.Batch.Name != "diamond" || row.Batch.Pending != 4 || row.Batch.Total != 4 || len(row.Jobs) != 4 || len(row.Edges) != 4 { + t.Fatalf("workflow = %+v", row) + } + if row.Jobs[0].ID != a.ID || row.Jobs[3].ID != dd.ID || row.Edges[0] != (driver.JobEdge{Job: b.ID, DependsOn: a.ID}) { + t.Errorf("workflow jobs/edges out of order: %+v", row) + } + + // Only the root is claimable. Completing it promotes b and c, not d. + running := claim(ctx, t, exec, clientID, 10) + if len(running) != 1 || running[0].ID != a.ID { + t.Fatalf("claimed %v, want only the root", running) + } + complete(running[0]) + if state(b.ID) != driver.JobStateAvailable || state(c.ID) != driver.JobStateAvailable || state(dd.ID) != driver.JobStatePending { + t.Errorf("after a: b %s c %s d %s", state(b.ID), state(c.ID), state(dd.ID)) + } + running = claim(ctx, t, exec, clientID, 10) + if len(running) != 2 { + t.Fatalf("claimed %d, want b and c", len(running)) + } + complete(running[0]) + if state(dd.ID) != driver.JobStatePending { + t.Errorf("d promoted with one dependency left") + } + complete(running[1]) + if state(dd.ID) != driver.JobStateAvailable { + t.Errorf("d not promoted after b and c") + } + running = claim(ctx, t, exec, clientID, 10) + if len(running) != 1 || running[0].ID != dd.ID { + t.Fatalf("claimed %v, want d", running) + } + if cbs := callbacks(); len(cbs) != 0 { + t.Errorf("callbacks before completion: %v", cbs) + } + complete(running[0]) + row, _ = exec.WorkflowGet(ctx, wf.ID) + if row.Batch.Pending != 0 || row.Batch.Failed != 0 || row.Batch.CompletedAt.IsZero() || len(row.Jobs) != 4 || len(row.Edges) != 4 { + t.Errorf("completed workflow = %+v", row.Batch) + } + for _, j := range row.Jobs { + if j.State != driver.JobStateCompleted || j.FinalizedAt.IsZero() { + t.Errorf("finished job = %+v", j) + } + } + if cbs := callbacks(); len(cbs) != 1 || cbs["wf-done"] == nil { + t.Errorf("callbacks = %v", cbs) + } + + // A failure cascades: e -> f -> g are cancelled below e, while h, which + // ignores f's failure, runs once f has finished. + wf2, err := exec.WorkflowInsert(ctx, driver.WorkflowInsertParams{ + Name: "chain", Jobs: params("chain", 4), + Deps: []driver.JobDependency{ + {Job: 1, DependsOn: 0}, {Job: 2, DependsOn: 1}, {Job: 3, DependsOn: 1, OnFailure: driver.DependencyIgnore}, + }, + OnSuccess: cb("wf-done2"), OnFailure: cb("wf-alert2"), + }) + if err != nil { + t.Fatal(err) + } + e, ff, g, h := wf2.Jobs[0].Job, wf2.Jobs[1].Job, wf2.Jobs[2].Job, wf2.Jobs[3].Job + running = claim(ctx, t, exec, clientID, 10) + if len(running) != 1 || running[0].ID != e.ID { + t.Fatalf("claimed %v, want e", running) + } + finalize(ctx, t, exec, driver.JobFinalize{ID: e.ID, AttemptedBy: clientID, State: driver.JobStateDiscarded, Error: &driver.AttemptError{Attempt: 1, Error: "boom"}}) + if state(ff.ID) != driver.JobStateCancelled || state(g.ID) != driver.JobStateCancelled { + t.Errorf("after e failed: f %s g %s, want cancelled", state(ff.ID), state(g.ID)) + } + if state(h.ID) != driver.JobStateAvailable { + t.Errorf("h = %s, want available (ignores f's failure)", state(h.ID)) + } + if j, _ := exec.JobGet(ctx, g.ID); len(j.Errors) != 1 || !strings.Contains(j.Errors[0].Error, "dependency failed") || j.Errors[0].At.IsZero() { + t.Errorf("cancelled dependent errors = %+v", j.Errors) + } + row, _ = exec.WorkflowGet(ctx, wf2.ID) + if row.Batch.Pending != 1 || row.Batch.Failed != 3 { + t.Errorf("chain after cascade = %+v", row.Batch) + } + running = claim(ctx, t, exec, clientID, 10) + if len(running) != 1 || running[0].ID != h.ID { + t.Fatalf("claimed %v, want h", running) + } + complete(running[0]) + row, _ = exec.WorkflowGet(ctx, wf2.ID) + if row.Batch.Pending != 0 || row.Batch.Failed != 3 || row.Batch.CompletedAt.IsZero() { + t.Errorf("chain completed = %+v", row.Batch) + } + cbs := callbacks() + if cbs["wf-alert2"] == nil || cbs["wf-done2"] != nil { + t.Errorf("chain callbacks = %v", cbs) + } else if !strings.Contains(string(cbs["wf-alert2"].Metadata), `"batch_failed": 3`) { + t.Errorf("alert metadata = %s", cbs["wf-alert2"].Metadata) + } + + // Cancelling a pending job cascades too, and counts as a failure. + wf3, err := exec.WorkflowInsert(ctx, driver.WorkflowInsertParams{ + Jobs: params("pair", 3), Deps: []driver.JobDependency{{Job: 1, DependsOn: 0}, {Job: 2, DependsOn: 1}}, + }) + if err != nil { + t.Fatal(err) + } + if j, err := exec.JobCancel(ctx, wf3.Jobs[1].Job.ID); err != nil || j.State != driver.JobStateCancelled { + t.Fatalf("cancel pending = %+v, %v", j, err) + } + if state(wf3.Jobs[2].Job.ID) != driver.JobStateCancelled { + t.Errorf("dependent of a cancelled job = %s", state(wf3.Jobs[2].Job.ID)) + } + running = claim(ctx, t, exec, clientID, 10) + if len(running) != 1 { + t.Fatalf("claimed %d, want the root", len(running)) + } + complete(running[0]) + row, _ = exec.WorkflowGet(ctx, wf3.ID) + if row.Batch.Name != "" || row.Batch.Pending != 0 || row.Batch.Failed != 2 || row.Batch.CompletedAt.IsZero() { + t.Errorf("pair = %+v", row.Batch) + } + + // Validation. + p := params("bad", 2) + p[1].UniqueKey = "k" + if _, err := exec.WorkflowInsert(ctx, driver.WorkflowInsertParams{Jobs: p}); err == nil { + t.Error("unique key accepted") + } + if _, err := exec.WorkflowInsert(ctx, driver.WorkflowInsertParams{Jobs: params("bad", 2), Deps: []driver.JobDependency{{Job: 1, DependsOn: 5}}}); err == nil { + t.Error("dangling dependency accepted") + } + if _, err := exec.WorkflowGet(ctx, driver.JobID{9}); !errors.Is(err, driver.ErrNotFound) { + t.Errorf("unknown workflow = %v", err) + } +} diff --git a/hoppermigrate/migrations/004_workflows.down.sql b/hoppermigrate/migrations/004_workflows.down.sql new file mode 100644 index 0000000..7b4b0a1 --- /dev/null +++ b/hoppermigrate/migrations/004_workflows.down.sql @@ -0,0 +1,3 @@ +DROP INDEX hopper_job_history_batch; +DROP TABLE hopper_job_deps; +ALTER TABLE hopper_batches DROP COLUMN edges, DROP COLUMN name; diff --git a/hoppermigrate/migrations/004_workflows.up.sql b/hoppermigrate/migrations/004_workflows.up.sql new file mode 100644 index 0000000..bc5b6d9 --- /dev/null +++ b/hoppermigrate/migrations/004_workflows.up.sql @@ -0,0 +1,25 @@ +-- hopper schema v4: workflows. +-- +-- A workflow is a batch whose jobs depend on each other. Dependents wait in +-- the 'pending' state; the statement that finalizes a job promotes the +-- dependents whose dependencies are all finalized, and cancels the +-- dependents of a failed job (transitively) unless their edge says to +-- ignore the failure. hopper_job_deps is the working set of unsatisfied +-- edges, deleted as dependents finalize; the batch row keeps the whole +-- graph for inspection. + +ALTER TABLE hopper_batches + ADD COLUMN name text, + ADD COLUMN edges jsonb; -- [[job_id, depends_on], ...] for workflows + +CREATE TABLE hopper_job_deps ( + job_id uuid NOT NULL, -- the dependent, pending until depends_on is finalized + depends_on uuid NOT NULL, + on_failure text NOT NULL DEFAULT 'cancel', -- 'cancel' the dependent when depends_on fails, or 'ignore' + PRIMARY KEY (job_id, depends_on) +); + +CREATE INDEX hopper_job_deps_depends_on ON hopper_job_deps (depends_on); + +-- Workflow inspection reads a batch's finished jobs from history. +CREATE INDEX hopper_job_history_batch ON hopper_job_history (batch_id) WHERE batch_id IS NOT NULL; diff --git a/workflow.go b/workflow.go new file mode 100644 index 0000000..cae15e1 --- /dev/null +++ b/workflow.go @@ -0,0 +1,209 @@ +package hopper + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/parallelworks/hopper/driver" +) + +// DependencyFailure says what happens to a step when a step it depends on +// is cancelled or discarded. +type DependencyFailure = driver.DependencyFailure + +const ( + // DependencyCancel cancels the dependent step, and its dependents in + // turn. It is the default. + DependencyCancel = driver.DependencyCancel + // DependencyIgnore runs the dependent step once the steps it depends on + // have all finished, whatever their outcome. + DependencyIgnore = driver.DependencyIgnore +) + +// WorkflowOpts sets a workflow's completion callbacks and metadata. A +// workflow is a batch, so the callbacks are those of BatchOpts: OnSuccess +// when every step completed, OnFailure when any was cancelled or +// discarded, OnComplete either way. +type WorkflowOpts struct { + OnSuccess JobArgs + OnFailure JobArgs + OnComplete JobArgs + // CallbackOpts applies to every callback job. + CallbackOpts *InsertOpts + // Metadata is stored with the workflow. + Metadata json.RawMessage +} + +// StepOpts sets a step's dependencies and insert options. +type StepOpts struct { + // After lists the steps this one waits for. It is inserted pending and + // becomes available once they have all finished. + After []Step + // OnDependencyFailure defaults to DependencyCancel. + OnDependencyFailure DependencyFailure + // Insert holds the job's own options. + Insert *InsertOpts +} + +// After returns options for a step that waits for the given steps. +func After(steps ...Step) *StepOpts { + return &StepOpts{After: steps} +} + +// Workflow collects steps with dependencies between them, to insert +// together. Steps are jobs; a step with dependencies waits, pending, until +// they have finished. Because a step can only depend on steps added before +// it, a workflow is always a DAG. +type Workflow struct { + name string + opts WorkflowOpts + steps []workflowStep +} + +type workflowStep struct { + params InsertParams + opts StepOpts +} + +// Step identifies a step of a workflow, for other steps to depend on. +type Step struct { + wf *Workflow + i int +} + +// WorkflowResult reports an inserted workflow. +type WorkflowResult struct { + ID JobID + // Jobs are in step order. + Jobs []*InsertResult +} + +// WorkflowRow is a workflow's progress, jobs and graph. +type WorkflowRow = driver.WorkflowRow + +// NewWorkflow starts a workflow. Add steps to it, then insert it with +// Client.InsertWorkflow or InsertWorkflowTx. +func NewWorkflow(name string, opts *WorkflowOpts) *Workflow { + w := &Workflow{name: name} + if opts != nil { + w.opts = *opts + } + return w +} + +// Add adds a step. opts may be nil for a step with no dependencies; +// After(steps...) is the usual way to declare them. +func (w *Workflow) Add(args JobArgs, opts *StepOpts) Step { + s := workflowStep{params: InsertParams{Args: args}} + if opts != nil { + s.opts = *opts + s.params.Opts = opts.Insert + } + w.steps = append(w.steps, s) + return Step{wf: w, i: len(w.steps) - 1} +} + +// InsertWorkflow inserts a workflow's steps and dependencies atomically. +func (c *Client[TTx]) InsertWorkflow(ctx context.Context, wf *Workflow) (*WorkflowResult, error) { + return c.insertWorkflow(ctx, c.exec, wf, false) +} + +// InsertWorkflowTx is InsertWorkflow in the caller's transaction. +func (c *Client[TTx]) InsertWorkflowTx(ctx context.Context, tx TTx, wf *Workflow) (*WorkflowResult, error) { + return c.insertWorkflow(ctx, c.driver.UnwrapTx(tx), wf, true) +} + +func (c *Client[TTx]) insertWorkflow(ctx context.Context, exec driver.Executor, wf *Workflow, inTx bool) (*WorkflowResult, error) { + if wf == nil || len(wf.steps) == 0 { + return nil, errors.New("hopper: workflow has no steps") + } + now := time.Now() + wp := driver.WorkflowInsertParams{Name: wf.name, Metadata: wf.opts.Metadata, Notify: inTx} + for _, cb := range []struct { + args JobArgs + dst **driver.JobInsertParams + }{{wf.opts.OnSuccess, &wp.OnSuccess}, {wf.opts.OnFailure, &wp.OnFailure}, {wf.opts.OnComplete, &wp.OnComplete}} { + if cb.args == nil { + continue + } + p, _, err := c.buildInsertParams(InsertParams{Args: cb.args, Opts: wf.opts.CallbackOpts}, now) + if err != nil { + return nil, fmt.Errorf("hopper: workflow callback: %w", err) + } + *cb.dst = &p + } + params := make([]InsertParams, len(wf.steps)) + for i, s := range wf.steps { + params[i] = s.params + for _, dep := range s.opts.After { + if dep.wf != wf || dep.i >= i { + return nil, fmt.Errorf("hopper: step %d depends on a step of another workflow", i) + } + wp.Deps = append(wp.Deps, driver.JobDependency{Job: i, DependsOn: dep.i, OnFailure: s.opts.OnDependencyFailure}) + } + } + + // The steps go through the insert middleware like any other insert, + // with the workflow insert as the innermost step. + res := &WorkflowResult{} + next := func(ctx context.Context, params []InsertParams) ([]*InsertResult, error) { + if len(params) != len(wf.steps) { + return nil, errors.New("hopper: insert middleware changed the number of workflow steps") + } + wp.Jobs = make([]driver.JobInsertParams, len(params)) + for i, p := range params { + dp, _, err := c.buildInsertParams(p, now) + if err != nil { + return nil, fmt.Errorf("step %d: %w", i, err) + } + if dp.UniqueKey != "" { + return nil, fmt.Errorf("hopper: step %d: workflow steps cannot be unique", i) + } + wp.Jobs[i] = dp + } + out, err := exec.WorkflowInsert(ctx, wp) + if err != nil { + return nil, err + } + res.ID = out.ID + results := make([]*InsertResult, len(out.Jobs)) + for i, r := range out.Jobs { + results[i] = &InsertResult{Job: r.Job} + } + return results, nil + } + for i := len(c.cfg.Middleware) - 1; i >= 0; i-- { + m, inner := c.cfg.Middleware[i], next + next = func(ctx context.Context, params []InsertParams) ([]*InsertResult, error) { + return m.Insert(ctx, params, inner) + } + } + results, err := next(ctx, params) + if err != nil { + return nil, err + } + res.Jobs = results + + if !inTx { + queues := map[string]struct{}{} + for _, r := range results { + if r.Job.State != driver.JobStatePending { + queues[r.Job.Queue] = struct{}{} + } + } + for q := range queues { + c.wakeQueue(q) + c.notifier.mark(q) + } + } + return res, nil +} + +// WorkflowGet returns a workflow's progress, its jobs (live and finished) +// and its edges. +func (c *Client[TTx]) WorkflowGet(ctx context.Context, id JobID) (*WorkflowRow, error) { + return c.exec.WorkflowGet(ctx, id) +} diff --git a/workflow_test.go b/workflow_test.go new file mode 100644 index 0000000..61ab031 --- /dev/null +++ b/workflow_test.go @@ -0,0 +1,162 @@ +package hopper_test + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/parallelworks/hopper" +) + +type wfStep struct { + Name string `json:"name"` + Fail bool `json:"fail,omitempty"` +} + +func (wfStep) Kind() string { return "wf_step" } + +// stepLog records when each step ran. +type stepLog struct { + mu sync.Mutex + order []string + at map[string]time.Time +} + +func (l *stepLog) record(name string) { + l.mu.Lock() + defer l.mu.Unlock() + if l.at == nil { + l.at = map[string]time.Time{} + } + l.order = append(l.order, name) + l.at[name] = time.Now() +} + +func (l *stepLog) before(a, b string) bool { + l.mu.Lock() + defer l.mu.Unlock() + ta, oka := l.at[a] + tb, okb := l.at[b] + return oka && okb && ta.Before(tb) +} + +func (l *stepLog) ran(name string) bool { + l.mu.Lock() + defer l.mu.Unlock() + _, ok := l.at[name] + return ok +} + +func TestWorkflow(t *testing.T) { + t.Parallel() + h := newHarness(t) + ctx := context.Background() + var ( + log stepLog + mu sync.Mutex + callbacks []string + ) + workers := hopper.NewWorkers() + hopper.AddWorkFunc(workers, func(_ context.Context, job *hopper.Job[wfStep]) error { + log.record(job.Args.Name) + if job.Args.Fail { + return hopper.Cancel(errors.New("step failed")) + } + return nil + }) + hopper.AddWorkFunc(workers, func(_ context.Context, job *hopper.Job[reportDone]) error { + mu.Lock() + callbacks = append(callbacks, "done") + mu.Unlock() + return nil + }) + hopper.AddWorkFunc(workers, func(_ context.Context, job *hopper.Job[alertOps]) error { + mu.Lock() + callbacks = append(callbacks, "alert") + mu.Unlock() + return nil + }) + c := h.started(workers, 4) + + // fetch -> parse -> {index, notify}, inside a transaction. + wf := hopper.NewWorkflow("ingest-9", &hopper.WorkflowOpts{OnSuccess: reportDone{RunID: 9}, OnFailure: alertOps{RunID: 9}}) + fetch := wf.Add(wfStep{Name: "fetch"}, nil) + parse := wf.Add(wfStep{Name: "parse"}, hopper.After(fetch)) + wf.Add(wfStep{Name: "index"}, hopper.After(parse)) + wf.Add(wfStep{Name: "notify"}, &hopper.StepOpts{After: []hopper.Step{parse}, Insert: &hopper.InsertOpts{Priority: hopper.PriorityHigh}}) + tx, err := h.pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + res, err := c.InsertWorkflowTx(ctx, tx, wf) + if err != nil { + t.Fatal(err) + } + if err := tx.Commit(ctx); err != nil { + t.Fatal(err) + } + if len(res.Jobs) != 4 || res.Jobs[0].Job.State != hopper.JobStateAvailable || res.Jobs[1].Job.State != hopper.JobStatePending || + res.Jobs[3].Job.Priority != int(hopper.PriorityHigh) || res.Jobs[1].Job.BatchID != res.ID { + t.Fatalf("workflow result = %+v", res) + } + waitFor(t, func() bool { mu.Lock(); defer mu.Unlock(); return len(callbacks) == 1 }) + if !log.before("fetch", "parse") || !log.before("parse", "index") || !log.before("parse", "notify") { + t.Errorf("steps ran out of order: %v", log.order) + } + row, err := c.WorkflowGet(ctx, res.ID) + if err != nil || row.Batch.Name != "ingest-9" || row.Batch.Pending != 0 || row.Batch.Failed != 0 || len(row.Jobs) != 4 || len(row.Edges) != 3 { + t.Errorf("workflow = %+v, %v", row, err) + } + mu.Lock() + if callbacks[0] != "done" { + t.Errorf("callbacks = %v", callbacks) + } + callbacks = nil + mu.Unlock() + + // A failing step cancels what depends on it and runs the failure + // callback; an independent branch still runs. + wf = hopper.NewWorkflow("ingest-10", &hopper.WorkflowOpts{OnSuccess: reportDone{RunID: 10}, OnFailure: alertOps{RunID: 10}}) + a := wf.Add(wfStep{Name: "a"}, nil) + b := wf.Add(wfStep{Name: "b", Fail: true}, hopper.After(a)) + wf.Add(wfStep{Name: "c"}, hopper.After(b)) + wf.Add(wfStep{Name: "d"}, hopper.After(a)) + res, err = c.InsertWorkflow(ctx, wf) + if err != nil { + t.Fatal(err) + } + waitFor(t, func() bool { mu.Lock(); defer mu.Unlock(); return len(callbacks) == 1 }) + cj := waitForJob(t, c, res.Jobs[2].Job.ID, hopper.JobStateCancelled) + if len(cj.Errors) != 1 || cj.Errors[0].Error != "hopper: dependency failed" { + t.Errorf("cancelled step = %+v", cj.Errors) + } + if log.ran("c") || !log.ran("d") { + t.Errorf("ran %v; c must not run, d must", log.order) + } + row, err = c.WorkflowGet(ctx, res.ID) + if err != nil || row.Batch.Failed != 2 || row.Batch.CompletedAt.IsZero() { + t.Errorf("failed workflow = %+v, %v", row.Batch, err) + } + mu.Lock() + if callbacks[0] != "alert" { + t.Errorf("callbacks = %v", callbacks) + } + mu.Unlock() + + // Validation. + if _, err := c.InsertWorkflow(ctx, hopper.NewWorkflow("empty", nil)); err == nil { + t.Error("empty workflow accepted") + } + other := hopper.NewWorkflow("other", nil) + other.Add(wfStep{Name: "x"}, hopper.After(a)) + if _, err := c.InsertWorkflow(ctx, other); err == nil { + t.Error("step from another workflow accepted") + } + wf = hopper.NewWorkflow("unique", nil) + wf.Add(wfStep{Name: "u"}, &hopper.StepOpts{Insert: &hopper.InsertOpts{Unique: &hopper.UniqueOpts{}}}) + if _, err := c.InsertWorkflow(ctx, wf); err == nil { + t.Error("unique step accepted") + } +}