fix error

Author: LexLuthr

harmony/harmonydb/sql/20240823-ipni.sql

@@ -105,7 +105,7 @@ CREATE TABLE ipni_task (
     reg_seal_proof INT NOT NULL,
     sector_offset BIGINT,
 
-    context_id BYTEA NOT NULL PRIMARY KEY,
+    context_id BYTEA NOT NULL,
     is_rm BOOLEAN NOT NULL,
 
     provider TEXT NOT NULL,

lib/paths/local_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/filecoin-project/curio/harmony/harmonydb"
-	storiface "github.com/filecoin-project/curio/lib/storiface"
+	"github.com/filecoin-project/curio/lib/storiface"
 
 	"github.com/filecoin-project/lotus/storage/sealer/fsutil"
 )
@@ -82,7 +82,9 @@ func TestLocalStorage(t *testing.T) {
 		root: root,
 	}
 
-	db, err := harmonydb.NewFromConfigWithITestID(t, "testlocalstorage")
+	sharedITestID := harmonydb.ITestNewID()
+
+	db, err := harmonydb.NewFromConfigWithITestID(t, sharedITestID)
 	require.NoError(t, err)
 
 	index := NewDBIndex(nil, db)

lib/paths/remote_test.go

@@ -25,7 +25,7 @@ import (
 	"github.com/filecoin-project/curio/lib/partialfile"
 	"github.com/filecoin-project/curio/lib/paths"
 	"github.com/filecoin-project/curio/lib/paths/mocks"
-	storiface "github.com/filecoin-project/curio/lib/storiface"
+	"github.com/filecoin-project/curio/lib/storiface"
 )
 
 const metaFile = "sectorstore.json"
@@ -59,7 +59,9 @@ func createTestStorage(t *testing.T, p string, seal bool, att ...*paths.Local) s
 func TestMoveShared(t *testing.T) {
 	logging.SetAllLoggers(logging.LevelDebug)
 
-	db, err := harmonydb.NewFromConfigWithITestID(t, "testlocalstorage")
+	sharedITestID := harmonydb.ITestNewID()
+
+	db, err := harmonydb.NewFromConfigWithITestID(t, sharedITestID)
 	require.NoError(t, err)
 
 	index := paths.NewDBIndex(nil, db)

tasks/indexing/task_indexing.go

@@ -214,13 +214,25 @@ func (i *IndexingTask) recordCompletion(ctx context.Context, task itask, taskID
 		return xerrors.Errorf("failed to update piece metadata and piece deal for deal %s: %w", task.UUID, err)
 	}
 
-	n, err := i.db.Exec(ctx, `UPDATE market_mk12_deal_pipeline SET indexed = TRUE, complete = TRUE WHERE uuid = $1`, task.UUID)
-	if err != nil {
-		return xerrors.Errorf("store indexing success: updating pipeline: %w", err)
-	}
-	if n != 1 {
-		return xerrors.Errorf("store indexing success: updated %d rows", n)
+	// If IPNI is disabled then mark deal as complete otherwise just mark as indexed
+	if i.cfg.Market.StorageMarketConfig.IPNI.Disable {
+		n, err := i.db.Exec(ctx, `UPDATE market_mk12_deal_pipeline SET indexed = TRUE, complete = TRUE WHERE uuid = $1`, task.UUID)
+		if err != nil {
+			return xerrors.Errorf("store indexing success: updating pipeline: %w", err)
+		}
+		if n != 1 {
+			return xerrors.Errorf("store indexing success: updated %d rows", n)
+		}
+	} else {
+		n, err := i.db.Exec(ctx, `UPDATE market_mk12_deal_pipeline SET indexed = TRUE WHERE uuid = $1`, task.UUID)
+		if err != nil {
+			return xerrors.Errorf("store indexing success: updating pipeline: %w", err)
+		}
+		if n != 1 {
+			return xerrors.Errorf("store indexing success: updated %d rows", n)
+		}
 	}
+
 	return nil
 }
 

tasks/indexing/task_ipni.go

@@ -317,6 +317,8 @@ func (I *IPNITask) TypeDetails() harmonytask.TaskTypeDetails {
 
 func (I *IPNITask) schedule(ctx context.Context, taskFunc harmonytask.AddTaskFunc) error {
 	// If IPNI is disabled then don't schedule any tasks
+	// Deals should already be marked as complete by task_indexing
+	// This check is to cover any edge case
 	if I.cfg.Market.StorageMarketConfig.IPNI.Disable {
 		return nil
 	}