added migration tooling

Author: emilbon99

.github/workflows/fuzz.go.yaml

@@ -1,4 +1,4 @@
-name: Fuzz - Go Codecs
+name: Fuzz - Go
 
 on:
   pull_request:
@@ -15,18 +15,11 @@ jobs:
       matrix:
         include:
           - module: x/go
-            packages: >-
-              ./color/ ./label/ ./spatial/ ./telem/
+            artifact: x-go
           - module: core
-            packages: >-
-              ./pkg/distribution/channel/ ./pkg/distribution/group/
-              ./pkg/distribution/ontology/ ./pkg/service/arc/ ./pkg/service/device/
-              ./pkg/service/lineplot/ ./pkg/service/log/ ./pkg/service/rack/
-              ./pkg/service/ranger/ ./pkg/service/schematic/ ./pkg/service/table/
-              ./pkg/service/user/ ./pkg/service/workspace/
+            artifact: core
           - module: arc/go
-            packages: >-
-              ./compiler/ ./graph/ ./ir/ ./program/ ./text/ ./types/
+            artifact: arc-go
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6
@@ -41,10 +34,16 @@ jobs:
 
       - name: Fuzz
         working-directory: ${{ matrix.module }}
+        # go test -fuzz requires a single package (cannot use ./...), so we
+        # discover packages containing fuzz targets via grep and iterate.
         run: |
-          for pkg in ${{ matrix.packages }}; do
+          packages=$(grep -rl '^func Fuzz' --include='*_test.go' . | sed 's|/[^/]*$||' | sort -u)
+          if [ -z "$packages" ]; then
+            echo "No fuzz targets found"
+            exit 0
+          fi
+          for pkg in $packages; do
             echo "::group::Fuzzing $pkg"
-            # List all fuzz targets in the package
             targets=$(go test "$pkg" -list '^Fuzz' -run '^$' 2>/dev/null | grep '^Fuzz' || true)
             for target in $targets; do
               echo "--- $target ---"
@@ -57,5 +56,5 @@ jobs:
         if: failure()
         uses: actions/upload-artifact@v4
         with:
-          name: fuzz-crashes-${{ matrix.module }}
+          name: fuzz-crashes-${{ matrix.artifact }}
           path: ${{ matrix.module }}/**/testdata/fuzz/

core/pkg/distribution/ontology/writer_dag.go

@@ -14,8 +14,8 @@ import (
 	"maps"
 
 	"github.com/samber/lo"
-	"github.com/synnaxlabs/x/errors"
 	"github.com/synnaxlabs/x/gorp"
+	"github.com/synnaxlabs/x/graph"
 )
 
 // dagWriter is a key-value backed directed acyclic graph that implements the Writer
@@ -30,7 +30,7 @@ type dagWriter struct {
 var _ Writer = dagWriter{}
 
 // ErrCycle is returned when a cycle is created in the graph.
-var ErrCycle = errors.New("[ontology] - cyclic dependency")
+var ErrCycle = graph.ErrCyclicDependency
 
 // DefineResource implements the Writer interface.
 func (d dagWriter) DefineResource(ctx context.Context, tk ID) error {

core/pkg/distribution/ontology/writer_test.go

@@ -106,7 +106,7 @@ var _ = Describe("Writer", func() {
 						Expect(w.DefineRelationship(ctx, idOne, ontology.RelationshipTypeParentOf, idTwo)).To(Succeed())
 						err := w.DefineRelationship(ctx, idTwo, ontology.RelationshipTypeParentOf, idOne)
 						Expect(err).To(HaveOccurred())
-						Expect(errors.Is(err, ontology.ErrCycle)).To(BeTrue())
+						Expect(err).To(MatchError(ontology.ErrCycle))
 					},
 				)
 				It("Should return an error is a relationships creates a cycle",
@@ -117,7 +117,7 @@ var _ = Describe("Writer", func() {
 						Expect(w.DefineRelationship(ctx, idTwo, ontology.RelationshipTypeParentOf, idThree)).To(Succeed())
 						err := w.DefineRelationship(ctx, idThree, ontology.RelationshipTypeParentOf, idOne)
 						Expect(err).To(HaveOccurred())
-						Expect(errors.Is(err, ontology.ErrCycle)).To(BeTrue())
+						Expect(err).To(MatchError(ontology.ErrCycle))
 					})
 			})
 		})
@@ -155,7 +155,7 @@ var _ = Describe("Writer", func() {
 					[]ontology.ID{idOne},
 				)
 				Expect(err).To(HaveOccurred())
-				Expect(errors.Is(err, ontology.ErrCycle)).To(BeTrue())
+				Expect(err).To(MatchError(ontology.ErrCycle))
 			})
 		})
 		Describe("Deleting a Relationship", func() {

x/go/encoding/orc/reader.go

@@ -10,14 +10,14 @@
 package orc
 
 import (
-	"errors"
-	"fmt"
 	"io"
 	"math"
+
+	"github.com/synnaxlabs/x/errors"
 )
 
 // ErrRecursionDepth is returned when decoding exceeds the maximum recursion depth.
-var ErrRecursionDepth = errors.New("orc: recursion depth exceeded")
+var ErrRecursionDepth = errors.New("[orc] recursion depth exceeded")
 
 // MaxStringLen is the maximum length of a string that can be decoded. In direct
 // mode (ResetBytes), the backing slice provides a natural bound. In io.Reader
@@ -185,7 +185,7 @@ func (r *Reader) String() (string, error) {
 		return s, nil
 	}
 	if n > MaxStringLen {
-		return "", fmt.Errorf(
+		return "", errors.Newf(
 			"orc: string length %d exceeds maximum %d", n, MaxStringLen,
 		)
 	}

x/go/gorp/migrate.go

@@ -16,6 +16,7 @@ import (
 	"github.com/synnaxlabs/alamos"
 	"github.com/synnaxlabs/x/binary"
 	"github.com/synnaxlabs/x/errors"
+	"github.com/synnaxlabs/x/graph"
 	"github.com/synnaxlabs/x/kv"
 	"go.uber.org/zap"
 )
@@ -262,16 +263,9 @@ func MigrationDepOpt[T any](ctx context.Context) (T, bool) {
 	return v, ok
 }
 
-// ErrCyclicDependency is returned when migrations form a dependency cycle.
-var ErrCyclicDependency = errors.New("cyclic dependency detected in migrations")
-
-// ErrMissingDependency is returned when a migration depends on a name that
-// does not exist in the migration list and has not already been applied.
-var ErrMissingDependency = errors.New("missing migration dependency")
-
 // topoSort filters out already-applied migrations, then produces a valid
-// execution order using Kahn's algorithm. Dependencies that are already applied
-// are considered satisfied and do not need to appear in the pending set.
+// execution order. Dependencies that are already applied are considered
+// satisfied and do not need to appear in the pending set.
 func topoSort(migrations []Migration, applied map[string]bool) ([]Migration, error) {
 	byName := make(map[string]Migration, len(migrations))
 	for _, m := range migrations {
@@ -299,59 +293,28 @@ func topoSort(migrations []Migration, applied map[string]bool) ([]Migration, err
 		return pending, nil
 	}
 
-	pendingSet := make(map[string]bool, len(pending))
-	for _, m := range pending {
-		pendingSet[m.Name()] = true
-	}
-
-	inDegree := make(map[string]int, len(pending))
-	dependents := make(map[string][]string, len(pending))
+	adj := make(map[string][]string, len(pending))
 	for _, m := range pending {
 		name := m.Name()
-		if _, exists := inDegree[name]; !exists {
-			inDegree[name] = 0
-		}
+		adj[name] = nil
 		if dd, ok := m.(DependencyDeclarer); ok {
 			for _, dep := range dd.Dependencies() {
 				if applied[dep] {
 					continue
 				}
-				if !pendingSet[dep] {
-					if _, known := byName[dep]; !known {
-						return nil, fmt.Errorf(
-							"%w: migration %q depends on %q which does not exist",
-							ErrMissingDependency, name, dep,
-						)
-					}
-				}
-				inDegree[name]++
-				dependents[dep] = append(dependents[dep], name)
+				adj[name] = append(adj[name], dep)
 			}
 		}
 	}
 
-	var queue []string
-	for _, m := range pending {
-		if inDegree[m.Name()] == 0 {
-			queue = append(queue, m.Name())
-		}
-	}
-
-	var sorted []Migration
-	for len(queue) > 0 {
-		name := queue[0]
-		queue = queue[1:]
-		sorted = append(sorted, byName[name])
-		for _, dep := range dependents[name] {
-			inDegree[dep]--
-			if inDegree[dep] == 0 {
-				queue = append(queue, dep)
-			}
-		}
+	order, err := graph.TopoSort(adj)
+	if err != nil {
+		return nil, err
 	}
 
-	if len(sorted) != len(pending) {
-		return nil, fmt.Errorf("%w: not all migrations could be ordered", ErrCyclicDependency)
+	sorted := make([]Migration, len(order))
+	for i, name := range order {
+		sorted[i] = byName[name]
 	}
 	return sorted, nil
 }

x/go/gorp/migrate_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/synnaxlabs/x/binary"
 	"github.com/synnaxlabs/x/errors"
 	"github.com/synnaxlabs/x/gorp"
+	"github.com/synnaxlabs/x/graph"
 	"github.com/synnaxlabs/x/kv/memkv"
 	"github.com/synnaxlabs/x/query"
 	. "github.com/synnaxlabs/x/testutil"
@@ -817,7 +818,7 @@ var _ = Describe("Gorp", func() {
 					Migrations: []gorp.Migration{m1},
 				})
 				Expect(err).To(HaveOccurred())
-				Expect(err).To(MatchError(ContainSubstring("missing migration dependency")))
+				Expect(err).To(MatchError(graph.ErrMissingDependency))
 				Expect(err).To(MatchError(ContainSubstring("nonexistent")))
 			})
 

x/go/graph/graph.go

@@ -13,6 +13,7 @@ import (
 	"cmp"
 	"slices"
 
+	"github.com/synnaxlabs/x/errors"
 	"github.com/synnaxlabs/x/set"
 )
 
@@ -83,3 +84,70 @@ func TarjanSCC[T cmp.Ordered](adj map[T][]T) [][]T {
 	}
 	return sccs
 }
+
+// ErrCyclicDependency is returned by TopoSort when the graph contains a cycle.
+var ErrCyclicDependency = errors.New("cyclic dependency detected")
+
+// ErrMissingDependency is returned by TopoSort when an edge references a node
+// that does not exist in the graph.
+var ErrMissingDependency = errors.New("missing dependency")
+
+// TopoSort returns a topological ordering of the directed acyclic graph
+// represented by the adjacency list adj (where adj[a] = [b, c] means a depends
+// on b and c, i.e. b and c must come before a). Returns ErrCyclicDependency if
+// the graph contains a cycle, or ErrMissingDependency if an edge references a
+// node not present as a key in adj. The output is deterministic: nodes at the
+// same topological level are sorted by their natural ordering.
+func TopoSort[T cmp.Ordered](adj map[T][]T) ([]T, error) {
+	for node, deps := range adj {
+		for _, dep := range deps {
+			if _, exists := adj[dep]; !exists {
+				return nil, errors.Wrapf(
+					ErrMissingDependency,
+					"%v depends on %v which does not exist",
+					node, dep,
+				)
+			}
+		}
+	}
+
+	inDegree := make(map[T]int, len(adj))
+	dependents := make(map[T][]T, len(adj))
+	for node := range adj {
+		if _, exists := inDegree[node]; !exists {
+			inDegree[node] = 0
+		}
+		for _, dep := range adj[node] {
+			inDegree[node]++
+			dependents[dep] = append(dependents[dep], node)
+		}
+	}
+
+	var queue []T
+	for node, deg := range inDegree {
+		if deg == 0 {
+			queue = append(queue, node)
+		}
+	}
+	slices.Sort(queue)
+
+	var sorted []T
+	for len(queue) > 0 {
+		node := queue[0]
+		queue = queue[1:]
+		sorted = append(sorted, node)
+		next := dependents[node]
+		slices.Sort(next)
+		for _, dep := range next {
+			inDegree[dep]--
+			if inDegree[dep] == 0 {
+				queue = append(queue, dep)
+			}
+		}
+	}
+
+	if len(sorted) != len(adj) {
+		return nil, errors.Wrap(ErrCyclicDependency, "not all nodes could be ordered")
+	}
+	return sorted, nil
+}