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
@@ -276,9 +276,8 @@ workflows:
           get-params: true
           resource_class: 2xlarge
       - test:
-          name: test-idxStore
+          name: test-all
           requires:
             - build
-          suite: idxStore
-          target: "./lib/indexing/indexstore/indexstore_test.go"
+          suite: test-all
           get-params: true

cmd/curio/market.go

@@ -186,7 +186,7 @@ var marketImportdataCmd = &cli.Command{
 			}
 
 			// Insert the offline deal into the deal pipeline
-			_, err = tx.Exec(`INSERT INTO market_mk12_deal_pipeline (url, file_size)
+			_, err = tx.Exec(`INSERT INTO market_mk12_deal_pipeline (url, raw_size)
 								VALUES ($1, $2) WHERE uuid = $3 ON CONFLICT (uuid) DO NOTHING`,
 				pieceIDUrl, rawSize)
 			if err != nil {

deps/config/types.go

@@ -74,7 +74,12 @@ func DefaultCurioConfig() *CurioConfig {
 					InsertBatchSize:   15000,
 				},
 				MK12: MK12Config{
-					Miners: []string{},
+					Miners:                    []string{},
+					PublishMsgPeriod:          Duration(30 * time.Minute),
+					MaxDealsPerPublishMsg:     8,
+					MaxPublishDealsFee:        types.MustParseFIL("5 FIL"),
+					ExpectedPoRepSealDuration: Duration(8 * time.Hour),
+					ExpectedSnapSealDuration:  Duration(2 * time.Hour),
 				},
 			},
 		},
@@ -551,7 +556,7 @@ type StorageMarketConfig struct {
 }
 
 type MK12Config struct {
-	// Miners is a list of miner to enable MK12 deals(Boost) for
+	// Miners is a list of miner addresses to enable MK12 deals(Boost) for
 	Miners []string
 
 	// When a deal is ready to publish, the amount of time to wait for more
@@ -567,10 +572,13 @@ type MK12Config struct {
 	// The maximum fee to pay when sending the PublishStorageDeals message
 	MaxPublishDealsFee types.FIL
 
-	// ExpectedSealDuration is the expected time it would take to seal the deal sector
+	// ExpectedPoRepSealDuration is the expected time it would take to seal the deal sector
 	// This will be used to fail the deals which cannot be sealed on time.
-	// Please make sure to update this to shorter duration for snap deals
-	ExpectedSealDuration Duration
+	ExpectedPoRepSealDuration Duration
+
+	// ExpectedSnapSealDuration is the expected time it would take to snap the deal sector
+	// This will be used to fail the deals which cannot be sealed on time.
+	ExpectedSnapSealDuration Duration
 
 	// SkipCommP can be used to skip doing a commP check before PublishDealMessage is sent on chain
 	// Warning: If this check is skipped and there is a commP mismatch, all deals in the

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

@@ -37,33 +37,14 @@ 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,
+
+    version INT NOT NULL DEFAULT 2
+
     created_at TIMESTAMPTZ  NOT NULL DEFAULT TIMEZONE('UTC', NOW()),
+
     indexed BOOLEAN NOT NULL DEFAULT FALSE,
     indexed_at TIMESTAMPTZ NOT NULL DEFAULT TIMEZONE('UTC', NOW()),
 
@@ -103,21 +84,30 @@ CREATE OR REPLACE FUNCTION process_piece_deal(
     _piece_offset BIGINT,
     _piece_length BIGINT,
     _raw_size BIGINT,
+    _indexed BOOLEAN,
     _legacy_deal BOOLEAN DEFAULT FALSE,
     _chain_deal_id BIGINT DEFAULT 0
 )
 RETURNS VOID AS $$
 BEGIN
-INSERT INTO market_piece_metadata (piece_cid, indexed) VALUES (_piece_cid, TRUE)
-    ON CONFLICT (piece_cid) DO UPDATE SET indexed = TRUE;
+    -- Insert or update the market_piece_metadata table
+INSERT INTO market_piece_metadata (piece_cid, indexed)
+VALUES (_piece_cid, _indexed)
+    ON CONFLICT (piece_cid) DO UPDATE SET
+    indexed = CASE
+                WHEN market_piece_metadata.indexed = FALSE THEN EXCLUDED.indexed
+                ELSE market_piece_metadata.indexed
+END;
 
+    -- Insert into the market_piece_deal table
 INSERT INTO market_piece_deal (
     id, piece_cid, boost_deal, legacy_deal, chain_deal_id,
     sp_id, sector_num, piece_offset, piece_length, raw_size
-) VALUES (
-           _id, _piece_cid, _boost_deal, _legacy_deal, _chain_deal_id,
-           _sp_id, _sector_num, _piece_offset, _piece_length, _raw_size
-       ) ON CONFLICT (sp_id, piece_cid, id) DO NOTHING;
+    ) VALUES (
+             _id, _piece_cid, _boost_deal, _legacy_deal, _chain_deal_id,
+             _sp_id, _sector_num, _piece_offset, _piece_length, _raw_size
+    ) ON CONFLICT (sp_id, piece_cid, id) DO NOTHING;
+
 END;
 $$ LANGUAGE plpgsql;
 
@@ -147,7 +137,7 @@ CREATE TABLE market_mk12_deal_pipeline (
 
     piece_cid TEXT NOT NULL,
     piece_size BIGINT NOT NULL,
-    file_size BIGINT DEFAULT NULL, -- raw piece size
+    raw_size BIGINT DEFAULT NULL,
 
     offline BOOLEAN NOT NULL,
 
@@ -166,14 +156,78 @@ 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 (
@@ -187,27 +241,6 @@ CREATE TABLE market_offline_urls (
     unique (piece_cid)
 );
 
--- 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,
-
-    created_at TIMESTAMPTZ NOT NULL DEFAULT TIMEZONE('UTC', NOW()),
-
-    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,
@@ -216,33 +249,38 @@ CREATE TABLE libp2p_keys (
     no_announce_address TEXT NOT NULL
 );
 
+-- 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/mk12/mk12.go

@@ -436,18 +436,19 @@ func (m *MK12) processDeal(ctx context.Context, deal *ProviderDealState) (*Provi
 				Opaque: fmt.Sprintf("%d", refID),
 			}
 
-			_, err = tx.Exec(`INSERT INTO market_mk12_deal_pipeline (uuid, sp_id, piece_cid, piece_size, offline, url, file_size)
-								VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (uuid) DO NOTHING`,
-				deal.DealUuid.String(), mid, prop.PieceCID.String(), prop.PieceSize, deal.IsOffline, pieceIDUrl, deal.Transfer.Size)
+			_, err = tx.Exec(`INSERT INTO market_mk12_deal_pipeline (uuid, sp_id, piece_cid, piece_size, offline, url, raw_size, should_index)
+								VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (uuid) DO NOTHING`,
+				deal.DealUuid.String(), mid, prop.PieceCID.String(), prop.PieceSize, deal.IsOffline, pieceIDUrl, deal.Transfer.Size,
+				deal.FastRetrieval)
 			if err != nil {
 				return false, xerrors.Errorf("inserting deal into deal pipeline: %w", err)
 			}
 
 		} else {
 			// Insert the offline deal into the deal pipeline
-			_, err = tx.Exec(`INSERT INTO market_mk12_deal_pipeline (uuid, sp_id, piece_cid, piece_size, offline)
-								VALUES ($1, $2, $3, $4, $5) ON CONFLICT (uuid) DO NOTHING`,
-				deal.DealUuid.String(), mid, prop.PieceCID.String(), prop.PieceSize, deal.IsOffline)
+			_, err = tx.Exec(`INSERT INTO market_mk12_deal_pipeline (uuid, sp_id, piece_cid, piece_size, offline, should_index)
+								VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (uuid) DO NOTHING`,
+				deal.DealUuid.String(), mid, prop.PieceCID.String(), prop.PieceSize, deal.IsOffline, deal.FastRetrieval)
 			if err != nil {
 				return false, xerrors.Errorf("inserting deal into deal pipeline: %w", err)
 			}