fix: itests: Fix flaky curio itest (#12037)

* fix: itests: Fix flaky curio itest

* fix curio test with ffiselect

* try bigger instance for curio itest

* curio itest: reduce block time

.github/workflows/test.yml

@@ -63,13 +63,13 @@ jobs:
               "itest-deals_concurrent": ["self-hosted", "linux", "x64", "4xlarge"],
               "itest-sector_pledge": ["self-hosted", "linux", "x64", "4xlarge"],
               "itest-worker": ["self-hosted", "linux", "x64", "4xlarge"],
+              "itest-curio": ["self-hosted", "linux", "x64", "4xlarge"],
 
               "itest-gateway": ["self-hosted", "linux", "x64", "2xlarge"],
               "itest-sector_import_full": ["self-hosted", "linux", "x64", "2xlarge"],
               "itest-sector_import_simple": ["self-hosted", "linux", "x64", "2xlarge"],
               "itest-wdpost": ["self-hosted", "linux", "x64", "2xlarge"],
               "unit-storage": ["self-hosted", "linux", "x64", "2xlarge"],
-              "itest-curio": ["self-hosted", "linux", "x64", "2xlarge"],
 
               "itest-batch_deal": ["self-hosted", "linux", "x64", "xlarge"],
               "itest-cli": ["self-hosted", "linux", "x64", "xlarge"],

curiosrc/seal/task_treerc.go

@@ -158,8 +158,10 @@ func (t *TreeRCTask) TypeDetails() harmonytask.TaskTypeDetails {
 		ssize = abi.SectorSize(2 << 20)
 	}
 	gpu := 1.0
+	ram := uint64(8 << 30)
 	if isDevnet {
 		gpu = 0
+		ram = 512 << 20
 	}
 
 	return harmonytask.TaskTypeDetails{
@@ -168,7 +170,7 @@ func (t *TreeRCTask) TypeDetails() harmonytask.TaskTypeDetails {
 		Cost: resources.Resources{
 			Cpu:     1,
 			Gpu:     gpu,
-			Ram:     8 << 30,
+			Ram:     ram,
 			Storage: t.sc.Storage(t.taskToSector, storiface.FTSealed, storiface.FTCache, ssize, storiface.PathSealing, paths.MinFreeStoragePercentage),
 		},
 		MaxFailures: 3,

itests/curio_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/docker/go-units"
 	"github.com/gbrlsnchs/jwt/v3"
 	"github.com/google/uuid"
+	logging "github.com/ipfs/go-log/v2"
 	manet "github.com/multiformats/go-multiaddr/net"
 	"github.com/stretchr/testify/require"
 	"github.com/urfave/cli/v2"
@@ -35,6 +36,7 @@ import (
 	"github.com/filecoin-project/lotus/curiosrc/market/lmrpc"
 	"github.com/filecoin-project/lotus/curiosrc/seal"
 	"github.com/filecoin-project/lotus/itests/kit"
+	"github.com/filecoin-project/lotus/lib/ffiselect"
 	"github.com/filecoin-project/lotus/lib/harmony/harmonydb"
 	"github.com/filecoin-project/lotus/node"
 	"github.com/filecoin-project/lotus/node/config"
@@ -54,7 +56,7 @@ func TestCurioNewActor(t *testing.T) {
 
 	esemble.Start()
 	blockTime := 100 * time.Millisecond
-	esemble.BeginMining(blockTime)
+	esemble.BeginMiningMustPost(blockTime)
 
 	db := miner.BaseAPI.(*impl.StorageMinerAPI).HarmonyDB
 
@@ -98,12 +100,16 @@ func TestCurioHappyPath(t *testing.T) {
 	full, miner, esemble := kit.EnsembleMinimal(t,
 		kit.LatestActorsAt(-1),
 		kit.WithSectorIndexDB(),
+		kit.PresealSectors(32),
+		kit.ThroughRPC(),
 	)
 
 	esemble.Start()
 	blockTime := 100 * time.Millisecond
 	esemble.BeginMining(blockTime)
 
+	full.WaitTillChain(ctx, kit.HeightAtLeast(15))
+
 	err := miner.LogSetLevel(ctx, "*", "ERROR")
 	require.NoError(t, err)
 
@@ -182,6 +188,7 @@ func TestCurioHappyPath(t *testing.T) {
 		}
 		return true, nil
 	})
+	require.NoError(t, err)
 	require.Len(t, num, 1)
 	// TODO: add DDO deal, f05 deal 2 MiB each in the sector
 
@@ -191,13 +198,17 @@ func TestCurioHappyPath(t *testing.T) {
 	}
 
 	require.Eventuallyf(t, func() bool {
+		h, err := full.ChainHead(ctx)
+		require.NoError(t, err)
+		t.Logf("head: %d", h.Height())
+
 		err = db.Select(ctx, &sectorParamsArr, `
 		SELECT sp_id, sector_number
 		FROM sectors_sdr_pipeline
 		WHERE after_commit_msg_success = True`)
 		require.NoError(t, err)
 		return len(sectorParamsArr) == 1
-	}, 5*time.Minute, 1*time.Second, "sector did not finish sealing in 5 minutes")
+	}, 10*time.Minute, 1*time.Second, "sector did not finish sealing in 5 minutes")
 
 	require.Equal(t, sectorParamsArr[0].SectorNumber, int64(0))
 	require.Equal(t, sectorParamsArr[0].SpID, int64(mid))
@@ -278,7 +289,7 @@ func createCliContext(dir string) (*cli.Context, error) {
 	sflag := fmt.Sprintf("--journal=%s", storage)
 
 	// Parse the flags with test values
-	err := set.Parse([]string{"--listen=0.0.0.0:12345", "--nosync", "--manage-fdlimit", sflag, cflag, "--layers=seal", "--layers=post"})
+	err := set.Parse([]string{"--listen=0.0.0.0:12345", "--nosync", "--manage-fdlimit", sflag, cflag, "--layers=seal"})
 	if err != nil {
 		return nil, xerrors.Errorf("Error setting flag: %s\n", err)
 	}
@@ -292,6 +303,8 @@ func createCliContext(dir string) (*cli.Context, error) {
 }
 
 func ConstructCurioTest(ctx context.Context, t *testing.T, dir string, db *harmonydb.DB, full v1api.FullNode, maddr address.Address, cfg *config.CurioConfig) (api.Curio, func(), jsonrpc.ClientCloser, <-chan struct{}) {
+	ffiselect.IsTest = true
+
 	cctx, err := createCliContext(dir)
 	require.NoError(t, err)
 
@@ -380,7 +393,7 @@ func ConstructCurioTest(ctx context.Context, t *testing.T, dir string, db *harmo
 	err = capi.StorageAddLocal(ctx, dir)
 	require.NoError(t, err)
 
-	err = capi.LogSetLevel(ctx, "harmonytask", "DEBUG")
+	_ = logging.SetLogLevel("harmonytask", "DEBUG")
 
 	return capi, taskEngine.GracefullyTerminate, ccloser, finishCh
 }

lib/ffiselect/ffiselect.go

@@ -22,6 +22,7 @@ import (
 	"github.com/filecoin-project/lotus/lib/ffiselect/ffidirect"
 )
 
+var IsTest = false
 var IsCuda = build.IsOpencl != "1"
 
 // Get all devices from ffi
@@ -54,6 +55,10 @@ func subStrInSet(set []string, sub string) bool {
 }
 
 func call(logctx []any, fn string, args ...interface{}) ([]interface{}, error) {
+	if IsTest {
+		return callTest(logctx, fn, args...)
+	}
+
 	// get dOrdinal
 	dOrdinal := <-ch
 	defer func() {

lib/ffiselect/testffi.go

@@ -0,0 +1,27 @@
+package ffiselect
+
+import (
+	"reflect"
+
+	"github.com/samber/lo"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/lib/ffiselect/ffidirect"
+)
+
+func callTest(logctx []any, fn string, rawargs ...interface{}) ([]interface{}, error) {
+	args := lo.Map(rawargs, func(arg any, i int) reflect.Value {
+		return reflect.ValueOf(arg)
+	})
+
+	resAry := reflect.ValueOf(ffidirect.FFI{}).MethodByName(fn).Call(args)
+	res := lo.Map(resAry, func(res reflect.Value, i int) any {
+		return res.Interface()
+	})
+
+	if res[len(res)-1].(ffidirect.ErrorString) != "" {
+		return nil, xerrors.Errorf("callTest error: %s", res[len(res)-1].(ffidirect.ErrorString))
+	}
+
+	return res, nil
+}

lib/harmony/harmonydb/harmonydb.go

@@ -279,6 +279,8 @@ func (db *DB) upgrade() error {
 			return err
 		}
 
+		logger.Infow("Upgrading", "file", name, "size", len(file))
+
 		for _, s := range parseSQLStatements(string(file)) { // Implement the changes.
 			if len(strings.TrimSpace(s)) == 0 {
 				continue