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/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) 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)) 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 "" +}