Repro + analysis: scheduler inconsistent-graph-state on edge-merge + job cancel (draft)

solver/inconsistent_graph_state_error_tracker.go

@@ -5,6 +5,7 @@ package solver
 import (
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	digest "github.com/opencontainers/go-digest"
@@ -19,6 +20,7 @@ type dgstTrackerItem struct {
 }
 
 type dgstTracker struct {
+	mu      sync.Mutex
 	head    int
 	records []dgstTrackerItem
 }
@@ -31,6 +33,8 @@ func newDgstTracker() *dgstTracker {
 }
 
 func (d *dgstTracker) add(dgst digest.Digest, action string) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
 	d.head++
 	if d.head >= len(d.records) {
 		d.head = 0
@@ -40,20 +44,64 @@ func (d *dgstTracker) add(dgst digest.Digest, action string) {
 	d.records[d.head].seen = time.Now()
 }
 
-func (d *dgstTracker) String() string {
-	var sb strings.Builder
-
+// eachNewestFirst walks the ring from newest to oldest, calling fn for each
+// populated record. Caller must hold d.mu.
+func (d *dgstTracker) eachNewestFirst(fn func(dgstTrackerItem) bool) {
 	for i := d.head; i >= 0; i-- {
 		if d.records[i].seen.IsZero() {
-			break
+			return
+		}
+		if !fn(d.records[i]) {
+			return
 		}
-		sb.WriteString(fmt.Sprintf("%s %s %s; ", d.records[i].dgst, d.records[i].action, d.records[i].seen))
 	}
 	for i := len(d.records) - 1; i > d.head; i-- {
 		if d.records[i].seen.IsZero() {
-			break
+			return
+		}
+		if !fn(d.records[i]) {
+			return
 		}
-		sb.WriteString(fmt.Sprintf("%s %s %s; ", d.records[i].dgst, d.records[i].action, d.records[i].seen))
+	}
+}
+
+func (d *dgstTracker) String() string {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	var sb strings.Builder
+	d.eachNewestFirst(func(it dgstTrackerItem) bool {
+		sb.WriteString(fmt.Sprintf("%s %s %s; ", it.dgst, it.action, it.seen))
+		return true
+	})
+	return sb.String()
+}
+
+// historyFor returns up to max recorded actions for a single digest, newest
+// first. Unlike String() it is bounded and digest-scoped, so it stays well
+// under log-ingestion truncation limits and preserves the ordering that
+// matters for diagnosing "inconsistent graph state" (e.g. a delete preceding
+// a get-edge-not-found for the same digest).
+func (d *dgstTracker) historyFor(dgst digest.Digest, max int) string {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	var sb strings.Builder
+	shown, total := 0, 0
+	d.eachNewestFirst(func(it dgstTrackerItem) bool {
+		if it.dgst != dgst {
+			return true
+		}
+		total++
+		if shown < max {
+			sb.WriteString(fmt.Sprintf("%s %s; ", it.action, it.seen))
+			shown++
+		}
+		return true
+	})
+	if total > shown {
+		sb.WriteString(fmt.Sprintf("(+%d older)", total-shown))
+	}
+	if total == 0 {
+		return "(no prior records for digest)"
 	}
 	return sb.String()
 }

solver/internal/pipe/pipe.go

@@ -153,7 +153,12 @@ func (pw *sender[Payload, Value]) Finalize(v Value, err error) {
 	pw.status.Value = v
 	pw.status.Err = err
 	pw.status.Completed = true
-	if errors.Is(err, context.Canceled) && pw.req.Canceled {
+	// req is written by setRequest (via Receiver.Cancel) under mu, possibly
+	// concurrently with this Finalize from the request function goroutine.
+	pw.mu.Lock()
+	reqCanceled := pw.req.Canceled
+	pw.mu.Unlock()
+	if errors.Is(err, context.Canceled) && reqCanceled {
 		pw.status.Canceled = true
 	}
 	pw.sendChannel.Send(pw.status)

solver/scheduler.go

@@ -364,7 +364,18 @@ func (pf *pipeFactory) NewInputRequest(ee Edge, req *edgeRequest) pipeReceiver {
 	target := pf.s.ef.getEdge(ee)
 	if target == nil {
 		dgst := ee.Vertex.Digest()
-		bklog.G(context.TODO()).Errorf("failed to get edge dgst=%s name=%s desiredState=%s; actives history: %s", dgst, ee.Vertex.Name(), req.desiredState, dgstTrackerInst.String()) // earthly-specific
+		// earthly-specific: emit a discrete, digest-scoped record rather than
+		// the full tracker dump. The whole-ring String() is tens of KB and gets
+		// truncated at log ingestion, dropping exactly the ordering evidence
+		// (was this digest added/deleted before this failed lookup?) needed to
+		// diagnose the failure. historyFor stays small and keeps that ordering.
+		bklog.G(context.TODO()).
+			WithField("edge_vertex_name", ee.Vertex.Name()).
+			WithField("edge_vertex_digest", dgst).
+			WithField("edge_index", ee.Index).
+			WithField("desired_state", req.desiredState).
+			WithField("digest_history", dgstTrackerInst.historyFor(dgst, 32)).
+			Error("failed to get edge: inconsistent graph state")
 		debugSchedulerInconsistentGraphState(ee)
 		return pf.NewFuncRequest(func(_ context.Context) (any, error) {
 			return nil, errdefs.Internal(errors.Errorf("failed to get edge: inconsistent graph state in edge %s %s %d", ee.Vertex.Name(), ee.Vertex.Digest(), ee.Index))

solver/scheduler_mergecancel_test.go

@@ -0,0 +1,131 @@
+package solver
+
+import (
+	"context"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// TestSchedulerMergeCancelInconsistentGraphState reproduces the long-standing
+// "inconsistent graph state" scheduler failure (moby/buildkit#4733, #2303;
+// EarthBuild/earthbuild#768).
+//
+// Two jobs build distinct vertices that share cache-key seeds, so their edges
+// MUST edge-merge. A shared cache-map barrier forces their key computation to
+// overlap (guaranteeing the merge races job teardown). One job is then
+// cancelled at the merge point; its edge state is deleted from the actives map
+// while the surviving job's merged edge still references it, so the surviving
+// job requests a deleted edge -> getEdge()==nil -> "inconsistent graph state".
+//
+// Job lifecycle matches llbsolver.Solve: Discard is deferred and therefore runs
+// only after Build returns (the realistic sequencing).
+//
+// Skipped by default because on current code it FAILS (that is the point): it
+// documents an open bug. Set BUILDKIT_TEST_SCHEDULER_MERGE_CANCEL=1 to run.
+// Reproduces in roughly 2 of 3 runs within the iteration budget below.
+func TestSchedulerMergeCancelInconsistentGraphState(t *testing.T) {
+	if os.Getenv("BUILDKIT_TEST_SCHEDULER_MERGE_CANCEL") == "" {
+		t.Skip("reproduces moby/buildkit#4733; set BUILDKIT_TEST_SCHEDULER_MERGE_CANCEL=1 to run")
+	}
+
+	const iters = 2000
+	for i := 0; i < iters; i++ {
+		if msg := runMergeCancelIter(); msg != "" {
+			t.Fatalf("iter %d: surviving job observed scheduler corruption: %s", i, msg)
+		}
+	}
+}
+
+func runMergeCancelIter() string {
+	s := NewSolver(SolverOpt{ResolveOpFunc: testOpResolver})
+	defer s.Close()
+
+	// Barrier: both roots block in CacheMap until both arrive, so their key
+	// computation (and therefore the edge merge) overlaps.
+	var arrived int32
+	gate := make(chan struct{})
+	barrier := func(ctx context.Context) error {
+		if atomic.AddInt32(&arrived, 1) == 2 {
+			close(gate)
+		}
+		select {
+		case <-gate:
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+
+	mkGraph := func(tag string) Edge {
+		dep := vtxConst(1, vtxOpt{name: "dep-" + tag, cacheKeySeed: "shared-dep"})
+		root := vtxAdd(2, vtxOpt{
+			name:         "root-" + tag,
+			cacheKeySeed: "shared-root",
+			cachePreFunc: barrier,
+			execDelay:    2 * time.Millisecond, // keep the surviving edge active during the cancel
+			inputs:       []Edge{{Vertex: dep}},
+		})
+		return Edge{Vertex: root}
+	}
+
+	jA, err := s.NewJob("A")
+	if err != nil {
+		return ""
+	}
+	jB, err := s.NewJob("B")
+	if err != nil {
+		return ""
+	}
+
+	ctxA, cancelA := context.WithCancel(context.Background())
+	ctxB, cancelB := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancelB()
+
+	var got atomic.Value
+	record := func(e error) {
+		if e == nil {
+			return
+		}
+		m := e.Error()
+		if strings.Contains(m, "inconsistent graph state") ||
+			strings.Contains(m, "return leaving outgoing open") ||
+			strings.Contains(m, "return leaving incoming open") {
+			got.Store(m)
+		}
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		defer func() { _ = jA.Discard() }()
+		_, e := jA.Build(ctxA, mkGraph("A"))
+		record(e)
+	}()
+	go func() {
+		defer wg.Done()
+		defer func() { _ = jB.Discard() }()
+		_, e := jB.Build(ctxB, mkGraph("B"))
+		record(e)
+	}()
+
+	// Cancel A the moment both edges have reached the merge point; A's deferred
+	// Discard then runs after its Build unwinds.
+	go func() {
+		select {
+		case <-gate:
+		case <-time.After(3 * time.Second):
+		}
+		cancelA()
+	}()
+
+	wg.Wait()
+	if v := got.Load(); v != nil {
+		return v.(string)
+	}
+	return ""
+}