diff --git a/README.md b/README.md index f3fe750..b3fe66f 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,49 @@ result, execution of the graph is performed as a parallelised breadth first sear This package contains some creative APIs which work around limitations in Golang generics; chief of these is the `ID` type used in place of slices of keys because Golang slices are invariant. +## Overriding tasks in tests + +A graph's tasks often create things for themselves: clients of other services, say. That makes a +whole graph hard to test, as there is no way to give it a fake client. `taskgraphtest` can replace +the task which provides a key: + +```go +taskgraphtest.RunWithOverrides(ctx, graph, inputs, []tg.Binding{keyClient.Bind(fakeClient)}) +``` + +The task providing `keyClient` is not run, and the tasks which depend on it see the fake. This makes +a graph's dependencies a matter of which keys its tasks provide: anything which a test may want to +replace should be a key of its own. + +Replacing tasks is only available from `taskgraphtest`. `Graph.Run` has no way to do it, and the +engine's support is in an `internal` package, which nothing outside this module can import. Code which +runs a graph and needs to be tested can take a function for running graphs, and be handed +`Overrider.Run` by its tests. + +* An input for a key which a task provides is still an error, so a key is only ever replaced on + purpose. The same goes for overriding one key twice, or overriding a key which is also an input. +* If a task provides several keys they must all be overridden, or none (`ErrPartialOverride`). + Otherwise the task would still have to run, with all of its effects, with some of its results + thrown away. +* Overriding a key which no task provides is an error (`ErrUnknownOverride`), which catches + overrides left behind by a task which has since been renamed or removed. `CheckWithOverrides` + finds these without running the graph. +* The override may be absent: `keyClient.BindError(err)` replaces the task with a failure, to see how + the tasks which depend on it cope. +* The overrides are included in the `Binder` returned, as the results of the tasks they replace would + have been. +* Inputs which only a replaced task needed are not required. +* Tasks which only produced inputs for a replaced task still run, as the graph can't know that its + caller doesn't want their results. +* Overrides apply to the graph they are given. A graph run as a task with `AsTask` is replaced as a + whole, by overriding every key it exposes, but its tasks can't be overridden individually. +* An `Overrider` holds one set of overrides for something which runs several graphs, such as an + operator with a graph to create and a graph to delete. Each graph is run with the overrides for the + keys it provides, and `Overrider.Check` makes sure that every override is for some graph. + +`taskgraphtest.Test` and `taskgraphtest.Suite` take `Overrides`, those of a `Test` replacing those +of its `Suite` for the same key. + ## Limitations and sharp edges * Taskgraph has no deadlock detection. In theory the fact that the graph is acyclic should prevent diff --git a/graph.go b/graph.go index efec4e9..10806e1 100644 --- a/graph.go +++ b/graph.go @@ -90,9 +90,16 @@ type Graph interface { type runState struct { Binder signals map[string]chan struct{} + // skipped holds the IDs of the nodes which are not run because their tasks are overridden. They + // have nothing listening for signals, and nothing which depends on them waits for them. + skipped map[string]struct{} } func (rs *runState) signal(ctx context.Context, childID string) (err error) { + if _, skipped := rs.skipped[childID]; skipped { + return nil + } + // Capture (and ignore) panics caused by sends to closed channels, which may occur if a task // starts slowly and sees that all of its dependencies are available before we can signal it. defer func() { @@ -127,6 +134,8 @@ type graphNode struct { const ( traceTaskgraphAbsentKeysPrefix = "taskgraph.absent_keys." + traceTaskgraphOverriddenKeys = "taskgraph.overridden_keys" + traceTaskgraphSkippedTasks = "taskgraph.skipped_tasks" ) // Execute the task against the binder provided in the runState. @@ -262,34 +271,24 @@ type graph struct { logger Logger } -func (g *graph) buildInputBinder(inputs ...Binding) (Binder, error) { - b := NewBinder() - - if err := b.Store(inputs...); err != nil { - return nil, wrapStackErrorf("duplicate input: %w", err) - } - - var missingInputs []string - for requiredInput := range g.allDependencies.Difference(g.allProvided).Iter() { - if !b.Has(requiredInput) { - missingInputs = append(missingInputs, requiredInput.String()) - } - } - if len(missingInputs) > 0 { - return nil, wrapStackErrorf("%w: %s", ErrMissingInputs, strings.Join(missingInputs, ", ")) - } - - return b, nil -} - // Check is Graph.Check. func (g *graph) Check(inputs ...Binding) error { - _, err := g.buildInputBinder(inputs...) + _, err := g.buildInputs(inputs, nil, false) return err } // Run is Graph.Run. -func (g *graph) Run(ctx context.Context, inputs ...Binding) (b Binder, err error) { +func (g *graph) Run(ctx context.Context, inputs ...Binding) (Binder, error) { + return g.runWithOverrides(ctx, inputs, nil, false) +} + +// runWithOverrides runs the graph with some of its tasks replaced by the overrides, which are for +// tests (see taskgraphtest). +func (g *graph) runWithOverrides( + ctx context.Context, + inputs, overrides []Binding, + ignoreUnprovided bool, +) (b Binder, err error) { startTime := time.Now() defer func() { result := "success" @@ -299,20 +298,33 @@ func (g *graph) Run(ctx context.Context, inputs ...Binding) (b Binder, err error executionLatency.WithLabelValues(g.name, result). Observe(float64(time.Since(startTime) / time.Millisecond)) }() - base, err := g.buildInputBinder(inputs...) + in, err := g.buildInputs(inputs, overrides, ignoreUnprovided) if err != nil { return nil, err } + // Overrides stand in for the results of the tasks they replace, so they are stored with the + // results, and are returned along with them. outputs := NewBinder() + if err := outputs.Store(in.overrides...); err != nil { + return nil, wrapStackErrorf("storing overrides: %w", err) + } overlay := &overlayBinder{ - base: base, + base: in.inputs, overlay: outputs, } tCtx, span := g.tracer.Start(ctx, g.name) defer span.End() - if err := g.runWithBinder(tCtx, overlay); err != nil { + if len(in.overrides) > 0 { + span.SetAttributes( + attribute.StringSlice(traceTaskgraphOverriddenKeys, in.overriddenKeys()), + attribute.StringSlice(traceTaskgraphSkippedTasks, in.skippedTaskNames()), + ) + g.logger.Debugf("graph %s overriding keys %v, so not running tasks %v", + g.name, in.overriddenKeys(), in.skippedTaskNames()) + } + if err := g.runWithBinder(tCtx, overlay, in.skipped); err != nil { span.RecordError(err) return nil, err } @@ -322,20 +334,31 @@ func (g *graph) Run(ctx context.Context, inputs ...Binding) (b Binder, err error // Sets up the per-run state of the graph, and runs all of the tasks in their own goroutines until // all have terminated. If any task returns an error, the entire graph run is cancelled. -func (g *graph) runWithBinder(ctx context.Context, binder Binder) error { +func (g *graph) runWithBinder( + ctx context.Context, + binder Binder, + skipped map[*graphNode]struct{}, +) error { rs := &runState{ Binder: binder, signals: map[string]chan struct{}{}, + skipped: map[string]struct{}{}, } for _, gn := range g.nodes { rs.signals[gn.id] = make(chan struct{}) } + for gn := range skipped { + rs.skipped[gn.id] = struct{}{} + } // errgroup always cancels the derived context before returning from Wait(), so the select below // must listen to the parent context's Done() channel. eg, egCtx := errgroup.WithContext(ctx) for _, gn := range g.nodes { + if _, isSkipped := skipped[gn]; isSkipped { + continue + } eg.Go(gn.runFunc(egCtx, rs)) } @@ -371,7 +394,8 @@ func (g *graph) AsTask(exposeKeys ...ID) (Task, error) { exposeKeys: exposeSet, } - if err := g.runWithBinder(ctx, gtb); err != nil { + // Overrides apply only to the graph they are given to, not to any graphs it runs as tasks. + if err := g.runWithBinder(ctx, gtb, nil); err != nil { return nil, err } diff --git a/internal/overridehook/hook.go b/internal/overridehook/hook.go new file mode 100644 index 0000000..b506399 --- /dev/null +++ b/internal/overridehook/hook.go @@ -0,0 +1,26 @@ +// Package overridehook lets taskgraphtest reach the part of the graph engine which replaces tasks. +// +// Replacing tasks is meant for tests, so it is not part of the taskgraph package's API. Go only lets +// packages within this module import an internal package, so this is how taskgraphtest, and nothing +// outside the module, can get at it. The taskgraph package sets the variables when it is +// initialised. +// +// The functions take the arguments typed as any, as this package can't import taskgraph, which +// imports it; taskgraphtest knows the types and converts. +package overridehook + +import "context" + +var ( + // RunWithOverrides runs a graph made by taskgraph.New, as Graph.Run does, with some of its tasks + // replaced by the override bindings, and returns a taskgraph.Binder. If ignoreUnprovided is set, + // overrides for keys which the graph doesn't provide are left out, rather than being an error. + RunWithOverrides func(g any, ctx context.Context, inputs, overrides []any, ignoreUnprovided bool) (any, error) + + // CheckWithOverrides is Graph.Check, with overrides. + CheckWithOverrides func(g any, inputs, overrides []any) error + + // Provides reports whether some task in a graph made by taskgraph.New provides the key (a + // taskgraph.ID). + Provides func(g any, id any) bool +) diff --git a/override.go b/override.go new file mode 100644 index 0000000..1a6ad46 --- /dev/null +++ b/override.go @@ -0,0 +1,208 @@ +package taskgraph + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + + set "github.com/deckarep/golang-set/v2" + "github.com/thought-machine/taskgraph/internal/overridehook" +) + +var ( + // ErrUnknownOverride is returned by taskgraphtest's RunWithOverrides and CheckWithOverrides if an + // override is given for a key which no task in the graph provides. This is almost always a stale + // override, left over from a task which has since been renamed or removed. + ErrUnknownOverride = errors.New("override of a key which no task provides") + + // ErrPartialOverride is returned by taskgraphtest's RunWithOverrides and CheckWithOverrides if + // overrides cover some, but not all, of the keys which a task provides. The task would still have to run, with all of its + // effects, only to have some of its results discarded, which is very unlikely to be what was + // intended. Override all of a task's keys, or none of them. + ErrPartialOverride = errors.New("override of only some of the keys which a task provides") +) + +func init() { + overridehook.RunWithOverrides = func(g any, ctx context.Context, inputs, overrides []any, ignoreUnprovided bool) (any, error) { + graph, err := asGraph(g) + if err != nil { + return nil, err + } + return graph.runWithOverrides( + ctx, + toBindings(inputs), + toBindings(overrides), + ignoreUnprovided, + ) + } + overridehook.CheckWithOverrides = func(g any, inputs, overrides []any) error { + graph, err := asGraph(g) + if err != nil { + return err + } + _, err = graph.buildInputs(toBindings(inputs), toBindings(overrides), false) + return err + } + overridehook.Provides = func(g any, id any) bool { + graph, err := asGraph(g) + if err != nil { + return false + } + return graph.allProvided.Contains(id.(ID)) + } +} + +func asGraph(g any) (*graph, error) { + graph, ok := g.(*graph) + if !ok { + return nil, fmt.Errorf( + "overrides can only be applied to a Graph made by taskgraph.New, not %T", + g, + ) + } + return graph, nil +} + +func toBindings(in []any) []Binding { + out := make([]Binding, 0, len(in)) + for _, b := range in { + out = append(out, b.(Binding)) + } + return out +} + +// runInputs is everything a graph is given to run with, once checked against the graph. +type runInputs struct { + // inputs holds the ordinary inputs. + inputs Binder + // overrides holds the overriding bindings, unwrapped. + overrides []Binding + // skipped holds the nodes of the tasks which the overrides replace. + skipped map[*graphNode]struct{} +} + +// overriddenKeys returns the IDs of the overridden keys, sorted, for reporting. +func (ri *runInputs) overriddenKeys() []string { + var ids []string + for _, o := range ri.overrides { + ids = append(ids, o.ID().String()) + } + sort.Strings(ids) + return ids +} + +// skippedTaskNames returns the names of the tasks which are not run, sorted, for reporting. +func (ri *runInputs) skippedTaskNames() []string { + var names []string + for node := range ri.skipped { + names = append(names, node.task.Name()) + } + sort.Strings(names) + return names +} + +// buildInputs checks the given inputs and overrides against the graph, working out which tasks, if +// any, the overrides replace. If ignoreUnprovided is set, overrides for keys which no task in the +// graph provides are left out, instead of being an error. +func (g *graph) buildInputs( + inputs, overrides []Binding, + ignoreUnprovided bool, +) (*runInputs, error) { + b := NewBinder() + if err := b.Store(inputs...); err != nil { + return nil, wrapStackErrorf("duplicate input: %w", err) + } + + result := &runInputs{inputs: b, overrides: overrides, skipped: map[*graphNode]struct{}{}} + if ignoreUnprovided { + result.overrides = nil + for _, override := range overrides { + if g.allProvided.Contains(override.ID()) { + result.overrides = append(result.overrides, override) + } + } + } + if len(result.overrides) > 0 { + if err := g.checkOverrides(b, result); err != nil { + return nil, err + } + } + + // The dependencies which have to be given as inputs are those of the tasks which will run, which + // are fewer if some are replaced. + required := g.allDependencies.Difference(g.allProvided) + if len(result.skipped) > 0 { + required = set.NewSet[ID]() + for _, node := range g.nodes { + if _, skipped := result.skipped[node]; !skipped { + required.Append(node.task.Depends()...) + } + } + required = required.Difference(g.allProvided) + } + var missingInputs []string + for requiredInput := range required.Iter() { + if !b.Has(requiredInput) { + missingInputs = append(missingInputs, requiredInput.String()) + } + } + if len(missingInputs) > 0 { + sort.Strings(missingInputs) + return nil, wrapStackErrorf("%w: %s", ErrMissingInputs, strings.Join(missingInputs, ", ")) + } + + return result, nil +} + +// checkOverrides checks the overrides, filling in which nodes they replace. +func (g *graph) checkOverrides(ordinary Binder, result *runInputs) error { + overridden := set.NewSet[ID]() + for _, o := range result.overrides { + if ordinary.Has(o.ID()) || overridden.Contains(o.ID()) { + return wrapStackErrorf("duplicate input: %w: %q", ErrDuplicateBinding, o.ID()) + } + overridden.Add(o.ID()) + } + + var unknown []string + for id := range overridden.Difference(g.allProvided).Iter() { + unknown = append(unknown, id.String()) + } + if len(unknown) > 0 { + sort.Strings(unknown) + return wrapStackErrorf("%w: %s", ErrUnknownOverride, strings.Join(unknown, ", ")) + } + + var partial []string + for _, node := range g.nodes { + provides := node.task.Provides() + var covered, uncovered []string + for _, id := range provides { + if overridden.Contains(id) { + covered = append(covered, id.String()) + } else { + uncovered = append(uncovered, id.String()) + } + } + switch { + case len(covered) == 0: + case len(uncovered) == 0: + result.skipped[node] = struct{}{} + default: + partial = append(partial, fmt.Sprintf( + "task %s provides [%s, %s], of which only [%s] are overridden", + node.task.Name(), + strings.Join(covered, ", "), + strings.Join(uncovered, ", "), + strings.Join(covered, ", "), + )) + } + } + if len(partial) > 0 { + sort.Strings(partial) + return wrapStackErrorf("%w: %s", ErrPartialOverride, strings.Join(partial, "; ")) + } + return nil +} diff --git a/override_test.go b/override_test.go new file mode 100644 index 0000000..b56dbad --- /dev/null +++ b/override_test.go @@ -0,0 +1,465 @@ +package taskgraph_test + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + tg "github.com/thought-machine/taskgraph" + tgt "github.com/thought-machine/taskgraph/taskgraphtest" +) + +// overrideFixture is a small graph, in -> A -> B, whose tasks count how often they run: +// +// A: a = in * 2 +// B: b = a + 1 +type overrideFixture struct { + keyIn, keyA, keyB tg.Key[int] + runsA, runsB atomic.Int32 + graph tg.Graph +} + +func newOverrideFixture(t *testing.T) *overrideFixture { + t.Helper() + f := &overrideFixture{ + keyIn: tg.NewKey[int]("in"), + keyA: tg.NewKey[int]("a"), + keyB: tg.NewKey[int]("b"), + } + f.graph = tgt.Must[tg.Graph](t)(tg.New("override_graph", tg.WithTasks( + tg.SimpleTask1[int, int]("A", f.keyA, func(_ context.Context, in int) (int, error) { + f.runsA.Add(1) + return in * 2, nil + }, f.keyIn), + tg.SimpleTask1[int, int]("B", f.keyB, func(_ context.Context, a int) (int, error) { + f.runsB.Add(1) + return a + 1, nil + }, f.keyA), + ))) + return f +} + +func (f *overrideFixture) run(t *testing.T, inputs ...tg.Binding) tg.Binder { + t.Helper() + result, err := runWith(testContext(t), f.graph, inputs...) + if err != nil { + t.Fatalf("Run() failed: %v", err) + } + return result +} + +// testContext returns a context which ends soon, so that a graph which wrongly waits for something +// which will never happen fails the test, rather than hanging it. +func testContext(t *testing.T) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + t.Cleanup(cancel) + return ctx +} + +// override marks an input as an override, for runWith and checkWith to separate from the inputs. +type override struct{ tg.Binding } + +func ov(b tg.Binding) tg.Binding { return override{b} } + +func splitOverrides(inputs []tg.Binding) (ordinary, overrides []tg.Binding) { + for _, input := range inputs { + if o, ok := input.(override); ok { + overrides = append(overrides, o.Binding) + } else { + ordinary = append(ordinary, input) + } + } + return ordinary, overrides +} + +func runWith(ctx context.Context, g tg.Graph, inputs ...tg.Binding) (tg.Binder, error) { + ordinary, overrides := splitOverrides(inputs) + return tgt.RunWithOverrides(ctx, g, ordinary, overrides) +} + +func checkWith(g tg.Graph, inputs ...tg.Binding) error { + ordinary, overrides := splitOverrides(inputs) + return tgt.CheckWithOverrides(g, ordinary, overrides) +} + +func mustGet[T any](t *testing.T, key tg.ReadOnlyKey[T], b tg.Binder) T { + t.Helper() + got, err := key.Get(b) + if err != nil { + t.Fatalf("getting %s: %v", key.ID(), err) + } + return got +} + +func TestOverride_WithoutOverrideTasksRun(t *testing.T) { + f := newOverrideFixture(t) + result := f.run(t, f.keyIn.Bind(1)) + + if got, want := mustGet(t, f.keyB, result), 3; got != want { + t.Errorf("b = %d; want %d", got, want) + } + if f.runsA.Load() != 1 || f.runsB.Load() != 1 { + t.Errorf("tasks ran A %d times, B %d times; want once each", f.runsA.Load(), f.runsB.Load()) + } +} + +func TestOverride_ReplacesTheTaskWhichProvidesTheKey(t *testing.T) { + f := newOverrideFixture(t) + result := f.run(t, f.keyIn.Bind(1), ov(f.keyA.Bind(100))) + + if f.runsA.Load() != 0 { + t.Errorf("A ran %d times; want it replaced", f.runsA.Load()) + } + if got, want := mustGet(t, f.keyB, result), 101; got != want { + t.Errorf("b = %d; want %d, from the override", got, want) + } + // The override takes the place of the result the task would have produced. + if got, want := mustGet(t, f.keyA, result), 100; got != want { + t.Errorf("a in the result = %d; want the override, %d", got, want) + } +} + +func TestOverride_GraphCanBeRunAgainWithDifferentOverrides(t *testing.T) { + f := newOverrideFixture(t) + + if got, want := mustGet( + t, + f.keyB, + f.run(t, f.keyIn.Bind(1), ov(f.keyA.Bind(10))), + ), 11; got != want { + t.Errorf("first run: b = %d; want %d", got, want) + } + if got, want := mustGet(t, f.keyB, f.run(t, f.keyIn.Bind(1))), 3; got != want { + t.Errorf("second run, without an override: b = %d; want %d", got, want) + } + if got, want := mustGet( + t, + f.keyB, + f.run(t, f.keyIn.Bind(1), ov(f.keyA.Bind(20))), + ), 21; got != want { + t.Errorf("third run: b = %d; want %d", got, want) + } + if f.runsA.Load() != 1 { + t.Errorf( + "A ran %d times over the three runs; want once, in the one without an override", + f.runsA.Load(), + ) + } +} + +func TestOverride_AbsentOverrideFailsTheTask(t *testing.T) { + errBoom := errors.New("boom") + keyIn, keyA, keyB := tg.NewKey[int]("in"), tg.NewKey[int]("a"), tg.NewKey[int]("b") + var runsA atomic.Int32 + g := tgt.Must[tg.Graph](t)(tg.New("absent_override", tg.WithTasks( + tg.SimpleTask1[int, int]("A", keyA, func(_ context.Context, in int) (int, error) { + runsA.Add(1) + return in, nil + }, keyIn), + // Takes the absence of a in its stride, as a task which can cope with a missing dependency does. + tg.NewTask("B", func(_ context.Context, b tg.Binder) ([]tg.Binding, error) { + if _, err := keyA.Get(b); err != nil { + return []tg.Binding{keyB.Bind(-1)}, nil + } + return []tg.Binding{keyB.Bind(1)}, nil + }, []tg.ID{keyA.ID()}, []tg.ID{keyB.ID()}), + ))) + + result, err := runWith(testContext(t), g, keyIn.Bind(1), ov(keyA.BindError(errBoom))) + if err != nil { + t.Fatalf("Run() failed: %v", err) + } + if runsA.Load() != 0 { + t.Errorf("A ran %d times; want it replaced by the failure", runsA.Load()) + } + if got := mustGet(t, keyB, result); got != -1 { + t.Errorf("b = %d; want -1, as B saw a absent", got) + } + tgt.ExpectAbsentError(t, result, keyA, errBoom) +} + +func TestOverride_ConditionalTaskIsReplacedWhateverItsCondition(t *testing.T) { + for _, flag := range []bool{true, false} { + keyFlag, keyT, keyOut := tg.NewKey[bool]("flag"), tg.NewKey[int]("t"), tg.NewKey[int]("out") + var runsT atomic.Int32 + g := tgt.Must[tg.Graph](t)(tg.New("conditional_override", tg.WithTasks( + tg.Conditional{ + Wrapped: tg.SimpleTask("T", keyT, func(context.Context, tg.Binder) (int, error) { + runsT.Add(1) + return 1, nil + }), + Condition: tg.ConditionAnd{keyFlag}, + }.Locate(), + tg.SimpleTask1[int, int]("Out", keyOut, func(_ context.Context, v int) (int, error) { + return v * 10, nil + }, keyT), + ))) + + result, err := runWith(testContext(t), g, keyFlag.Bind(flag), ov(keyT.Bind(7))) + if err != nil { + t.Fatalf("flag=%v: Run() failed: %v", flag, err) + } + if runsT.Load() != 0 { + t.Errorf("flag=%v: T ran %d times; want it replaced", flag, runsT.Load()) + } + if got, want := mustGet(t, keyOut, result), 70; got != want { + t.Errorf("flag=%v: out = %d; want %d", flag, got, want) + } + } +} + +func TestOverride_TasksWhichFedAReplacedTaskStillRunAndDoNotBlock(t *testing.T) { + // This documents a limitation: the graph can't know which results its caller wants, so it can't + // tell that nothing needs what only the replaced task consumed. + f := newOverrideFixture(t) + keyU, runsU := tg.NewKey[int]("u"), atomic.Int32{} + g := tgt.Must[tg.Graph](t)(tg.New("upstream_still_runs", tg.WithTasks( + tg.SimpleTask1[int, int]("U", keyU, func(_ context.Context, in int) (int, error) { + runsU.Add(1) + return in, nil + }, f.keyIn), + tg.SimpleTask1[int, int]("A", f.keyA, func(_ context.Context, u int) (int, error) { + return u * 2, nil + }, keyU), + ))) + + if _, err := runWith(testContext(t), g, f.keyIn.Bind(1), ov(f.keyA.Bind(5))); err != nil { + t.Fatalf("Run() failed: %v", err) + } + if runsU.Load() != 1 { + t.Errorf("U ran %d times; want once", runsU.Load()) + } +} + +func TestOverride_InputsOnlyAReplacedTaskNeededAreNotRequired(t *testing.T) { + keyOnlyForA, keyA, keyOut := tg.NewKey[int]( + "only_for_a", + ), tg.NewKey[int]( + "a", + ), tg.NewKey[int]( + "out", + ) + g := tgt.Must[tg.Graph](t)(tg.New("needed_inputs", tg.WithTasks( + tg.SimpleTask1[int, int]( + "A", + keyA, + func(_ context.Context, v int) (int, error) { return v, nil }, + keyOnlyForA, + ), + tg.SimpleTask1[int, int]( + "Out", + keyOut, + func(_ context.Context, a int) (int, error) { return a + 1, nil }, + keyA, + ), + ))) + + if _, err := runWith(testContext(t), g); !errors.Is(err, tg.ErrMissingInputs) { + t.Errorf("without the override: got %v; want ErrMissingInputs", err) + } + result, err := runWith(testContext(t), g, ov(keyA.Bind(4))) + if err != nil { + t.Fatalf("with the override: Run() failed: %v", err) + } + if got, want := mustGet(t, keyOut, result), 5; got != want { + t.Errorf("out = %d; want %d", got, want) + } +} + +func TestOverride_TasksDependingOnReplacedAndOrdinaryKeysDoNotWaitForeverOrDeadlock(t *testing.T) { + keyIn, keyA, keyP, keyBoth, keyOnlyA := tg.NewKey[int]( + "in", + ), tg.NewKey[int]( + "a", + ), tg.NewKey[int]( + "p", + ), + tg.NewKey[int]( + "both", + ), tg.NewKey[int]( + "only_a", + ) + g := tgt.Must[tg.Graph](t)(tg.New("mixed_dependents", tg.WithTasks( + tg.SimpleTask1[int, int]( + "A", + keyA, + func(_ context.Context, v int) (int, error) { return v, nil }, + keyIn, + ), + tg.SimpleTask1[int, int]("P", keyP, func(_ context.Context, v int) (int, error) { + time.Sleep(20 * time.Millisecond) + return v * 100, nil + }, keyIn), + tg.SimpleTask2[int, int, int]( + "Both", + keyBoth, + func(_ context.Context, a, p int) (int, error) { + return a + p, nil + }, + keyA, + keyP, + ), + tg.SimpleTask1[int, int]( + "OnlyA", + keyOnlyA, + func(_ context.Context, a int) (int, error) { return a, nil }, + keyA, + ), + ))) + + result, err := runWith(testContext(t), g, keyIn.Bind(1), ov(keyA.Bind(7))) + if err != nil { + t.Fatalf("Run() failed: %v", err) + } + if got, want := mustGet(t, keyBoth, result), 107; got != want { + t.Errorf("both = %d; want %d", got, want) + } + if got, want := mustGet(t, keyOnlyA, result), 7; got != want { + t.Errorf("only_a = %d; want %d", got, want) + } +} + +func TestOverride_Errors(t *testing.T) { + keyIn, keyA, keyM1, keyM2, keyZ := tg.NewKey[int]( + "in", + ), tg.NewKey[int]( + "a", + ), tg.NewKey[int]( + "m1", + ), + tg.NewKey[int]( + "m2", + ), tg.NewKey[int]( + "z", + ) + var runsM atomic.Int32 + g := tgt.Must[tg.Graph](t)(tg.New("override_errors", tg.WithTasks( + tg.SimpleTask1[int, int]( + "A", + keyA, + func(_ context.Context, v int) (int, error) { return v, nil }, + keyIn, + ), + tg.NewTask("M", func(_ context.Context, _ tg.Binder) ([]tg.Binding, error) { + runsM.Add(1) + return []tg.Binding{keyM1.Bind(1), keyM2.Bind(2)}, nil + }, nil, []tg.ID{keyM1.ID(), keyM2.ID()}), + ))) + + for _, tc := range []struct { + description string + inputs []tg.Binding + wantError error + // Whether the problem is found before anything runs, which is true of everything but a plain + // input for a key which a task provides, which is only found when that task stores its result. + foundBeforeRunning bool + }{ + { + description: "a key which no task provides", + inputs: []tg.Binding{keyIn.Bind(1), ov(keyZ.Bind(1))}, + wantError: tg.ErrUnknownOverride, + foundBeforeRunning: true, + }, + { + description: "an input, not an override, for a key which a task provides", + inputs: []tg.Binding{keyIn.Bind(1), keyA.Bind(5)}, + wantError: tg.ErrDuplicateBinding, + }, + { + description: "the same key overridden twice", + inputs: []tg.Binding{keyIn.Bind(1), ov(keyA.Bind(1)), ov(keyA.Bind(2))}, + wantError: tg.ErrDuplicateBinding, + foundBeforeRunning: true, + }, + { + description: "a key which is also an ordinary input", + inputs: []tg.Binding{keyIn.Bind(1), keyA.Bind(1), ov(keyA.Bind(2))}, + wantError: tg.ErrDuplicateBinding, + foundBeforeRunning: true, + }, + { + description: "only some of the keys which a task provides", + inputs: []tg.Binding{keyIn.Bind(1), ov(keyM1.Bind(1))}, + wantError: tg.ErrPartialOverride, + foundBeforeRunning: true, + }, + } { + t.Run(tc.description, func(t *testing.T) { + runsM.Store(0) + if _, err := runWith(testContext(t), g, tc.inputs...); !errors.Is(err, tc.wantError) { + t.Errorf("Run(): got %v; want %v", err, tc.wantError) + } + if tc.foundBeforeRunning { + if runsM.Load() != 0 { + t.Errorf( + "M ran %d times; want nothing to run, as this is rejected first", + runsM.Load(), + ) + } + if err := checkWith(g, tc.inputs...); !errors.Is(err, tc.wantError) { + t.Errorf("Check(): got %v; want %v", err, tc.wantError) + } + } + }) + } + + t.Run("all of the keys which a task provides", func(t *testing.T) { + runsM.Store(0) + inputs := []tg.Binding{keyIn.Bind(1), ov(keyM1.Bind(10)), ov(keyM2.Bind(20))} + if err := checkWith(g, inputs...); err != nil { + t.Fatalf("Check(): %v", err) + } + result, err := runWith(testContext(t), g, inputs...) + if err != nil { + t.Fatalf("Run(): %v", err) + } + if runsM.Load() != 0 { + t.Errorf("M ran %d times; want it replaced", runsM.Load()) + } + if got := mustGet(t, keyM2, result); got != 20 { + t.Errorf("m2 = %d; want 20", got) + } + }) +} + +func TestOverride_GraphRunAsATaskIsReplacedAsAWhole(t *testing.T) { + keyIn, keyX, keyY := tg.NewKey[int]("in"), tg.NewKey[int]("x"), tg.NewKey[int]("y") + var runsInner atomic.Int32 + inner := tgt.Must[tg.Graph](t)(tg.New("inner", tg.WithTasks( + tg.NewTask("XY", func(_ context.Context, b tg.Binder) ([]tg.Binding, error) { + runsInner.Add(1) + in := mustGet(t, keyIn, b) + return []tg.Binding{keyX.Bind(in), keyY.Bind(in * 2)}, nil + }, []tg.ID{keyIn.ID()}, []tg.ID{keyX.ID(), keyY.ID()}), + ))) + asTask := tgt.Must[tg.Task](t)(inner.AsTask(keyX.ID(), keyY.ID())) + outer := tgt.Must[tg.Graph](t)(tg.New("outer", tg.WithTasks(asTask))) + + if _, err := runWith( + testContext(t), + outer, + keyIn.Bind(1), + ov(keyX.Bind(9)), + ); !errors.Is( + err, + tg.ErrPartialOverride, + ) { + t.Errorf( + "overriding one of the keys the task exposes: got %v; want ErrPartialOverride", + err, + ) + } + result, err := runWith(testContext(t), outer, keyIn.Bind(1), ov(keyX.Bind(9)), ov(keyY.Bind(8))) + if err != nil { + t.Fatalf("overriding both: Run() failed: %v", err) + } + if runsInner.Load() != 0 { + t.Errorf("the inner graph ran %d times; want it replaced", runsInner.Load()) + } + if got := mustGet(t, keyY, result); got != 8 { + t.Errorf("y = %d; want 8", got) + } +} diff --git a/taskgraphtest/overrides.go b/taskgraphtest/overrides.go new file mode 100644 index 0000000..c12632f --- /dev/null +++ b/taskgraphtest/overrides.go @@ -0,0 +1,120 @@ +package taskgraphtest + +import ( + "context" + "fmt" + "sort" + "strings" + + tg "github.com/thought-machine/taskgraph" + "github.com/thought-machine/taskgraph/internal/overridehook" +) + +func toAny(bindings []tg.Binding) []any { + out := make([]any, 0, len(bindings)) + for _, b := range bindings { + out = append(out, b) + } + return out +} + +// RunWithOverrides runs the graph, as Graph.Run does with the inputs, with the tasks which provide +// the keys of the overrides replaced by them. It is the way to substitute what the tasks of a graph +// would otherwise create for themselves, such as clients of other services, in a test: +// +// taskgraphtest.RunWithOverrides(ctx, graph, inputs, []tg.Binding{keyClient.Bind(fakeClient)}) +// +// The task providing the key is not run. If the task provides other keys as well, they must all be +// overridden too (tg.ErrPartialOverride), so that the real task's effects can never happen +// unnoticed. Tasks which have no other purpose than to produce the inputs of a task which is not run +// are still run, as are tasks which depend on the overridden key; they see the override as if the +// task which it replaces had produced it. The overrides are included in the Binder returned, as the +// replaced tasks' results would have been, and inputs which only a replaced task needed are not +// required. +// +// An override may be absent, which replaces the task with a failure: +// +// keyClient.BindError(errUnreachable) +// +// Overriding a key which no task provides is an error (tg.ErrUnknownOverride), as is overriding the +// same key twice, or overriding a key which is also given as an input. An input for a key which a task +// provides is still an error (tg.ErrDuplicateBinding), so a key can only ever be replaced on purpose. +// +// Overrides apply to the graph's own tasks. A graph run as a task with Graph.AsTask is replaced as a +// whole, by overriding every key it exposes, but its tasks can't be overridden individually. +func RunWithOverrides( + ctx context.Context, + g tg.Graph, + inputs, overrides []tg.Binding, +) (tg.Binder, error) { + return runWithOverrides(ctx, g, inputs, overrides, false) +} + +func runWithOverrides( + ctx context.Context, + g tg.Graph, + inputs, overrides []tg.Binding, + ignoreUnprovided bool, +) (tg.Binder, error) { + result, err := overridehook.RunWithOverrides( + g, + ctx, + toAny(inputs), + toAny(overrides), + ignoreUnprovided, + ) + if err != nil { + return nil, err + } + return result.(tg.Binder), nil +} + +// CheckWithOverrides is Graph.Check, with the overrides RunWithOverrides would be given: it finds +// unknown, duplicated and partial overrides, and which inputs are still needed, without running +// anything. +func CheckWithOverrides(g tg.Graph, inputs, overrides []tg.Binding) error { + return overridehook.CheckWithOverrides(g, toAny(inputs), toAny(overrides)) +} + +// An Overrider runs graphs with one set of overrides. It is for something which runs several graphs, +// such as an operator with a graph to create and one to delete, and which can only be given one set: +// each graph is run with those of the overrides which are for keys it provides, and Check makes sure +// that every override is for some graph. +type Overrider struct { + overrides []tg.Binding +} + +// NewOverrider returns an Overrider with the overrides. +func NewOverrider(overrides ...tg.Binding) *Overrider { + return &Overrider{overrides: overrides} +} + +// Run runs the graph with the inputs, and those of the Overrider's overrides which are for keys the +// graph provides. Its signature is that of a function for running a graph, so it can be handed to +// code which runs graphs. +func (o *Overrider) Run(ctx context.Context, g tg.Graph, inputs ...tg.Binding) (tg.Binder, error) { + return runWithOverrides(ctx, g, inputs, o.overrides, true) +} + +// Check returns an error if any of the overrides is for a key which none of the graphs provide, which +// is nearly always one left behind by a task which has been renamed or removed. +func (o *Overrider) Check(graphs ...tg.Graph) error { + var unknown []string + for _, override := range o.overrides { + provided := false + for _, g := range graphs { + if overridehook.Provides(g, override.ID()) { + provided = true + break + } + } + if !provided { + unknown = append(unknown, override.ID().String()) + } + } + if len(unknown) > 0 { + sort.Strings(unknown) + return fmt.Errorf("%w: %s", tg.ErrUnknownOverride, strings.Join(unknown, ", ")) + } + return nil +} diff --git a/taskgraphtest/taskgraphtest.go b/taskgraphtest/taskgraphtest.go index b783131..ebf1791 100644 --- a/taskgraphtest/taskgraphtest.go +++ b/taskgraphtest/taskgraphtest.go @@ -263,6 +263,14 @@ type Test struct { // Inputs defines the inputs passed to the graph for the test. Inputs []tg.Binding + // Overrides defines bindings which replace the tasks providing their keys, as with + // RunWithOverrides; the replaced tasks are not run. Overrides given by the Suite + // are applied as well, those given here replacing them for the same key. + // + // The overrides are part of the result, as the replaced tasks' results would have been, so + // CheckExcessBindings counts them among the bindings to be wanted. + Overrides []tg.Binding + // WantError defines the expected error returned from the Graph's execution (note that the test // will always fail if Graph setup/configuration fails). Compared using errors.Is. WantError error @@ -314,7 +322,7 @@ func (test Test) Run(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), test.Timeout) defer cancel() - result, err := test.Graph.Run(ctx, test.Inputs...) + result, err := RunWithOverrides(ctx, test.Graph, test.Inputs, test.Overrides) if !errors.Is(err, test.WantError) { t.Fatalf("Difference in error from Graph.Run(): got %v; want %v", err, test.WantError) @@ -354,6 +362,11 @@ type Suite struct { // the set is expected to contain a single element. Task tg.TaskSet + // Overrides defines bindings which replace the tasks providing their keys in every test of the + // suite (see Test.Overrides). A test which overrides the same key itself uses its own binding in + // place of the suite's. + Overrides []tg.Binding + // Tests defines the individual tests to be run. Tests []Test @@ -382,10 +395,30 @@ func (s Suite) Run(t *testing.T) { if test.Timeout <= 0 { test.Timeout = s.Timeout } + test.Overrides = mergeOverrides(s.Overrides, test.Overrides) t.Run(test.Description, test.Run) } } +// mergeOverrides returns the overrides of a suite and of one of its tests together, those of the test +// replacing those of the suite which are for the same key. +func mergeOverrides(suite, test []tg.Binding) []tg.Binding { + if len(suite) == 0 { + return test + } + overridden := set.NewSet[tg.ID]() + for _, b := range test { + overridden.Add(b.ID()) + } + merged := append([]tg.Binding{}, test...) + for _, b := range suite { + if !overridden.Contains(b.ID()) { + merged = append(merged, b) + } + } + return merged +} + // DummyTaskFunc returns a function suitable for passing to taskgraph.NewTask which returns the // given bindings. func DummyTaskFunc(bindings ...tg.Binding) func(context.Context, tg.Binder) ([]tg.Binding, error) { diff --git a/taskgraphtest/taskgraphtest_test.go b/taskgraphtest/taskgraphtest_test.go new file mode 100644 index 0000000..379c52e --- /dev/null +++ b/taskgraphtest/taskgraphtest_test.go @@ -0,0 +1,148 @@ +package taskgraphtest_test + +import ( + "context" + "errors" + "sync/atomic" + "testing" + + tg "github.com/thought-machine/taskgraph" + tgt "github.com/thought-machine/taskgraph/taskgraphtest" +) + +func TestOverrides(t *testing.T) { + keyIn, keyA, keyB := tg.NewKey[int]("in"), tg.NewKey[int]("a"), tg.NewKey[int]("b") + var runsA atomic.Int32 + graph := tgt.Must[tg.Graph](t)(tg.New("overrides", tg.WithTasks( + tg.SimpleTask1[int, int]("A", keyA, func(_ context.Context, in int) (int, error) { + runsA.Add(1) + return in * 2, nil + }, keyIn), + tg.SimpleTask1[int, int]("B", keyB, func(_ context.Context, a int) (int, error) { + return a + 1, nil + }, keyA), + ))) + + tgt.Suite{ + Graph: graph, + Overrides: []tg.Binding{keyA.Bind(10)}, + Tests: []tgt.Test{ + { + Description: "a suite's overrides apply to its tests", + Inputs: []tg.Binding{keyIn.Bind(1)}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyB.Bind(11))}, + }, + { + Description: "a test's override replaces the suite's for the same key", + Inputs: []tg.Binding{keyIn.Bind(1)}, + Overrides: []tg.Binding{keyA.Bind(20)}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyB.Bind(21))}, + }, + { + Description: "overrides are part of the result", + Inputs: []tg.Binding{keyIn.Bind(1)}, + WantBindings: []tgt.BindingMatcher{ + tgt.Match(keyA.Bind(10)), + tgt.Match(keyB.Bind(11)), + }, + }, + { + Description: "excess bindings includes them", + Inputs: []tg.Binding{keyIn.Bind(1)}, + WantBindings: []tgt.BindingMatcher{ + tgt.Match(keyA.Bind(10)), + tgt.Match(keyB.Bind(11)), + }, + CheckExcessBindings: true, + }, + { + Description: "an absent override makes the task fail", + Inputs: []tg.Binding{keyIn.Bind(1)}, + Overrides: []tg.Binding{keyA.BindAbsent()}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyA.BindAbsent())}, + WantError: tg.ErrIsAbsent, + }, + }, + }.Run(t) + + if runsA.Load() != 0 { + t.Errorf("A ran %d times; want it replaced in every test", runsA.Load()) + } + + // A test which isn't part of a suite can override too. + tgt.Test{ + Graph: graph, + Inputs: []tg.Binding{keyIn.Bind(1)}, + Overrides: []tg.Binding{keyA.Bind(5)}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyB.Bind(6))}, + }.Run(t) +} + +// An operator which has a graph to create and one to delete is given one set of overrides. +func TestOverrider(t *testing.T) { + keyIn, keyA, keyB := tg.NewKey[int]("in"), tg.NewKey[int]("a"), tg.NewKey[int]("b") + keyD, keyUnused := tg.NewKey[int]("d"), tg.NewKey[int]("unused") + create := tgt.Must[tg.Graph](t)(tg.New("create", tg.WithTasks( + tg.SimpleTask1[int, int]( + "A", + keyA, + func(_ context.Context, in int) (int, error) { return in * 2, nil }, + keyIn, + ), + tg.SimpleTask1[int, int]( + "B", + keyB, + func(_ context.Context, a int) (int, error) { return a + 1, nil }, + keyA, + ), + ))) + del := tgt.Must[tg.Graph](t)(tg.New("delete", tg.WithTasks( + tg.SimpleTask1[int, int]( + "D", + keyD, + func(_ context.Context, in int) (int, error) { return in * 3, nil }, + keyIn, + ), + ))) + + overrider := tgt.NewOverrider(keyA.Bind(10), keyD.Bind(20)) + if err := overrider.Check(create, del); err != nil { + t.Fatalf("Check(): %v", err) + } + + // Each graph takes the overrides which are for it, and ignores the others. + result, err := overrider.Run(context.Background(), create, keyIn.Bind(1)) + if err != nil { + t.Fatalf("running the graph to create: %v", err) + } + if got, err := keyB.Get(result); err != nil || got != 11 { + t.Errorf("b = %d, %v; want 11, from the override of a", got, err) + } + result, err = overrider.Run(context.Background(), del, keyIn.Bind(1)) + if err != nil { + t.Fatalf("running the graph to delete: %v", err) + } + if got, err := keyD.Get(result); err != nil || got != 20 { + t.Errorf("d = %d, %v; want 20, from the override", got, err) + } + + // An override for neither graph is found by Check, though each Run leaves it out. + stale := tgt.NewOverrider(keyA.Bind(10), keyUnused.Bind(1)) + if err := stale.Check(create, del); !errors.Is(err, tg.ErrUnknownOverride) { + t.Errorf("Check() with a stale override: got %v; want ErrUnknownOverride", err) + } + if _, err := stale.Run(context.Background(), create, keyIn.Bind(1)); err != nil { + t.Errorf("Run() with an override for another graph: %v", err) + } +} + +// fakeGraph is a Graph which taskgraph.New didn't make. +type fakeGraph struct{ tg.Graph } + +func TestOverridesNeedAGraphMadeByTaskgraph(t *testing.T) { + if _, err := tgt.RunWithOverrides(context.Background(), fakeGraph{}, nil, nil); err == nil { + t.Error( + "RunWithOverrides() on a Graph which taskgraph.New didn't make: got nil; want an error", + ) + } +}