lang: interfaces, funcs: Add a new graph engine called dage

This adds a new implementation of the function engine that runs the DAG
function graph. This version is notable in that it can run a graph that
changes shape over time. To make changes to the same of the graph, you
must use the new transaction (Txn) system. This system implements a
simple garbage collector (GC) for scheduled removal of nodes that the
transaction system "reverses" out of the graph.

Special thanks to Samuel Gélineau <gelisam@gmail.com> for his help
hacking on and debugging so much of this concurrency work with me.

lang/funcs/dage/dage_test.go

@@ -0,0 +1,793 @@
+// Mgmt
+// Copyright (C) 2013-2023+ James Shubin and the project contributors
+// Written by James Shubin <james@shubin.ca> and the project contributors
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package dage
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/purpleidea/mgmt/lang/interfaces"
+	"github.com/purpleidea/mgmt/lang/types"
+	"github.com/purpleidea/mgmt/util"
+)
+
+type testFunc struct {
+	Name string
+	Type *types.Type
+	Func func(types.Value) (types.Value, error)
+	Meta *meta
+
+	value types.Value
+	init  *interfaces.Init
+}
+
+func (obj *testFunc) String() string { return obj.Name }
+
+func (obj *testFunc) Info() *interfaces.Info {
+	return &interfaces.Info{
+		Pure: true,
+		Memo: false, // TODO: should this be something we specify here?
+		Sig:  obj.Type,
+		Err:  obj.Validate(),
+	}
+}
+
+func (obj *testFunc) Validate() error {
+	if obj.Meta == nil {
+		return fmt.Errorf("test case error: did you add the vertex to the vertices list?")
+	}
+	return nil
+}
+
+func (obj *testFunc) Init(init *interfaces.Init) error {
+	obj.init = init
+	return nil
+}
+
+func (obj *testFunc) Stream(ctx context.Context) error {
+	defer close(obj.init.Output) // the sender closes
+	defer obj.init.Logf("stream closed")
+	obj.init.Logf("stream startup")
+
+	// make some placeholder value because obj.value is nil
+	constValue, err := types.ValueOfGolang("hello")
+	if err != nil {
+		return err // unlikely
+	}
+
+	for {
+		select {
+		case input, ok := <-obj.init.Input:
+			if !ok {
+				obj.init.Logf("stream input closed")
+				obj.init.Input = nil // don't get two closes
+				// already sent one value, so we can shutdown
+				if obj.value != nil {
+					return nil // can't output any more
+				}
+
+				obj.value = constValue
+			} else {
+				obj.init.Logf("stream got input type(%T) value: (%+v)", input, input)
+				if obj.Func == nil {
+					obj.value = constValue
+				}
+
+				if obj.Func != nil {
+					//obj.init.Logf("running internal function...")
+					v, err := obj.Func(input) // run me!
+					if err != nil {
+						return err
+					}
+					obj.value = v
+				}
+			}
+
+		case <-ctx.Done():
+			return nil
+		}
+
+		select {
+		case obj.init.Output <- obj.value: // send anything
+			// add some monitoring...
+			obj.Meta.wg.Add(1)
+			go func() {
+				// no need for lock here
+				defer obj.Meta.wg.Done()
+				if obj.Meta.debug {
+					obj.Meta.logf("sending an internal event!")
+				}
+
+				select {
+				case obj.Meta.Events[obj.Name] <- struct{}{}:
+				case <-obj.Meta.ctx.Done():
+				}
+			}()
+
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+type meta struct {
+	EventCount int
+	Event      chan struct{}
+	Events     map[string]chan struct{}
+
+	ctx   context.Context
+	wg    *sync.WaitGroup
+	mutex *sync.Mutex
+
+	debug bool
+	logf  func(format string, v ...interface{})
+}
+
+func (obj *meta) Lock()   { obj.mutex.Lock() }
+func (obj *meta) Unlock() { obj.mutex.Unlock() }
+
+type dageTestOp func(*Engine, interfaces.Txn, *meta) error
+
+func TestDageTable(t *testing.T) {
+
+	type test struct { // an individual test
+		name     string
+		vertices []interfaces.Func
+		actions  []dageTestOp
+	}
+	testCases := []test{}
+	{
+		testCases = append(testCases, test{
+			name:     "empty graph",
+			vertices: []interfaces.Func{},
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					defer engine.Unlock()
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					meta.Lock()
+					defer meta.Unlock()
+					// We don't expect an empty graph to send events.
+					if meta.EventCount != 0 {
+						return fmt.Errorf("got too many stream events")
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+
+		testCases = append(testCases, test{
+			name:     "simple add vertex",
+			vertices: []interfaces.Func{f1}, // so the test engine can pass in debug/observability handles
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					defer engine.Unlock()
+					return engine.AddVertex(f1)
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					meta.Lock()
+					defer meta.Unlock()
+					if meta.EventCount < 1 {
+						return fmt.Errorf("didn't get any stream events")
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+		// e1 arg name must match incoming edge to it
+		f2 := &testFunc{Name: "f2", Type: types.NewType("func(e1 str) str")}
+		e1 := testEdge("e1")
+
+		testCases = append(testCases, test{
+			name:     "simple add edge",
+			vertices: []interfaces.Func{f1, f2},
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					defer engine.Unlock()
+					return engine.AddVertex(f1)
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					engine.Lock()
+					defer engine.Unlock()
+					// This newly added node should get a notification after it starts.
+					return engine.AddEdge(f1, f2, e1)
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					meta.Lock()
+					defer meta.Unlock()
+					if meta.EventCount < 2 {
+						return fmt.Errorf("didn't get enough stream events")
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		// diamond
+		f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+		f2 := &testFunc{Name: "f2", Type: types.NewType("func(e1 str) str")}
+		f3 := &testFunc{Name: "f3", Type: types.NewType("func(e2 str) str")}
+		f4 := &testFunc{Name: "f4", Type: types.NewType("func(e3 str, e4 str) str")}
+		e1 := testEdge("e1")
+		e2 := testEdge("e2")
+		e3 := testEdge("e3")
+		e4 := testEdge("e4")
+
+		testCases = append(testCases, test{
+			name:     "simple add multiple edges",
+			vertices: []interfaces.Func{f1, f2, f3, f4},
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					defer engine.Unlock()
+					return engine.AddVertex(f1)
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					defer engine.Unlock()
+					if err := engine.AddEdge(f1, f2, e1); err != nil {
+						return err
+					}
+					if err := engine.AddEdge(f1, f3, e2); err != nil {
+						return err
+					}
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					defer engine.Unlock()
+					if err := engine.AddEdge(f2, f4, e3); err != nil {
+						return err
+					}
+					if err := engine.AddEdge(f3, f4, e4); err != nil {
+						return err
+					}
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					//meta.Lock()
+					//defer meta.Unlock()
+					num := 1
+					for {
+						if num == 0 {
+							break
+						}
+						select {
+						case _, ok := <-meta.Event:
+							if !ok {
+								return fmt.Errorf("unexpectedly channel close")
+							}
+							num--
+							if meta.debug {
+								meta.logf("got an event!")
+							}
+						case <-meta.ctx.Done():
+							return meta.ctx.Err()
+						}
+					}
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					meta.Lock()
+					defer meta.Unlock()
+					if meta.EventCount < 1 {
+						return fmt.Errorf("didn't get enough stream events")
+					}
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					//meta.Lock()
+					//defer meta.Unlock()
+					num := 1
+					for {
+						if num == 0 {
+							break
+						}
+						bt := util.BlockedTimer{Seconds: 2}
+						defer bt.Cancel()
+						bt.Printf("waiting for f4...\n")
+						select {
+						case _, ok := <-meta.Events["f4"]:
+							bt.Cancel()
+							if !ok {
+								return fmt.Errorf("unexpectedly channel close")
+							}
+							num--
+							if meta.debug {
+								meta.logf("got an event from f4!")
+							}
+						case <-meta.ctx.Done():
+							return meta.ctx.Err()
+						}
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+
+		testCases = append(testCases, test{
+			name:     "simple add/delete vertex",
+			vertices: []interfaces.Func{f1},
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					defer engine.Unlock()
+					return engine.AddVertex(f1)
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					meta.Lock()
+					defer meta.Unlock()
+					if meta.EventCount < 1 {
+						return fmt.Errorf("didn't get enough stream events")
+					}
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					defer engine.Unlock()
+
+					//meta.Lock()
+					//defer meta.Unlock()
+					if meta.debug {
+						meta.logf("about to delete vertex f1!")
+						defer meta.logf("done deleting vertex f1!")
+					}
+
+					return engine.DeleteVertex(f1)
+				},
+			},
+		})
+	}
+	{
+		f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+		// e1 arg name must match incoming edge to it
+		f2 := &testFunc{Name: "f2", Type: types.NewType("func(e1 str) str")}
+		e1 := testEdge("e1")
+
+		testCases = append(testCases, test{
+			name:     "simple add/delete edge",
+			vertices: []interfaces.Func{f1, f2},
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					defer engine.Unlock()
+					return engine.AddVertex(f1)
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					engine.Lock()
+					defer engine.Unlock()
+					// This newly added node should get a notification after it starts.
+					return engine.AddEdge(f1, f2, e1)
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					meta.Lock()
+					defer meta.Unlock()
+					if meta.EventCount < 2 {
+						return fmt.Errorf("didn't get enough stream events")
+					}
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					engine.Lock()
+					defer engine.Unlock()
+					return engine.DeleteEdge(e1)
+				},
+			},
+		})
+	}
+
+	// the following tests use the txn instead of direct locks
+	{
+		f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+
+		testCases = append(testCases, test{
+			name:     "txn simple add vertex",
+			vertices: []interfaces.Func{f1}, // so the test engine can pass in debug/observability handles
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					return txn.AddVertex(f1).Commit()
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					meta.Lock()
+					defer meta.Unlock()
+					if meta.EventCount < 1 {
+						return fmt.Errorf("didn't get any stream events")
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+		// e1 arg name must match incoming edge to it
+		f2 := &testFunc{Name: "f2", Type: types.NewType("func(e1 str) str")}
+		e1 := testEdge("e1")
+
+		testCases = append(testCases, test{
+			name:     "txn simple add edge",
+			vertices: []interfaces.Func{f1, f2},
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					return txn.AddVertex(f1).Commit()
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					// This newly added node should get a notification after it starts.
+					return txn.AddEdge(f1, f2, e1).Commit()
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					meta.Lock()
+					defer meta.Unlock()
+					if meta.EventCount < 2 {
+						return fmt.Errorf("didn't get enough stream events")
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		// diamond
+		f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+		f2 := &testFunc{Name: "f2", Type: types.NewType("func(e1 str) str")}
+		f3 := &testFunc{Name: "f3", Type: types.NewType("func(e2 str) str")}
+		f4 := &testFunc{Name: "f4", Type: types.NewType("func(e3 str, e4 str) str")}
+		e1 := testEdge("e1")
+		e2 := testEdge("e2")
+		e3 := testEdge("e3")
+		e4 := testEdge("e4")
+
+		testCases = append(testCases, test{
+			name:     "txn simple add multiple edges",
+			vertices: []interfaces.Func{f1, f2, f3, f4},
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					return txn.AddVertex(f1).Commit()
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					return txn.AddEdge(f1, f2, e1).AddEdge(f1, f3, e2).Commit()
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					return txn.AddEdge(f2, f4, e3).AddEdge(f3, f4, e4).Commit()
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					//meta.Lock()
+					//defer meta.Unlock()
+					num := 1
+					for {
+						if num == 0 {
+							break
+						}
+						select {
+						case _, ok := <-meta.Event:
+							if !ok {
+								return fmt.Errorf("unexpectedly channel close")
+							}
+							num--
+							if meta.debug {
+								meta.logf("got an event!")
+							}
+						case <-meta.ctx.Done():
+							return meta.ctx.Err()
+						}
+					}
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					meta.Lock()
+					defer meta.Unlock()
+					if meta.EventCount < 1 {
+						return fmt.Errorf("didn't get enough stream events")
+					}
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					//meta.Lock()
+					//defer meta.Unlock()
+					num := 1
+					for {
+						if num == 0 {
+							break
+						}
+						bt := util.BlockedTimer{Seconds: 2}
+						defer bt.Cancel()
+						bt.Printf("waiting for f4...\n")
+						select {
+						case _, ok := <-meta.Events["f4"]:
+							bt.Cancel()
+							if !ok {
+								return fmt.Errorf("unexpectedly channel close")
+							}
+							num--
+							if meta.debug {
+								meta.logf("got an event from f4!")
+							}
+						case <-meta.ctx.Done():
+							return meta.ctx.Err()
+						}
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+
+		testCases = append(testCases, test{
+			name:     "txn simple add/delete vertex",
+			vertices: []interfaces.Func{f1},
+			actions: []dageTestOp{
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					return txn.AddVertex(f1).Commit()
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					time.Sleep(1 * time.Second) // XXX: unfortunate
+					meta.Lock()
+					defer meta.Unlock()
+					if meta.EventCount < 1 {
+						return fmt.Errorf("didn't get enough stream events")
+					}
+					return nil
+				},
+				func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+					//meta.Lock()
+					//defer meta.Unlock()
+					if meta.debug {
+						meta.logf("about to delete vertex f1!")
+						defer meta.logf("done deleting vertex f1!")
+					}
+
+					return txn.DeleteVertex(f1).Commit()
+				},
+			},
+		})
+	}
+	//{
+	//	f1 := &testFunc{Name: "f1", Type: types.NewType("func() str")}
+	//	// e1 arg name must match incoming edge to it
+	//	f2 := &testFunc{Name: "f2", Type: types.NewType("func(e1 str) str")}
+	//	e1 := testEdge("e1")
+	//
+	//	testCases = append(testCases, test{
+	//		name:     "txn simple add/delete edge",
+	//		vertices: []interfaces.Func{f1, f2},
+	//		actions: []dageTestOp{
+	//			func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+	//				return txn.AddVertex(f1).Commit()
+	//			},
+	//			func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+	//				time.Sleep(1 * time.Second) // XXX: unfortunate
+	//				// This newly added node should get a notification after it starts.
+	//				return txn.AddEdge(f1, f2, e1).Commit()
+	//			},
+	//			func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+	//				time.Sleep(1 * time.Second) // XXX: unfortunate
+	//				meta.Lock()
+	//				defer meta.Unlock()
+	//				if meta.EventCount < 2 {
+	//					return fmt.Errorf("didn't get enough stream events")
+	//				}
+	//				return nil
+	//			},
+	//			func(engine *Engine, txn interfaces.Txn, meta *meta) error {
+	//				return txn.DeleteEdge(e1).Commit() // XXX: not implemented
+	//			},
+	//		},
+	//	})
+	//}
+
+	if testing.Short() {
+		t.Logf("available tests:")
+	}
+	names := []string{}
+	for index, tc := range testCases { // run all the tests
+		if tc.name == "" {
+			t.Errorf("test #%d: not named", index)
+			continue
+		}
+		if util.StrInList(tc.name, names) {
+			t.Errorf("test #%d: duplicate sub test name of: %s", index, tc.name)
+			continue
+		}
+		names = append(names, tc.name)
+
+		//if index != 3 { // hack to run a subset (useful for debugging)
+		//if tc.name != "simple txn" {
+		//	continue
+		//}
+
+		testName := fmt.Sprintf("test #%d (%s)", index, tc.name)
+		if testing.Short() { // make listing tests easier
+			t.Logf("%s", testName)
+			continue
+		}
+		t.Run(testName, func(t *testing.T) {
+			name, vertices, actions := tc.name, tc.vertices, tc.actions
+
+			t.Logf("\n\ntest #%d (%s) ----------------\n\n", index, name)
+
+			//logf := func(format string, v ...interface{}) {
+			//	t.Logf(fmt.Sprintf("test #%d", index)+": "+format, v...)
+			//}
+
+			//now := time.Now()
+
+			wg := &sync.WaitGroup{}
+			defer wg.Wait() // defer is correct b/c we're in a func!
+
+			min := 5 * time.Second // approx min time needed for the test
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			if deadline, ok := t.Deadline(); ok {
+				d := deadline.Add(-min)
+				//t.Logf("  now: %+v", now)
+				//t.Logf("    d: %+v", d)
+				newCtx, cancel := context.WithDeadline(ctx, d)
+				ctx = newCtx
+				defer cancel()
+			}
+
+			debug := testing.Verbose() // set via the -test.v flag to `go test`
+
+			meta := &meta{
+				Event:  make(chan struct{}),
+				Events: make(map[string]chan struct{}),
+
+				ctx:   ctx,
+				wg:    &sync.WaitGroup{},
+				mutex: &sync.Mutex{},
+
+				debug: debug,
+				logf: func(format string, v ...interface{}) {
+					// safe Logf in case f.String contains %? chars...
+					s := fmt.Sprintf(format, v...)
+					t.Logf("%s", s)
+				},
+			}
+			defer meta.wg.Wait()
+
+			for _, f := range vertices {
+				testFunc, ok := f.(*testFunc)
+				if !ok {
+					t.Errorf("bad test function: %+v", f)
+					return
+				}
+				meta.Events[testFunc.Name] = make(chan struct{})
+				testFunc.Meta = meta // add the handle
+			}
+
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				select {
+				case <-ctx.Done():
+					t.Logf("cancelling test...")
+				}
+			}()
+
+			engine := &Engine{
+				Name: "dage",
+
+				Debug: debug,
+				Logf:  t.Logf,
+			}
+
+			if err := engine.Setup(); err != nil {
+				t.Errorf("could not setup engine: %+v", err)
+				return
+			}
+			defer engine.Cleanup()
+
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				if err := engine.Run(ctx); err != nil {
+					t.Errorf("error while running engine: %+v", err)
+					return
+				}
+				t.Logf("engine shutdown cleanly...")
+			}()
+
+			<-engine.Started() // wait for startup (will not block forever)
+
+			txn := engine.Txn()
+			defer txn.Free() // remember to call Free()
+
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				ch := engine.Stream()
+				for {
+					select {
+					case err, ok := <-ch: // channel must close to shutdown
+						if !ok {
+							return
+						}
+						meta.Lock()
+						meta.EventCount++
+						meta.Unlock()
+						meta.wg.Add(1)
+						go func() {
+							// no need for lock here
+							defer meta.wg.Done()
+							if meta.debug {
+								meta.logf("sending an event!")
+							}
+							select {
+							case meta.Event <- struct{}{}:
+							case <-meta.ctx.Done():
+							}
+						}()
+						if err != nil {
+							t.Errorf("graph error event: %v", err)
+							continue
+						}
+						t.Logf("graph stream event!")
+					}
+				}
+			}()
+
+			// Run a list of actions. Any error kills it all.
+			t.Logf("starting actions...")
+			for i, action := range actions {
+				t.Logf("running action %d...", i+1)
+				if err := action(engine, txn, meta); err != nil {
+					t.Errorf("test #%d: FAIL", index)
+					t.Errorf("test #%d: action #%d failed with: %+v", index, i, err)
+					break // so that cancel runs
+				}
+			}
+
+			t.Logf("test done...")
+			cancel()
+		})
+	}
+
+	if testing.Short() {
+		t.Skip("skipping all tests...")
+	}
+}

lang/funcs/dage/ref.go

@@ -0,0 +1,282 @@
+// Mgmt
+// Copyright (C) 2013-2023+ James Shubin and the project contributors
+// Written by James Shubin <james@shubin.ca> and the project contributors
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+// Package dage implements a DAG function engine.
+// TODO: can we rename this to something more interesting?
+package dage
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/purpleidea/mgmt/lang/interfaces"
+	"github.com/purpleidea/mgmt/util/errwrap"
+)
+
+// RefCount keeps track of vertex and edge references across the entire graph.
+// Make sure to lock access somehow, ideally with the provided Locker interface.
+type RefCount struct {
+	// mutex locks this database for read or write.
+	mutex *sync.Mutex
+
+	// vertices is a reference count of the number of vertices used.
+	vertices map[interfaces.Func]int64
+
+	// edges is a reference count of the number of edges used.
+	edges map[*RefCountEdge]int64 // TODO: hash *RefCountEdge as a key instead
+}
+
+// RefCountEdge is a virtual "hash" entry for the RefCount edges map key.
+type RefCountEdge struct {
+	f1  interfaces.Func
+	f2  interfaces.Func
+	arg string
+}
+
+// String prints a representation of the references held.
+func (obj *RefCount) String() string {
+	s := ""
+	s += fmt.Sprintf("vertices (%d):\n", len(obj.vertices))
+	for vertex, count := range obj.vertices {
+		s += fmt.Sprintf("\tvertex (%d): %p %s\n", count, vertex, vertex)
+	}
+	s += fmt.Sprintf("edges (%d):\n", len(obj.edges))
+	for edge, count := range obj.edges {
+		s += fmt.Sprintf("\tedge (%d): %p %s -> %p %s # %s\n", count, edge.f1, edge.f1, edge.f2, edge.f2, edge.arg)
+	}
+	return s
+}
+
+// Init must be called to initialized the struct before first use.
+func (obj *RefCount) Init() *RefCount {
+	obj.mutex = &sync.Mutex{}
+	obj.vertices = make(map[interfaces.Func]int64)
+	obj.edges = make(map[*RefCountEdge]int64)
+	return obj // return self so it can be called in a chain
+}
+
+// Lock the mutex that should be used when reading or writing from this.
+func (obj *RefCount) Lock() { obj.mutex.Lock() }
+
+// Unlock the mutex that should be used when reading or writing from this.
+func (obj *RefCount) Unlock() { obj.mutex.Unlock() }
+
+// VertexInc increments the reference count for the input vertex. It returns
+// true if the reference count for this vertex was previously undefined or zero.
+// True usually means we'd want to actually add this vertex now. If you attempt
+// to increment a vertex which already has a less than zero count, then this
+// will panic. This situation is likely impossible unless someone modified the
+// reference counting struct directly.
+func (obj *RefCount) VertexInc(f interfaces.Func) bool {
+	count, _ := obj.vertices[f]
+	obj.vertices[f] = count + 1
+	if count == -1 { // unlikely, but catch any bugs
+		panic("negative reference count")
+	}
+	return count == 0
+}
+
+// VertexDec decrements the reference count for the input vertex. It returns
+// true if the reference count for this vertex is now zero. True usually means
+// we'd want to actually remove this vertex now. If you attempt to decrement a
+// vertex which already has a zero count, then this will panic.
+func (obj *RefCount) VertexDec(f interfaces.Func) bool {
+	count, _ := obj.vertices[f]
+	obj.vertices[f] = count - 1
+	if count == 0 {
+		panic("negative reference count")
+	}
+	return count == 1 // now it's zero
+}
+
+// EdgeInc increments the reference count for the input edge. It adds a
+// reference for each arg name in the edge. Since this also increments the
+// references for the two input vertices, it returns the corresponding two
+// boolean values for these calls. (This function makes two calls to VertexInc.)
+func (obj *RefCount) EdgeInc(f1, f2 interfaces.Func, fe *interfaces.FuncEdge) (bool, bool) {
+	for _, arg := range fe.Args { // ref count each arg
+		r := obj.makeEdge(f1, f2, arg)
+		count := obj.edges[r]
+		obj.edges[r] = count + 1
+		if count == -1 { // unlikely, but catch any bugs
+			panic("negative reference count")
+		}
+	}
+
+	return obj.VertexInc(f1), obj.VertexInc(f2)
+}
+
+// EdgeDec decrements the reference count for the input edge. It removes a
+// reference for each arg name in the edge. Since this also decrements the
+// references for the two input vertices, it returns the corresponding two
+// boolean values for these calls. (This function makes two calls to VertexDec.)
+func (obj *RefCount) EdgeDec(f1, f2 interfaces.Func, fe *interfaces.FuncEdge) (bool, bool) {
+	for _, arg := range fe.Args { // ref count each arg
+		r := obj.makeEdge(f1, f2, arg)
+		count := obj.edges[r]
+		obj.edges[r] = count - 1
+		if count == 0 {
+			panic("negative reference count")
+		}
+	}
+
+	return obj.VertexDec(f1), obj.VertexDec(f2)
+}
+
+// FreeVertex removes exactly one entry from the Vertices list or it errors.
+func (obj *RefCount) FreeVertex(f interfaces.Func) error {
+	if count, exists := obj.vertices[f]; !exists || count != 0 {
+		return fmt.Errorf("no vertex of count zero found")
+	}
+	delete(obj.vertices, f)
+	return nil
+}
+
+// FreeEdge removes exactly one entry from the Edges list or it errors.
+func (obj *RefCount) FreeEdge(f1, f2 interfaces.Func, arg string) error {
+	found := []*RefCountEdge{}
+	for k, count := range obj.edges {
+		//if k == nil { // programming error
+		//	continue
+		//}
+		if k.f1 == f1 && k.f2 == f2 && k.arg == arg && count == 0 {
+			found = append(found, k)
+		}
+	}
+	if len(found) > 1 {
+		return fmt.Errorf("inconsistent ref count for edge")
+	}
+	if len(found) == 0 {
+		return fmt.Errorf("no edge of count zero found")
+	}
+	delete(obj.edges, found[0]) // delete from map
+	return nil
+}
+
+// GC runs the garbage collector on any zeroed references. Note the distinction
+// between count == 0 (please delete now) and absent from the map.
+func (obj *RefCount) GC(graphAPI interfaces.GraphAPI) error {
+	// debug
+	//fmt.Printf("start refs\n%s", obj.String())
+	//defer func() { fmt.Printf("end refs\n%s", obj.String()) }()
+	free := make(map[interfaces.Func]map[interfaces.Func][]string) // f1 -> f2
+	for x, count := range obj.edges {
+		if count != 0 { // we only care about freed things
+			continue
+		}
+		if _, exists := free[x.f1]; !exists {
+			free[x.f1] = make(map[interfaces.Func][]string)
+		}
+		if _, exists := free[x.f1][x.f2]; !exists {
+			free[x.f1][x.f2] = []string{}
+		}
+		free[x.f1][x.f2] = append(free[x.f1][x.f2], x.arg) // exists as refcount zero
+	}
+
+	// These edges have a refcount of zero.
+	for f1, x := range free {
+		for f2, args := range x {
+			for _, arg := range args {
+				edge := graphAPI.FindEdge(f1, f2)
+				// any errors here are programming errors
+				if edge == nil {
+					return fmt.Errorf("missing edge from %p %s -> %p %s", f1, f1, f2, f2)
+				}
+
+				once := false // sanity check
+				newArgs := []string{}
+				for _, a := range edge.Args {
+					if arg == a {
+						if once {
+							// programming error, duplicate arg
+							return fmt.Errorf("duplicate arg (%s) in edge", arg)
+						}
+						once = true
+						continue
+					}
+					newArgs = append(newArgs, a)
+				}
+
+				if len(edge.Args) == 1 { // edge gets deleted
+					if a := edge.Args[0]; a != arg { // one arg
+						return fmt.Errorf("inconsistent arg: %s != %s", a, arg)
+					}
+
+					if err := graphAPI.DeleteEdge(edge); err != nil {
+						return errwrap.Wrapf(err, "edge deletion error")
+					}
+				} else {
+					// just remove the one arg for now
+					edge.Args = newArgs
+				}
+
+				// always free the database entry
+				if err := obj.FreeEdge(f1, f2, arg); err != nil {
+					return err
+				}
+			}
+		}
+	}
+
+	// Now check the vertices...
+	vs := []interfaces.Func{}
+	for vertex, count := range obj.vertices {
+		if count != 0 {
+			continue
+		}
+
+		// safety check, vertex is still in use by an edge
+		for x := range obj.edges {
+			if x.f1 == vertex || x.f2 == vertex {
+				// programming error
+				return fmt.Errorf("vertex unexpectedly still in use: %p %s", vertex, vertex)
+			}
+		}
+
+		vs = append(vs, vertex)
+	}
+
+	for _, vertex := range vs {
+		if err := graphAPI.DeleteVertex(vertex); err != nil {
+			return errwrap.Wrapf(err, "vertex deletion error")
+		}
+		// free the database entry
+		if err := obj.FreeVertex(vertex); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// makeEdge looks up an edge with the "hash" input we are seeking. If it doesn't
+// find a match, it returns a new one with those fields.
+func (obj *RefCount) makeEdge(f1, f2 interfaces.Func, arg string) *RefCountEdge {
+	for k := range obj.edges {
+		//if k == nil { // programming error
+		//	continue
+		//}
+		if k.f1 == f1 && k.f2 == f2 && k.arg == arg {
+			return k
+		}
+	}
+	return &RefCountEdge{ // not found, so make a new one!
+		f1:  f1,
+		f2:  f2,
+		arg: arg,
+	}
+}

lang/funcs/dage/txn.go

@@ -0,0 +1,626 @@
+// Mgmt
+// Copyright (C) 2013-2023+ James Shubin and the project contributors
+// Written by James Shubin <james@shubin.ca> and the project contributors
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+// Package dage implements a DAG function engine.
+// TODO: can we rename this to something more interesting?
+package dage
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+
+	"github.com/purpleidea/mgmt/lang/interfaces"
+	"github.com/purpleidea/mgmt/pgraph"
+)
+
+// PostReverseCommit specifies that if we run Reverse, and we had previous items
+// pending for Commit, that we should Commit them after our Reverse runs.
+// Otherwise they remain on the pending queue and wait for you to run Commit.
+const PostReverseCommit = false
+
+// GraphvizDebug enables writing graphviz graphs on each commit. This is very
+// slow.
+const GraphvizDebug = false
+
+// opapi is the input for any op. This allows us to keeps things compact and it
+// also allows us to change API slightly without re-writing code.
+type opapi struct {
+	GraphAPI interfaces.GraphAPI
+	RefCount *RefCount
+}
+
+// opfn is an interface that holds the normal op, and the reverse op if we need
+// to rollback from the forward fn. Implementations of each op can decide to
+// store some internal state when running the forward op which might be needed
+// for the possible future reverse op.
+type opfn interface {
+	fmt.Stringer
+
+	Fn(*opapi) error
+	Rev(*opapi) error
+}
+
+type opfnSkipRev interface {
+	opfn
+
+	// Skip tells us if this op should be skipped from reversing.
+	Skip() bool
+
+	// SetSkip specifies that this op should be skipped from reversing.
+	SetSkip(bool)
+}
+
+type opfnFlag interface {
+	opfn
+
+	// Flag reads some misc data.
+	Flag() interface{}
+
+	// SetFlag sets some misc data.
+	SetFlag(interface{})
+}
+
+// revOp returns the reversed op from an op by packing or unpacking it.
+func revOp(op opfn) opfn {
+	if skipOp, ok := op.(opfnSkipRev); ok && skipOp.Skip() {
+		return nil // skip
+	}
+
+	// XXX: is the reverse of a reverse just undoing it? maybe not but might not matter for us
+	if newOp, ok := op.(*opRev); ok {
+
+		if newFlagOp, ok := op.(opfnFlag); ok {
+			newFlagOp.SetFlag("does this rev of rev even happen?")
+		}
+
+		return newOp.Op // unpack it
+	}
+
+	return &opRev{
+		Op: op,
+
+		opFlag: &opFlag{},
+	} // pack it
+}
+
+// opRev switches the Fn and Rev methods by wrapping the contained op in each
+// other.
+type opRev struct {
+	Op opfn
+
+	*opFlag
+}
+
+func (obj *opRev) Fn(opapi *opapi) error {
+	return obj.Op.Rev(opapi)
+}
+
+func (obj *opRev) Rev(opapi *opapi) error {
+	return obj.Op.Fn(opapi)
+}
+
+func (obj *opRev) String() string {
+	return "rev(" + obj.Op.String() + ")" // TODO: is this correct?
+}
+
+type opSkip struct {
+	skip bool
+}
+
+func (obj *opSkip) Skip() bool {
+	return obj.skip
+}
+
+func (obj *opSkip) SetSkip(skip bool) {
+	obj.skip = skip
+}
+
+type opFlag struct {
+	flag interface{}
+}
+
+func (obj *opFlag) Flag() interface{} {
+	return obj.flag
+}
+
+func (obj *opFlag) SetFlag(flag interface{}) {
+	obj.flag = flag
+}
+
+type opAddVertex struct {
+	F interfaces.Func
+
+	*opSkip
+	*opFlag
+}
+
+func (obj *opAddVertex) Fn(opapi *opapi) error {
+	if opapi.RefCount.VertexInc(obj.F) {
+		// add if we're the first reference
+		return opapi.GraphAPI.AddVertex(obj.F)
+	}
+
+	return nil
+}
+
+func (obj *opAddVertex) Rev(opapi *opapi) error {
+	opapi.RefCount.VertexDec(obj.F)
+	// any removal happens in gc
+	return nil
+}
+
+func (obj *opAddVertex) String() string {
+	return fmt.Sprintf("AddVertex: %+v", obj.F)
+}
+
+type opAddEdge struct {
+	F1 interfaces.Func
+	F2 interfaces.Func
+	FE *interfaces.FuncEdge
+
+	*opSkip
+	*opFlag
+}
+
+func (obj *opAddEdge) Fn(opapi *opapi) error {
+	if obj.F1 == obj.F2 { // simplify below code/logic with this easy check
+		return fmt.Errorf("duplicate vertex cycle")
+	}
+
+	opapi.RefCount.EdgeInc(obj.F1, obj.F2, obj.FE)
+
+	fe := obj.FE // squish multiple edges together if one already exists
+	if edge := opapi.GraphAPI.FindEdge(obj.F1, obj.F2); edge != nil {
+		args := make(map[string]struct{})
+		for _, x := range obj.FE.Args {
+			args[x] = struct{}{}
+		}
+		for _, x := range edge.Args {
+			args[x] = struct{}{}
+		}
+		if len(args) != len(obj.FE.Args)+len(edge.Args) {
+			// programming error
+			return fmt.Errorf("duplicate arg found")
+		}
+		newArgs := []string{}
+		for x := range args {
+			newArgs = append(newArgs, x)
+		}
+		sort.Strings(newArgs) // for consistency?
+		fe = &interfaces.FuncEdge{
+			Args: newArgs,
+		}
+	}
+
+	// The dage API currently smooshes together any existing edge args with
+	// our new edge arg names. It also adds the vertices if needed.
+	if err := opapi.GraphAPI.AddEdge(obj.F1, obj.F2, fe); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (obj *opAddEdge) Rev(opapi *opapi) error {
+	opapi.RefCount.EdgeDec(obj.F1, obj.F2, obj.FE)
+	return nil
+}
+
+func (obj *opAddEdge) String() string {
+	return fmt.Sprintf("AddEdge: %+v -> %+v (%+v)", obj.F1, obj.F2, obj.FE)
+}
+
+type opDeleteVertex struct {
+	F interfaces.Func
+
+	*opSkip
+	*opFlag
+}
+
+func (obj *opDeleteVertex) Fn(opapi *opapi) error {
+	if opapi.RefCount.VertexDec(obj.F) {
+		//delete(opapi.RefCount.Vertices, obj.F)    // don't GC this one
+		if err := opapi.RefCount.FreeVertex(obj.F); err != nil {
+			panic("could not free vertex")
+		}
+		return opapi.GraphAPI.DeleteVertex(obj.F) // do it here instead
+	}
+	return nil
+}
+
+func (obj *opDeleteVertex) Rev(opapi *opapi) error {
+	if opapi.RefCount.VertexInc(obj.F) {
+		return opapi.GraphAPI.AddVertex(obj.F)
+	}
+	return nil
+}
+
+func (obj *opDeleteVertex) String() string {
+	return fmt.Sprintf("DeleteVertex: %+v", obj.F)
+}
+
+// graphTxn holds the state of a transaction and runs it when needed. When this
+// has been setup and initialized, it implements the Txn API that can be used by
+// functions in their Stream method to modify the function graph while it is
+// "running".
+type graphTxn struct {
+	// Lock is a handle to the lock function to call before the operation.
+	Lock func()
+
+	// Unlock is a handle to the unlock function to call before the
+	// operation.
+	Unlock func()
+
+	// GraphAPI is a handle pointing to the graph API implementation we're
+	// using for any txn operations.
+	GraphAPI interfaces.GraphAPI
+
+	// RefCount keeps track of vertex and edge references across the entire
+	// graph.
+	RefCount *RefCount
+
+	// FreeFunc is a function that will get called by a well-behaved user
+	// when we're done with this Txn.
+	FreeFunc func()
+
+	// ops is a list of operations to run on a graph
+	ops []opfn
+
+	// rev is a list of reverse operations to run on a graph
+	rev []opfn
+
+	// mutex guards changes to the ops list
+	mutex *sync.Mutex
+}
+
+// init must be called to initialized the struct before first use. This is
+// private because the creator, not the user should run it.
+func (obj *graphTxn) init() interfaces.Txn {
+	obj.ops = []opfn{}
+	obj.rev = []opfn{}
+	obj.mutex = &sync.Mutex{}
+
+	return obj // return self so it can be called in a chain
+}
+
+// Copy returns a new child Txn that has the same handles, but a separate state.
+// This allows you to do an Add*/Commit/Reverse that isn't affected by a
+// different user of this transaction.
+// TODO: FreeFunc isn't well supported here. Replace or remove this entirely?
+func (obj *graphTxn) Copy() interfaces.Txn {
+	txn := &graphTxn{
+		Lock:     obj.Lock,
+		Unlock:   obj.Unlock,
+		GraphAPI: obj.GraphAPI,
+		RefCount: obj.RefCount, // this is shared across all txn's
+		// FreeFunc is shared with the parent.
+	}
+	return txn.init()
+}
+
+// AddVertex adds a vertex to the running graph. The operation will get
+// completed when Commit is run.
+// XXX: should this be pgraph.Vertex instead of interfaces.Func ?
+func (obj *graphTxn) AddVertex(f interfaces.Func) interfaces.Txn {
+	obj.mutex.Lock()
+	defer obj.mutex.Unlock()
+
+	opfn := &opAddVertex{
+		F: f,
+
+		opSkip: &opSkip{},
+		opFlag: &opFlag{},
+	}
+	obj.ops = append(obj.ops, opfn)
+
+	return obj // return self so it can be called in a chain
+}
+
+// AddEdge adds an edge to the running graph. The operation will get completed
+// when Commit is run.
+// XXX: should this be pgraph.Vertex instead of interfaces.Func ?
+// XXX: should this be pgraph.Edge instead of *interfaces.FuncEdge ?
+func (obj *graphTxn) AddEdge(f1, f2 interfaces.Func, fe *interfaces.FuncEdge) interfaces.Txn {
+	obj.mutex.Lock()
+	defer obj.mutex.Unlock()
+
+	opfn := &opAddEdge{
+		F1: f1,
+		F2: f2,
+		FE: fe,
+
+		opSkip: &opSkip{},
+		opFlag: &opFlag{},
+	}
+	obj.ops = append(obj.ops, opfn)
+
+	// NOTE: we can't build obj.rev yet because in this case, we'd need to
+	// know if the runtime graph contained one of the two pre-existing
+	// vertices or not, or if it would get added implicitly by this op!
+
+	return obj // return self so it can be called in a chain
+}
+
+// DeleteVertex adds a vertex to the running graph. The operation will get
+// completed when Commit is run.
+// XXX: should this be pgraph.Vertex instead of interfaces.Func ?
+func (obj *graphTxn) DeleteVertex(f interfaces.Func) interfaces.Txn {
+	obj.mutex.Lock()
+	defer obj.mutex.Unlock()
+
+	opfn := &opDeleteVertex{
+		F: f,
+
+		opSkip: &opSkip{},
+		opFlag: &opFlag{},
+	}
+	obj.ops = append(obj.ops, opfn)
+
+	return obj // return self so it can be called in a chain
+}
+
+// AddGraph adds a graph to the running graph. The operation will get completed
+// when Commit is run. This function panics if your graph contains vertices that
+// are not of type interfaces.Func or if your edges are not of type
+// *interfaces.FuncEdge.
+func (obj *graphTxn) AddGraph(g *pgraph.Graph) interfaces.Txn {
+	obj.mutex.Lock()
+	defer obj.mutex.Unlock()
+
+	for _, v := range g.Vertices() {
+		f, ok := v.(interfaces.Func)
+		if !ok {
+			panic("not a Func")
+		}
+		//obj.AddVertex(f) // easy
+		opfn := &opAddVertex{ // replicate AddVertex
+			F: f,
+
+			opSkip: &opSkip{},
+			opFlag: &opFlag{},
+		}
+		obj.ops = append(obj.ops, opfn)
+	}
+
+	for v1, m := range g.Adjacency() {
+		f1, ok := v1.(interfaces.Func)
+		if !ok {
+			panic("not a Func")
+		}
+		for v2, e := range m {
+			f2, ok := v2.(interfaces.Func)
+			if !ok {
+				panic("not a Func")
+			}
+			fe, ok := e.(*interfaces.FuncEdge)
+			if !ok {
+				panic("not a *FuncEdge")
+			}
+
+			//obj.AddEdge(f1, f2, fe) // easy
+			opfn := &opAddEdge{ // replicate AddEdge
+				F1: f1,
+				F2: f2,
+				FE: fe,
+
+				opSkip: &opSkip{},
+				opFlag: &opFlag{},
+			}
+			obj.ops = append(obj.ops, opfn)
+		}
+	}
+
+	return obj // return self so it can be called in a chain
+}
+
+// commit runs the pending transaction. This is the lockless version that is
+// only used internally.
+func (obj *graphTxn) commit() error {
+	if len(obj.ops) == 0 { // nothing to do
+		return nil
+	}
+
+	// TODO: Instead of requesting the below locks, it's conceivable that we
+	// could either write an engine that doesn't require pausing the graph
+	// with a lock, or one that doesn't in the specific case being changed
+	// here need locks. And then in theory we'd have improved performance
+	// from the function engine. For our function consumers, the Txn API
+	// would never need to change, so we don't break API! A simple example
+	// is the len(ops) == 0 one right above. A simplification, but shows we
+	// aren't forced to call the locks even when we get Commit called here.
+
+	// Now request the lock from the actual graph engine.
+	obj.Lock()
+	defer obj.Unlock()
+
+	// Now request the ref count mutex. This may seem redundant, but it's
+	// not. The above graph engine Lock might allow more than one commit
+	// through simultaneously depending on implementation. The actual count
+	// mathematics must not, and so it has a separate lock. We could lock it
+	// per-operation, but that would probably be a lot slower.
+	obj.RefCount.Lock()
+	defer obj.RefCount.Unlock()
+
+	// TODO: we don't need to do this anymore, because the engine does it!
+	// Copy the graph structure, perform the ops, check we didn't add a
+	// cycle, and if it's safe, do the real thing. Otherwise error here.
+	//g := obj.Graph.Copy() // copy the graph structure
+	//for _, x := range obj.ops {
+	//	x(g) // call it
+	//}
+	//if _, err := g.TopologicalSort(); err != nil {
+	//	return errwrap.Wrapf(err, "topo sort failed in txn commit")
+	//}
+	// FIXME: is there anything else we should check? Should we type-check?
+
+	// Now do it for real...
+	obj.rev = []opfn{} // clear it for safety
+	opapi := &opapi{
+		GraphAPI: obj.GraphAPI,
+		RefCount: obj.RefCount,
+	}
+	for _, op := range obj.ops {
+		if err := op.Fn(opapi); err != nil { // call it
+			// something went wrong (we made a cycle?)
+			obj.rev = []opfn{} // clear it, we didn't succeed
+			return err
+		}
+
+		op = revOp(op) // reverse the op!
+		if op != nil {
+			obj.rev = append(obj.rev, op) // add the reverse op
+			//obj.rev = append([]opfn{op}, obj.rev...) // add to front
+		}
+	}
+	obj.ops = []opfn{} // clear it
+
+	// garbage collect anything that hit zero!
+	// XXX: add gc function to this struct and pass in opapi instead?
+	if err := obj.RefCount.GC(obj.GraphAPI); err != nil {
+		// programming error or ghosts
+		return err
+	}
+
+	// XXX: running this on each commit has a huge performance hit.
+	// XXX: we could write out the .dot files and run graphviz afterwards
+	if engine, ok := obj.GraphAPI.(*Engine); ok && GraphvizDebug {
+		//d := time.Now().Unix()
+		//if err := engine.graph.ExecGraphviz(fmt.Sprintf("/tmp/txn-graphviz-%d.dot", d)); err != nil {
+		//	panic("no graphviz")
+		//}
+		if err := engine.Graphviz(""); err != nil {
+			panic(err) // XXX: improve me
+		}
+
+		//gv := &pgraph.Graphviz{
+		//	Filename: fmt.Sprintf("/tmp/txn-graphviz-%d.dot", d),
+		//	Graphs: map[*pgraph.Graph]*pgraph.GraphvizOpts{
+		//		engine.graph: nil,
+		//	},
+		//}
+		//if err := gv.Exec(); err != nil {
+		//	panic("no graphviz")
+		//}
+	}
+	return nil
+}
+
+// Commit runs the pending transaction. If there was a pending reverse
+// transaction that could have run (it would have been available following a
+// Commit success) then this will erase that transaction. Usually you run cycles
+// of Commit, followed by Reverse, or only Commit. (You obviously have to
+// populate operations before the Commit is run.)
+func (obj *graphTxn) Commit() error {
+	// Lock our internal state mutex first... this prevents other AddVertex
+	// or similar calls from interferring with our work here.
+	obj.mutex.Lock()
+	defer obj.mutex.Unlock()
+
+	return obj.commit()
+}
+
+// Clear erases any pending transactions that weren't committed yet.
+func (obj *graphTxn) Clear() {
+	obj.mutex.Lock()
+	defer obj.mutex.Unlock()
+
+	obj.ops = []opfn{} // clear it
+}
+
+// Reverse is like Commit, but it commits the reverse transaction to the one
+// that previously ran with Commit. If the PostReverseCommit global has been set
+// then if there were pending commit operations when this was run, then they are
+// run at the end of a successful Reverse. It is generally recommended to not
+// queue any operations for Commit if you plan on doing a Reverse, or to run a
+// Clear before running Reverse if you want to discard the pending commits.
+func (obj *graphTxn) Reverse() error {
+	obj.mutex.Lock()
+	defer obj.mutex.Unlock()
+
+	// first commit all the rev stuff... and then run the pending ops...
+
+	ops := []opfn{}              // save a copy
+	for _, op := range obj.ops { // copy
+		ops = append(ops, op)
+	}
+	obj.ops = []opfn{} // clear
+
+	//for _, op := range obj.rev
+	for i := len(obj.rev) - 1; i >= 0; i-- { // copy in the rev stuff to commit!
+		op := obj.rev[i]
+		// mark these as being not reversible (so skip them on reverse!)
+		if skipOp, ok := op.(opfnSkipRev); ok {
+			skipOp.SetSkip(true)
+		}
+		obj.ops = append(obj.ops, op)
+	}
+
+	//rev := []func(interfaces.GraphAPI){} // for the copy
+	//for _, op := range obj.rev { // copy
+	//	rev = append(rev, op)
+	//}
+	obj.rev = []opfn{} // clear
+
+	//rollback := func() {
+	//	//for _, op := range rev { // from our safer copy
+	//	//for _, op := range obj.ops { // copy back out the rev stuff
+	//	for i := len(obj.ops) - 1; i >= 0; i-- { // copy in the rev stuff to commit!
+	//		op := obj.rev[i]
+	//		obj.rev = append(obj.rev, op)
+	//	}
+	//	obj.ops = []opfn{}       // clear
+	//	for _, op := range ops { // copy the original ops back in
+	//		obj.ops = append(obj.ops, op)
+	//	}
+	//}
+	// first commit the reverse stuff
+	if err := obj.commit(); err != nil { // lockless version
+		// restore obj.rev and obj.ops
+		//rollback() // probably not needed
+		return err
+	}
+
+	// then if we had normal ops queued up, run those or at least restore...
+	for _, op := range ops { // copy
+		obj.ops = append(obj.ops, op)
+	}
+
+	if PostReverseCommit {
+		return obj.commit() // lockless version
+	}
+
+	return nil
+}
+
+// Erase removes the historical information that Reverse would run after Commit.
+func (obj *graphTxn) Erase() {
+	obj.mutex.Lock()
+	defer obj.mutex.Unlock()
+
+	obj.rev = []opfn{} // clear it
+}
+
+// Free releases the wait group that was used to lock around this Txn if needed.
+// It should get called when we're done with any Txn.
+// TODO: this is only used for the initial Txn. Consider expanding it's use. We
+// might need to allow Clear to call it as part of the clearing.
+func (obj *graphTxn) Free() {
+	if obj.FreeFunc != nil {
+		obj.FreeFunc()
+	}
+}

lang/funcs/dage/txn_test.go

@@ -0,0 +1,503 @@
+// Mgmt
+// Copyright (C) 2013-2023+ James Shubin and the project contributors
+// Written by James Shubin <james@shubin.ca> and the project contributors
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//go:build !root
+
+package dage
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/purpleidea/mgmt/lang/interfaces"
+	"github.com/purpleidea/mgmt/pgraph"
+	"github.com/purpleidea/mgmt/util"
+)
+
+type testGraphAPI struct {
+	graph *pgraph.Graph
+}
+
+func (obj *testGraphAPI) AddVertex(f interfaces.Func) error {
+	v, ok := f.(pgraph.Vertex)
+	if !ok {
+		return fmt.Errorf("can't use func as vertex")
+	}
+	obj.graph.AddVertex(v)
+	return nil
+}
+func (obj *testGraphAPI) AddEdge(f1, f2 interfaces.Func, fe *interfaces.FuncEdge) error {
+	v1, ok := f1.(pgraph.Vertex)
+	if !ok {
+		return fmt.Errorf("can't use func as vertex")
+	}
+	v2, ok := f2.(pgraph.Vertex)
+	if !ok {
+		return fmt.Errorf("can't use func as vertex")
+	}
+	obj.graph.AddEdge(v1, v2, fe)
+	return nil
+}
+
+func (obj *testGraphAPI) DeleteVertex(f interfaces.Func) error {
+	v, ok := f.(pgraph.Vertex)
+	if !ok {
+		return fmt.Errorf("can't use func as vertex")
+	}
+	obj.graph.DeleteVertex(v)
+	return nil
+}
+
+func (obj *testGraphAPI) DeleteEdge(fe *interfaces.FuncEdge) error {
+	obj.graph.DeleteEdge(fe)
+	return nil
+}
+
+//func (obj *testGraphAPI) AddGraph(*pgraph.Graph) error {
+//	return fmt.Errorf("not implemented")
+//}
+
+//func (obj *testGraphAPI) Adjacency() map[interfaces.Func]map[interfaces.Func]*interfaces.FuncEdge {
+//	panic("not implemented")
+//}
+
+func (obj *testGraphAPI) HasVertex(f interfaces.Func) bool {
+	v, ok := f.(pgraph.Vertex)
+	if !ok {
+		panic("can't use func as vertex")
+	}
+	return obj.graph.HasVertex(v)
+}
+
+func (obj *testGraphAPI) LookupEdge(fe *interfaces.FuncEdge) (interfaces.Func, interfaces.Func, bool) {
+	v1, v2, b := obj.graph.LookupEdge(fe)
+	if !b {
+		return nil, nil, b
+	}
+
+	f1, ok := v1.(interfaces.Func)
+	if !ok {
+		panic("can't use vertex as func")
+	}
+	f2, ok := v2.(interfaces.Func)
+	if !ok {
+		panic("can't use vertex as func")
+	}
+	return f1, f2, true
+}
+
+func (obj *testGraphAPI) FindEdge(f1, f2 interfaces.Func) *interfaces.FuncEdge {
+	edge := obj.graph.FindEdge(f1, f2)
+	if edge == nil {
+		return nil
+	}
+	fe, ok := edge.(*interfaces.FuncEdge)
+	if !ok {
+		panic("edge is not a FuncEdge")
+	}
+
+	return fe
+}
+
+type testNullFunc struct {
+	name string
+}
+
+func (obj *testNullFunc) String() string               { return obj.name }
+func (obj *testNullFunc) Info() *interfaces.Info       { return nil }
+func (obj *testNullFunc) Validate() error              { return nil }
+func (obj *testNullFunc) Init(*interfaces.Init) error  { return nil }
+func (obj *testNullFunc) Stream(context.Context) error { return nil }
+
+func TestTxn1(t *testing.T) {
+	graph, err := pgraph.NewGraph("test")
+	if err != nil {
+		t.Errorf("err: %+v", err)
+		return
+	}
+	testGraphAPI := &testGraphAPI{graph: graph}
+	mutex := &sync.Mutex{}
+
+	graphTxn := &graphTxn{
+		GraphAPI: testGraphAPI,
+		Lock:     mutex.Lock,
+		Unlock:   mutex.Unlock,
+		RefCount: (&RefCount{}).Init(),
+	}
+	txn := graphTxn.init()
+
+	f1 := &testNullFunc{"f1"}
+
+	if err := txn.AddVertex(f1).Commit(); err != nil {
+		t.Errorf("commit err: %+v", err)
+		return
+	}
+
+	if l, i := len(graph.Adjacency()), 1; l != i {
+		t.Errorf("got len of: %d", l)
+		t.Errorf("exp len of: %d", i)
+		return
+	}
+
+	if err := txn.Reverse(); err != nil {
+		t.Errorf("reverse err: %+v", err)
+		return
+	}
+
+	if l, i := len(graph.Adjacency()), 0; l != i {
+		t.Errorf("got len of: %d", l)
+		t.Errorf("exp len of: %d", i)
+		return
+	}
+}
+
+type txnTestOp func(*pgraph.Graph, interfaces.Txn) error
+
+func TestTxnTable(t *testing.T) {
+
+	type test struct { // an individual test
+		name    string
+		actions []txnTestOp
+	}
+	testCases := []test{}
+	{
+		f1 := &testNullFunc{"f1"}
+
+		testCases = append(testCases, test{
+			name: "simple add vertex",
+			actions: []txnTestOp{
+				//func(g *pgraph.Graph, txn interfaces.Txn) error {
+				//	txn.AddVertex(f1)
+				//	return nil
+				//},
+				//func(g *pgraph.Graph, txn interfaces.Txn) error {
+				//	return txn.Commit()
+				//},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.AddVertex(f1).Commit()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 1; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.Reverse()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 0; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		f1 := &testNullFunc{"f1"}
+		f2 := &testNullFunc{"f2"}
+		e1 := testEdge("e1")
+
+		testCases = append(testCases, test{
+			name: "simple add edge",
+			actions: []txnTestOp{
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.AddEdge(f1, f2, e1).Commit()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 2; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Edges()), 1; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.Reverse()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 0; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Edges()), 0; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		f1 := &testNullFunc{"f1"}
+		f2 := &testNullFunc{"f2"}
+		e1 := testEdge("e1")
+
+		testCases = append(testCases, test{
+			name: "simple add edge two-step",
+			actions: []txnTestOp{
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.AddVertex(f1).Commit()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.AddEdge(f1, f2, e1).Commit()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 2; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Edges()), 1; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				// Reverse only undoes what happened since the
+				// previous commit, so only one of the nodes is
+				// left at the end.
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.Reverse()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 1; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Edges()), 0; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		f1 := &testNullFunc{"f1"}
+		f2 := &testNullFunc{"f2"}
+		e1 := testEdge("e1")
+
+		testCases = append(testCases, test{
+			name: "simple two add edge, reverse",
+			actions: []txnTestOp{
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.AddVertex(f1).AddVertex(f2).Commit()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.AddEdge(f1, f2, e1).Commit()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 2; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Edges()), 1; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				// Reverse only undoes what happened since the
+				// previous commit, so only one of the nodes is
+				// left at the end.
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.Reverse()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 2; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Edges()), 0; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+			},
+		})
+	}
+	{
+		f1 := &testNullFunc{"f1"}
+		f2 := &testNullFunc{"f2"}
+		f3 := &testNullFunc{"f3"}
+		f4 := &testNullFunc{"f4"}
+		e1 := testEdge("e1")
+		e2 := testEdge("e2")
+		e3 := testEdge("e3")
+		e4 := testEdge("e4")
+
+		testCases = append(testCases, test{
+			name: "simple add/delete",
+			actions: []txnTestOp{
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					txn.AddVertex(f1).AddEdge(f1, f2, e1)
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					txn.AddVertex(f1).AddEdge(f1, f3, e2)
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.Commit()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 3; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Edges()), 2; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					txn.AddEdge(f2, f4, e3)
+					txn.AddEdge(f3, f4, e4)
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.Commit()
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 4; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Edges()), 4; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				// debug
+				//func(g *pgraph.Graph, txn interfaces.Txn) error {
+				//	fileName := "/tmp/graphviz-txn1.dot"
+				//	if err := g.ExecGraphviz(fileName); err != nil {
+				//		return fmt.Errorf("writing graph failed at: %s, err: %+v", fileName, err)
+				//	}
+				//	return nil
+				//},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					return txn.Reverse()
+				},
+				// debug
+				//func(g *pgraph.Graph, txn interfaces.Txn) error {
+				//	fileName := "/tmp/graphviz-txn2.dot"
+				//	if err := g.ExecGraphviz(fileName); err != nil {
+				//		return fmt.Errorf("writing graph failed at: %s, err: %+v", fileName, err)
+				//	}
+				//	return nil
+				//},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Adjacency()), 3; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+				func(g *pgraph.Graph, txn interfaces.Txn) error {
+					if l, i := len(g.Edges()), 2; l != i {
+						return fmt.Errorf("got len of: %d, exp len of: %d", l, i)
+					}
+					return nil
+				},
+			},
+		})
+	}
+	if testing.Short() {
+		t.Logf("available tests:")
+	}
+	names := []string{}
+	for index, tc := range testCases { // run all the tests
+		if tc.name == "" {
+			t.Errorf("test #%d: not named", index)
+			continue
+		}
+		if util.StrInList(tc.name, names) {
+			t.Errorf("test #%d: duplicate sub test name of: %s", index, tc.name)
+			continue
+		}
+		names = append(names, tc.name)
+
+		//if index != 3 { // hack to run a subset (useful for debugging)
+		//if tc.name != "simple txn" {
+		//	continue
+		//}
+
+		testName := fmt.Sprintf("test #%d (%s)", index, tc.name)
+		if testing.Short() { // make listing tests easier
+			t.Logf("%s", testName)
+			continue
+		}
+		t.Run(testName, func(t *testing.T) {
+			name, actions := tc.name, tc.actions
+
+			t.Logf("\n\ntest #%d (%s) ----------------\n\n", index, name)
+
+			//logf := func(format string, v ...interface{}) {
+			//	t.Logf(fmt.Sprintf("test #%d", index)+": "+format, v...)
+			//}
+
+			graph, err := pgraph.NewGraph("test")
+			if err != nil {
+				t.Errorf("err: %+v", err)
+				return
+			}
+			testGraphAPI := &testGraphAPI{graph: graph}
+			mutex := &sync.Mutex{}
+
+			graphTxn := &graphTxn{
+				GraphAPI: testGraphAPI,
+				Lock:     mutex.Lock,
+				Unlock:   mutex.Unlock,
+				RefCount: (&RefCount{}).Init(),
+			}
+			txn := graphTxn.init()
+
+			// Run a list of actions, passing the returned txn (if
+			// any) to the next action. Any error kills it all.
+			for i, action := range actions {
+				if err := action(graph, txn); err != nil {
+					t.Errorf("test #%d: FAIL", index)
+					t.Errorf("test #%d: action #%d failed with: %+v", index, i, err)
+					return
+				}
+			}
+		})
+	}
+
+	if testing.Short() {
+		t.Skip("skipping all tests...")
+	}
+}

lang/funcs/dage/util_test.go

@@ -0,0 +1,30 @@
+// Mgmt
+// Copyright (C) 2013-2023+ James Shubin and the project contributors
+// Written by James Shubin <james@shubin.ca> and the project contributors
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//go:build !root
+
+package dage
+
+import (
+	"github.com/purpleidea/mgmt/lang/interfaces"
+)
+
+func testEdge(name string) *interfaces.FuncEdge {
+	return &interfaces.FuncEdge{
+		Args: []string{name},
+	}
+}

lang/interfaces/func.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/purpleidea/mgmt/engine"
 	"github.com/purpleidea/mgmt/lang/types"
+	"github.com/purpleidea/mgmt/pgraph"
 )
 
 // FuncSig is the simple signature that is used throughout our implementations.
@@ -45,8 +46,20 @@ type Info struct {
 type Init struct {
 	Hostname string // uuid for the host
 	//Noop bool
-	Input  chan types.Value // Engine will close `input` chan
+
-	Output chan types.Value // Stream must close `output` chan
+	// Input is where a chan (stream) of values will get sent to this node.
+	// The engine will close this `input` chan.
+	Input chan types.Value
+
+	// Output is the chan (stream) of values to get sent out from this node.
+	// The Stream function must close this `output` chan.
+	Output chan types.Value
+
+	// Txn provides a transaction API that can be used to modify the
+	// function graph while it is "running". This should not be used by most
+	// nodes, and when it is used, it should be used carefully.
+	Txn Txn
+
 	// TODO: should we pass in a *Scope here for functions like template() ?
 	World engine.World
 	Debug bool
@@ -230,3 +243,93 @@ type FuncEdge struct {
 func (obj *FuncEdge) String() string {
 	return strings.Join(obj.Args, ", ")
 }
+
+// GraphAPI is a subset of the available graph operations that are possible on a
+// pgraph that is used for storing functions. The minimum subset are those which
+// are needed for implementing the Txn interface.
+type GraphAPI interface {
+	AddVertex(Func) error
+	AddEdge(Func, Func, *FuncEdge) error
+	DeleteVertex(Func) error
+	DeleteEdge(*FuncEdge) error
+	//AddGraph(*pgraph.Graph) error
+
+	//Adjacency() map[Func]map[Func]*FuncEdge
+	HasVertex(Func) bool
+	FindEdge(Func, Func) *FuncEdge
+	LookupEdge(*FuncEdge) (Func, Func, bool)
+}
+
+// Txn is the interface that the engine graph API makes available so that
+// functions can modify the function graph dynamically while it is "running".
+// This could be implemented in one of two methods.
+//
+// Method 1: Have a pair of graph Lock and Unlock methods. Queue up the work to
+// do and when we "commit" the transaction, we're just queuing up the work to do
+// and then we run it all surrounded by the lock.
+//
+// Method 2: It's possible that we might eventually be able to actually modify
+// the running graph without even causing it to pause at all. In this scenario,
+// the "commit" would just directly perform those operations without even using
+// the Lock and Unlock mutex operations. This is why we don't expose those in
+// the API. It's also safer because someone can't forget to run Unlock which
+// would block the whole code base.
+type Txn interface {
+	// AddVertex adds a vertex to the running graph. The operation will get
+	// completed when Commit is run.
+	AddVertex(Func) Txn
+
+	// AddEdge adds an edge to the running graph. The operation will get
+	// completed when Commit is run.
+	AddEdge(Func, Func, *FuncEdge) Txn
+
+	// DeleteVertex removes a vertex from the running graph. The operation
+	// will get completed when Commit is run.
+	DeleteVertex(Func) Txn
+
+	// DeleteEdge removes an edge from the running graph. It removes the
+	// edge that is found between the two input vertices. The operation will
+	// get completed when Commit is run. The edge is part of the signature
+	// so that it is both symmetrical with AddEdge, and also easier to
+	// reverse in theory.
+	// NOTE: This is not supported since there's no sane Reverse with GC.
+	// XXX: Add this in but just don't let it be reversible?
+	//DeleteEdge(Func, Func, *FuncEdge) Txn
+
+	// AddGraph adds a graph to the running graph. The operation will get
+	// completed when Commit is run. This function panics if your graph
+	// contains vertices that are not of type interfaces.Func or if your
+	// edges are not of type *interfaces.FuncEdge.
+	AddGraph(*pgraph.Graph) Txn
+
+	// Commit runs the pending transaction.
+	Commit() error
+
+	// Clear erases any pending transactions that weren't committed yet.
+	Clear()
+
+	// Reverse runs the reverse commit of the last successful operation to
+	// Commit. AddVertex is reversed by DeleteVertex, and vice-versa, and
+	// the same for AddEdge and DeleteEdge. Keep in mind that if AddEdge is
+	// called with either vertex not already part of the graph, it will
+	// implicitly add them, but the Reverse operation will not necessarily
+	// know that. As a result, it's recommended to not perform operations
+	// that have implicit Adds or Deletes. Notwithstanding the above, the
+	// initial Txn implementation can and does try to track these changes
+	// so that it can correctly reverse them, but this is not guaranteed by
+	// API, and it could contain bugs.
+	Reverse() error
+
+	// Erase removes the historical information that Reverse would run after
+	// Commit.
+	Erase()
+
+	// Free releases the wait group that was used to lock around this Txn if
+	// needed. It should get called when we're done with any Txn.
+	Free()
+
+	// Copy returns a new child Txn that has the same handles, but a
+	// separate state. This allows you to do an Add*/Commit/Reverse that
+	// isn't affected by a different user of this transaction.
+	Copy() Txn
+}