From 2154829eaf4d48ccb9d612bd8911a790691c9cd0 Mon Sep 17 00:00:00 2001 From: Kieran Mann Date: Wed, 19 Aug 2026 08:37:13 -0700 Subject: [PATCH 1/3] fix(solver): read req under lock in pipe Finalize to avoid data race sender.Finalize read pw.req.Canceled without holding pw.mu while receiver.Cancel -> setRequest writes pw.req under pw.mu. The request-function goroutine's Finalize can run concurrently with a Cancel, so the read races the write (confirmed by go test -race). Read the flag under the mutex. Co-Authored-By: Claude Opus 4.8 --- solver/internal/pipe/pipe.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/solver/internal/pipe/pipe.go b/solver/internal/pipe/pipe.go index 55309c2a9..027f20ed3 100644 --- a/solver/internal/pipe/pipe.go +++ b/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) From 81d3034af5070ec05c9c0f457925349df118e4ed Mon Sep 17 00:00:00 2001 From: Kieran Mann Date: Wed, 19 Aug 2026 08:37:13 -0700 Subject: [PATCH 2/3] test(solver): reproduce inconsistent-graph-state on edge-merge + job cancel Skip-guarded unit repro for the scheduler failure "failed to get edge: inconsistent graph state" (moby/buildkit#4733, #2303; EarthBuild/earthbuild#768). Two jobs build distinct vertices sharing cache-key seeds (forcing an edge merge); a shared cache-map barrier overlaps their key computation so the merge races job teardown. One job is cancelled at the merge point (Discard deferred, matching llbsolver.Solve), deleting its edge state from the actives map while the surviving job's merged edge still references it -> getEdge()==nil. Skipped by default (fails on current code by design); run with BUILDKIT_TEST_SCHEDULER_MERGE_CANCEL=1. Reproduces in ~2/3 runs within the iteration budget. Root cause: edgeIndex can hand back a merge target whose state was concurrently deleted (edge.release cleans the index only once releaserCount hits 0) -- the "merge to inactive state" scenario #4887 targeted, still reachable under the parallel-cancel window. Co-Authored-By: Claude Opus 4.8 --- solver/scheduler_mergecancel_test.go | 131 +++++++++++++++++++++++++++ 1 file changed, 131 insertions(+) create mode 100644 solver/scheduler_mergecancel_test.go diff --git a/solver/scheduler_mergecancel_test.go b/solver/scheduler_mergecancel_test.go new file mode 100644 index 000000000..a6f4806f8 --- /dev/null +++ b/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 "" +} From 2c7e207136634cb7164b41276befa12c598f4933 Mon Sep 17 00:00:00 2001 From: Kieran Mann Date: Wed, 19 Aug 2026 08:58:08 -0700 Subject: [PATCH 3/3] fix(solver): emit digest-scoped, untruncated log at inconsistent-graph-state The inconsistent-graph-state error site logged the entire ~10k-entry tracker ring via dgstTracker.String() (tens of KB). Datadog/CI ingestion truncates these giant messages (~76KB), dropping the newest-first tail that carries the one thing needed to diagnose the failure: whether the failing digest was added/deleted before the failed lookup. Replace that dump with a discrete structured line (edge_vertex_name/digest/ index + desired_state + a bounded, digest-scoped history) so the delete-> get-edge-not-found ordering for the failing digest survives ingestion. Also add a mutex to dgstTracker: String()/add() previously raced (read via %+v/ String concurrent with ring writes). Refs EarthBuild/earthbuild#768, moby/buildkit#4733. Co-Authored-By: Claude Opus 4.8 --- .../inconsistent_graph_state_error_tracker.go | 62 ++++++++++++++++--- solver/scheduler.go | 13 +++- 2 files changed, 67 insertions(+), 8 deletions(-) diff --git a/solver/inconsistent_graph_state_error_tracker.go b/solver/inconsistent_graph_state_error_tracker.go index 8c8cd05a7..9c4512f62 100644 --- a/solver/inconsistent_graph_state_error_tracker.go +++ b/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() } diff --git a/solver/scheduler.go b/solver/scheduler.go index 70f8fbdfc..6b47e252c 100644 --- a/solver/scheduler.go +++ b/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))