fix(sealing): don't let a cancelled task block scheduling of other tasks

CHANGELOG.md

@@ -34,6 +34,7 @@ This is the **recommended** Lotus release for the nv29 (Solstice) network upgrad
 
 ## 🐛 Bug Fixes
 
+- fix(sealing): a cancelled task in the scheduler queue no longer stops other tasks from being scheduled. ([filecoin-project/lotus#13869](https://github.com/filecoin-project/lotus/pull/13869))
 - fix(api): check `/health/readyz` immediately on startup instead of waiting for the first one-minute polling interval. ([filecoin-project/lotus#13796](https://github.com/filecoin-project/lotus/pull/13796))
 - fix(rpc): allow `ChainPutObj` to store blocks over JSON-RPC. ([filecoin-project/lotus#13868](https://github.com/filecoin-project/lotus/pull/13868))
 - fix(chain): p2p ingress hardening. Tipsets from peers are limited to 34 distinct blocks, and ChainExchange serves messages for at most 8 tipsets per request. ([filecoin-project/lotus#13909](https://github.com/filecoin-project/lotus/pull/13909))

lib/lazy/getonce.go

@@ -2,6 +2,7 @@ package lazy
 
 import (
 	"context"
+	"errors"
 	"sync"
 )
 
@@ -30,7 +31,8 @@ func (l *Lazy[T]) Val() (T, error) {
 type LazyCtx[T any] struct {
 	Get func(context.Context) (T, error)
 
-	once sync.Once
+	lk   sync.Mutex
+	done bool
 
 	val T
 	err error
@@ -42,9 +44,18 @@ func MakeLazyCtx[T any](get func(ctx context.Context) (T, error)) *LazyCtx[T] {
 	}
 }
 
+// Val calls Get once and caches the result. A failed call whose context was
+// cancelled is not cached, so the next caller with a live context calls Get again.
 func (l *LazyCtx[T]) Val(ctx context.Context) (T, error) {
-	l.once.Do(func() {
-		l.val, l.err = l.Get(ctx)
-	})
+	l.lk.Lock()
+	defer l.lk.Unlock()
+
+	if !l.done {
+		val, err := l.Get(ctx)
+		if err != nil && errors.Is(ctx.Err(), context.Canceled) {
+			return val, err
+		}
+		l.val, l.err, l.done = val, err, true
+	}
 	return l.val, l.err
 }

lib/lazy/getonce_test.go

@@ -0,0 +1,82 @@
+package lazy
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLazyCtxRetriesAfterCancelledCall(t *testing.T) {
+	calls := 0
+	l := MakeLazyCtx(func(ctx context.Context) (int, error) {
+		calls++
+		if err := ctx.Err(); err != nil {
+			return 0, err
+		}
+		return 42, nil
+	})
+
+	cctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, err := l.Val(cctx)
+	require.ErrorIs(t, err, context.Canceled)
+
+	v, err := l.Val(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, 42, v)
+
+	v, err = l.Val(cctx)
+	require.NoError(t, err)
+	require.Equal(t, 42, v)
+	require.Equal(t, 2, calls)
+}
+
+func TestLazyCtxCachesOtherErrors(t *testing.T) {
+	errUnreachable := errors.New("worker unreachable")
+
+	tests := []struct {
+		name string
+		ctx  func() (context.Context, context.CancelFunc)
+		get  func(ctx context.Context) (int, error)
+		want error
+	}{
+		{
+			name: "error with a live context",
+			ctx:  func() (context.Context, context.CancelFunc) { return context.WithCancel(context.Background()) },
+			get:  func(context.Context) (int, error) { return 0, errUnreachable },
+			want: errUnreachable,
+		},
+		{
+			name: "deadline exceeded",
+			ctx: func() (context.Context, context.CancelFunc) {
+				return context.WithDeadline(context.Background(), time.Now())
+			},
+			get:  func(ctx context.Context) (int, error) { return 0, ctx.Err() },
+			want: context.DeadlineExceeded,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			calls := 0
+			l := MakeLazyCtx(func(ctx context.Context) (int, error) {
+				calls++
+				return tt.get(ctx)
+			})
+
+			ctx, cancel := tt.ctx()
+			defer cancel()
+
+			_, err := l.Val(ctx)
+			require.ErrorIs(t, err, tt.want)
+
+			_, err = l.Val(context.Background())
+			require.ErrorIs(t, err, tt.want)
+			require.Equal(t, 1, calls)
+		})
+	}
+}

storage/sealer/request_queue.go

@@ -37,6 +37,27 @@ func (q *RequestQueue) Push(x *WorkerRequest) {
 	sort.Sort(q)
 }
 
+// RemoveCancelled removes the requests whose context is done, keeping the
+// order of the rest, and returns how many were removed.
+func (q *RequestQueue) RemoveCancelled() int {
+	old := *q
+	n := 0
+	for _, req := range old {
+		if req.Ctx.Err() != nil {
+			req.index = -1
+			continue
+		}
+		req.index = n
+		old[n] = req
+		n++
+	}
+	for i := n; i < len(old); i++ {
+		old[i] = nil
+	}
+	*q = old[:n]
+	return len(old) - n
+}
+
 func (q *RequestQueue) Remove(i int) *WorkerRequest {
 	old := *q
 	n := len(old)

storage/sealer/sched.go

@@ -394,6 +394,11 @@ func (sh *Scheduler) trySched() {
 	done := metrics.Timer(sh.mctx, metrics.SchedAssignerCycleDuration)
 	defer done()
 
+	// Schedule already returned ctx.Err() to the callers of cancelled requests.
+	if n := sh.SchedQueue.RemoveCancelled(); n > 0 {
+		log.Debugw("dropped cancelled scheduling requests", "count", n)
+	}
+
 	sh.assigner.TrySched(sh)
 }
 

storage/sealer/sched_cancelled_test.go

@@ -0,0 +1,95 @@
+package sealer
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+
+	"github.com/filecoin-project/go-state-types/abi"
+
+	"github.com/filecoin-project/lotus/storage/sealer/sealtasks"
+	"github.com/filecoin-project/lotus/storage/sealer/storiface"
+)
+
+// ctxTaskTypesWorker fails TaskTypes on a done context, like a worker behind
+// an RPC client does.
+type ctxTaskTypesWorker struct {
+	*schedTestWorker
+}
+
+func (w *ctxTaskTypesWorker) TaskTypes(ctx context.Context) (map[sealtasks.TaskType]struct{}, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	return w.schedTestWorker.TaskTypes(ctx)
+}
+
+func testRequest(ctx context.Context, sector abi.SectorNumber) *WorkerRequest {
+	return &WorkerRequest{
+		Sector: storiface.SectorRef{
+			ID:        abi.SectorID{Miner: 1000, Number: sector},
+			ProofType: abi.RegisteredSealProof_StackedDrg2KiBV1_1,
+		},
+		TaskType: sealtasks.TTPreCommit1,
+		Sel:      newTaskSelector(),
+		Ctx:      ctx,
+	}
+}
+
+func TestSchedCancelledRequestDoesNotBlockOthers(t *testing.T) {
+	sched, err := newScheduler(context.Background(), "")
+	require.NoError(t, err)
+
+	w := &ctxTaskTypesWorker{&schedTestWorker{
+		name:      "fred",
+		taskTypes: map[sealtasks.TaskType]struct{}{sealtasks.TTPreCommit1: {}},
+		session:   uuid.New(),
+		resources: decentWorkerResources,
+	}}
+	wh, err := newWorkerHandle(context.Background(), w)
+	require.NoError(t, err)
+	wid := storiface.WorkerID(w.session)
+	sched.Workers[wid] = wh
+
+	// With one open window the assigner checks requests one at a time, in
+	// queue order, so the cancelled request is checked first.
+	done := make(chan *SchedWindow, 1)
+	sched.OpenWindows = []*SchedWindowRequest{{Worker: wid, Done: done}}
+
+	cctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	sched.SchedQueue.Push(testRequest(cctx, 1))
+	live := testRequest(context.Background(), 2)
+	sched.SchedQueue.Push(live)
+
+	sched.trySched()
+
+	require.Zero(t, sched.SchedQueue.Len())
+	select {
+	case wnd := <-done:
+		require.Equal(t, []*WorkerRequest{live}, wnd.Todo)
+	default:
+		t.Fatal("expected the live request to be scheduled")
+	}
+}
+
+func TestRequestQueueRemoveCancelled(t *testing.T) {
+	cctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	rq := &RequestQueue{}
+	for i, ctx := range []context.Context{context.Background(), cctx, context.Background(), cctx} {
+		rq.Push(testRequest(ctx, abi.SectorNumber(i)))
+	}
+
+	require.Equal(t, 2, rq.RemoveCancelled())
+	require.Equal(t, 2, rq.Len())
+	for i, req := range *rq {
+		require.NoError(t, req.Ctx.Err())
+		require.Equal(t, i, req.index)
+	}
+	require.Equal(t, abi.SectorNumber(0), (*rq)[0].Sector.ID.Number)
+	require.Equal(t, abi.SectorNumber(2), (*rq)[1].Sector.ID.Number)
+}