remove indexing table, add defaults

Author: LexLuthr

.circleci/config.yml

@@ -144,7 +144,7 @@ jobs:
         description: Flags passed to go test.
       target:
         type: string
-        default: "./..."
+        default: "./... | grep -v itests"
         description: Import paths of packages to be tested.
       proofs-log-test:
         type: string
@@ -287,6 +287,5 @@ workflows:
           name: test-idxStore
           requires:
             - build
-          suite: idxStore
-          target: "./lib/indexing/indexstore/indexstore_test.go"
+          suite: test-all
           get-params: true

harmony/harmonydb/sql/20240730-market-migration.sql

@@ -1,14 +1,9 @@
 -- Table for Mk12 or Boost deals
 CREATE TABLE market_mk12_deals (
     uuid TEXT NOT NULL,
-<<<<<<< HEAD
     created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP AT TIME ZONE 'UTC',
-=======
     sp_id BIGINT NOT NULL,
 
-    created_at TIMESTAMPTZ NOT NULL DEFAULT TIMEZONE('UTC', NOW()),
-
->>>>>>> 48e953d (poller redesign)
     signed_proposal_cid TEXT NOT NULL,
     proposal_signature BYTEA NOT NULL,
     proposal jsonb NOT NULL,
@@ -41,33 +36,13 @@ CREATE TABLE market_mk12_deals (
     unique (signed_proposal_cid)
 );
 
--- Table for old lotus market deals. This is just for deal
--- which are still alive. It should not be used for any processing
-CREATE TABLE market_legacy_deals (
-    signed_proposal_cid TEXT,
-    proposal_signature BYTEA,
-    proposal jsonb,
-    piece_cid TEXT,
-    piece_size BIGINT,
-    offline BOOLEAN,
-    verified BOOLEAN,
-    sp_id BIGINT,
-    start_epoch BIGINT,
-    end_epoch BIGINT,
-    publish_cid TEXT,
-    fast_retrieval BOOLEAN,
-    chain_deal_id BIGINT,
-    created_at TIMESTAMPTZ,
-    sector_num BIGINT,
-
-    primary key (sp_id, piece_cid, signed_proposal_cid)
-);
-
 -- This table is used for storing piece metadata (piece indexing)
 CREATE TABLE market_piece_metadata (
     piece_cid TEXT NOT NULL PRIMARY KEY,
+
     version INT NOT NULL DEFAULT 2,
     created_at TIMESTAMPTZ  NOT NULL DEFAULT CURRENT_TIMESTAMP AT TIME ZONE 'UTC',
+
     indexed BOOLEAN NOT NULL DEFAULT FALSE,
     indexed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP AT TIME ZONE 'UTC',
 
@@ -157,13 +132,9 @@ CREATE TABLE market_mk12_deal_pipeline (
     started BOOLEAN DEFAULT FALSE,
 
     piece_cid TEXT NOT NULL,
-<<<<<<< HEAD
     piece_size BOOLEAN NOT NULL,
-=======
-    piece_size BIGINT NOT NULL,
-    file_size BIGINT DEFAULT NULL, -- raw piece size
+    raw_size BIGINT DEFAULT NULL,
 
->>>>>>> 48e953d (poller redesign)
     offline BOOLEAN NOT NULL,
 
     url TEXT DEFAULT NULL,
@@ -181,52 +152,94 @@ CREATE TABLE market_mk12_deal_pipeline (
     after_find_deal BOOLEAN DEFAULT FALSE,
 
     sector BIGINT,
+    reg_seal_proof INT NOT NULL,
     sector_offset BIGINT,
 
     sealed BOOLEAN DEFAULT FALSE,
+
+    should_index BOOLEAN DEFAULT FALSE,
+    indexing_created_at TIMESTAMPTZ,
+    indexing_task_id BIGINT DEFAULT NULL,
     indexed BOOLEAN DEFAULT FALSE,
 
+    complete BOOLEAN NOT NULL DEFAULT FALSE,
+
     constraint market_mk12_deal_pipeline_identity_key unique (uuid)
 );
 
+-- This function creates indexing task based from move_storage tasks
+CREATE OR REPLACE FUNCTION create_indexing_task(task_id BIGINT, sealing_table TEXT)
+RETURNS VOID AS $$
+DECLARE
+query TEXT;   -- Holds the dynamic SQL query
+    pms RECORD;   -- Holds each row returned by the query in the loop
+BEGIN
+    -- Construct the dynamic SQL query based on the sealing_table
+    IF sealing_table = 'sectors_sdr_pipeline' THEN
+        query := format(
+            'SELECT
+                dp.uuid,
+                ssp.reg_seal_proof
+            FROM
+                %I ssp
+            JOIN
+                market_mk12_deal_pipeline dp ON ssp.sp_id = dp.sp_id AND ssp.sector_num = dp.sector
+            WHERE
+                ssp.task_id_move_storage = $1', sealing_table);
+    ELSIF sealing_table = 'sectors_snap_pipeline' THEN
+        query := format(
+            'SELECT
+                dp.uuid,
+                (SELECT reg_seal_proof FROM sectors_meta WHERE sp_id = ssp.sp_id AND sector_num = ssp.sector_num) AS reg_seal_proof
+            FROM
+                %I ssp
+            JOIN
+                market_mk12_deal_pipeline dp ON ssp.sp_id = dp.sp_id AND ssp.sector_num = dp.sector
+            WHERE
+                ssp.task_id_move_storage = $1', sealing_table);
+ELSE
+        RAISE EXCEPTION 'Invalid sealing_table name: %', sealing_table;
+END IF;
+
+    -- Execute the dynamic SQL query with the task_id parameter
+FOR pms IN EXECUTE query USING task_id
+    LOOP
+        -- Update the market_mk12_deal_pipeline table with the reg_seal_proof and indexing_created_at values
+UPDATE market_mk12_deal_pipeline
+SET
+    reg_seal_proof = pms.reg_seal_proof,
+    indexing_created_at = NOW() AT TIME ZONE 'UTC'
+WHERE
+    uuid = pms.uuid;
+END LOOP;
+
+    -- If everything is successful, simply exit
+    RETURN;
+
+EXCEPTION
+    WHEN OTHERS THEN
+        -- Rollback the transaction and raise the exception for Go to catch
+        ROLLBACK;
+        RAISE EXCEPTION 'Failed to create indexing task: %', SQLERRM;
+END;
+$$ LANGUAGE plpgsql;
+
 -- This table can be used to track remote piece for offline deals
 -- The entries must be created by users
 CREATE TABLE market_offline_urls (
-    piece_cid TEXT NOT NULL,
+     uuid TEXT NOT NULL,
 
-    url TEXT NOT NULL,
-    headers jsonb NOT NULL DEFAULT '{}',
+     url TEXT NOT NULL,
+     headers jsonb NOT NULL DEFAULT '{}',
 
-    raw_size BIGINT NOT NULL,
+     raw_size BIGINT NOT NULL,
 
-    unique (piece_cid)
+     CONSTRAINT market_offline_urls_uuid_fk FOREIGN KEY (uuid)
+         REFERENCES market_mk12_deal_pipeline (uuid)
+         ON DELETE CASCADE,
+     CONSTRAINT market_offline_urls_uuid_unique UNIQUE (uuid)
 );
 
--- indexing tracker is separate from
-CREATE TABLE market_indexing_tasks (
-    uuid TEXT NOT NULL,
-
-    sp_id BIGINT NOT NULL,
-    sector_number BIGINT NOT NULL,
-    reg_seal_proof INT NOT NULL,
-
-    piece_offset BIGINT NOT NULL,
-    piece_size BIGINT NOT NULL,
-    raw_size BIGINT NOT NULL,
-    piece_cid TEXT NOT NULL,
-<<<<<<< HEAD
-    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP AT TIME ZONE 'UTC',
-=======
-
-    created_at TIMESTAMPTZ NOT NULL DEFAULT TIMEZONE('UTC', NOW()),
-
->>>>>>> 48e953d (poller redesign)
-    task_id BIGINT DEFAULT NULL,
-
-    constraint market_indexing_tasks_identity_key
-       unique (id, sp_id, sector_number, piece_offset, piece_size, piece_cid, reg_seal_proof)
-)
-
 CREATE TABLE libp2p_keys (
     sp_id BIGINT NOT NULL,
     priv_key BYTEA NOT NULL,
@@ -255,33 +268,38 @@ CREATE TABLE direct_deals (
     keep_unsealed_copy BOOLEAN
 );
 
+-- Add host column to allow local file based
+-- piece park
 ALTER TABLE parked_piece_refs
     ADD COLUMN host text;
 
-create table file_parked_pieces (
-    id bigserial primary key,
-    created_at timestamp default current_timestamp,
-    piece_cid text not null,
-    piece_padded_size bigint not null,
-    piece_raw_size bigint not null,
-    complete boolean not null default false,
-    task_id bigint default null,
-    cleanup_task_id bigint default null,
-    unique (piece_cid)
-);
+-- Table for old lotus market deals. This is just for deal
+-- which are still alive. It should not be used for any processing
+CREATE TABLE market_legacy_deals (
+    signed_proposal_cid TEXT,
+    sp_id BIGINT,
 
-/*
- * This table is used to keep track of the references to the file parked pieces
- * so that we can delete them when they are no longer needed.
- *
- * All references into the file_parked_pieces table should be done through this table.
- */
-create table file_parked_piece_refs (
-    ref_id bigserial primary key,
-    piece_id bigint not null,
-    data_url text not null,
-    node text not null,
-    foreign key (piece_id) references file_parked_pieces(id) on delete cascade
+    proposal_signature BYTEA,
+    proposal jsonb,
+
+    piece_cid TEXT,
+    piece_size BIGINT,
+
+    offline BOOLEAN,
+    verified BOOLEAN,
+
+    start_epoch BIGINT,
+    end_epoch BIGINT,
+
+    publish_cid TEXT,
+    chain_deal_id BIGINT,
+
+    fast_retrieval BOOLEAN,
+
+    created_at TIMESTAMPTZ,
+    sector_num BIGINT,
+
+    primary key (sp_id, piece_cid, signed_proposal_cid)
 );
 
 

market/storageingest/deal_ingest_seal.go

@@ -9,6 +9,8 @@ import (
 	"net/url"
 	"time"
 
+	"github.com/filecoin-project/curio/build"
+	"github.com/filecoin-project/curio/deps/config"
 	logging "github.com/ipfs/go-log/v2"
 	"golang.org/x/xerrors"
 

market/storageingest/deal_ingest_snap.go

@@ -9,6 +9,8 @@ import (
 	"net/url"
 	"time"
 
+	"github.com/filecoin-project/curio/build"
+	"github.com/filecoin-project/curio/deps/config"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-address"

tasks/indexing/task_indexing.go

@@ -102,8 +102,8 @@ func (i *IndexingTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (do
 		return false, xerrors.Errorf("checking if piece is already indexed: %w", err)
 	}
 
-	// Return early if already indexed
-	if indexed {
+	// Return early if already indexed or should not be indexed
+	if indexed || !task.ShouldIndex {
 		err = i.recordCompletion(ctx, task, taskID, false)
 		if err != nil {
 			return false, err

tasks/storage-market/storage_market.go

@@ -407,7 +407,7 @@ func (d *CurioStorageDealMarket) findURLForOfflineDeals(ctx context.Context, dea
 			if err != nil {
 				return false, xerrors.Errorf("error making GET request: %w", err)
 			}
-
+			
 			// Check the response code for 404
 			if resp.StatusCode != http.StatusOK {
 				if resp.StatusCode != 404 {