poller redesign

Author: LexLuthr

.circleci/config.yml

@@ -283,3 +283,10 @@ workflows:
           suite: test-all
           get-params: true
           resource_class: 2xlarge
+      - test:
+          name: test-idxStore
+          requires:
+            - build
+          suite: idxStore
+          target: "./lib/indexing/indexstore/indexstore_test.go"
+          get-params: true

deps/config/doc_gen.go

@@ -493,6 +493,13 @@ description: The default curio configuration
         # type: []string
         #NoAnnounceAddresses = []
 
+      # 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
+      # sector will need to be sent again
+      #
+      # type: bool
+      #SkipCommP = false
+
 
 [Ingest]
   # Maximum number of sectors that can be queued waiting for deals to start processing.

documentation/en/configuration/default-curio-configuration.md

@@ -639,9 +639,15 @@ USAGE:
    curio market command [command options] [arguments...]
 
 COMMANDS:
+<<<<<<< HEAD
    seal     start sealing a deal sector early
    add-url  Add URL to fetch data for offline deals
    help, h  Shows a list of commands or help for one command
+=======
+   seal         start sealing a deal sector early
+   import-data  Import data for offline deal
+   help, h      Shows a list of commands or help for one command
+>>>>>>> 48e953d (poller redesign)
 
 OPTIONS:
    --help, -h  show help
@@ -661,6 +667,7 @@ OPTIONS:
    --help, -h     show help
 ```
 
+<<<<<<< HEAD
 ### curio market add-url
 ```
 NAME:
@@ -674,6 +681,19 @@ OPTIONS:
    --header HEADER, -H HEADER [ --header HEADER, -H HEADER ]  Custom HEADER to include in the HTTP request
    --url URL, -u URL                                          URL to send the request to
    --help, -h                                                 show help
+=======
+### curio market import-data
+```
+NAME:
+   curio market import-data - Import data for offline deal
+
+USAGE:
+   curio market import-data [command options] <deal UUID> <file> <host:port>
+
+OPTIONS:
+   --actor value  Specify actor address to start sealing sectors for
+   --help, -h     show help
+>>>>>>> 48e953d (poller redesign)
 ```
 
 ## curio fetch-params

documentation/en/curio-cli/curio.md

@@ -191,6 +191,7 @@ github.com/cpuguy83/go-md2man/v2 v2.0.4 h1:wfIWP927BUkWJb2NmU/kNDYIBTh/ziUX91+lV
 github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/crackcomm/go-gitignore v0.0.0-20231225121904-e25f5bc08668 h1:ZFUue+PNxmHlu7pYv+IYMtqlaO/0VwaGEqKepZf9JpA=
 github.com/crackcomm/go-gitignore v0.0.0-20231225121904-e25f5bc08668/go.mod h1:p1d6YEZWvFzEh4KLyvBcVSnrfNDDvK2zfK/4x2v/4pE=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/cskr/pubsub v1.0.2 h1:vlOzMhl6PFn60gRlTQQsIfVwaPB/B/8MziK8FhEPt/0=
 github.com/cskr/pubsub v1.0.2/go.mod h1:/8MzYXk/NJAz782G8RPkFzXTZVu63VotefPnR9TIRis=
 github.com/daaku/go.zipexe v1.0.2 h1:Zg55YLYTr7M9wjKn8SY/WcpuuEi+kR2u4E8RhvpyXmk=

go.sum

@@ -34,7 +34,11 @@ create table parked_piece_refs (
     data_headers jsonb not null default '{}',
 
     -- host Added in 202240730-market-migrations.sql
+<<<<<<< HEAD
     -- host text,
+=======
+    host text,
+>>>>>>> 48e953d (poller redesign)
 
     foreign key (piece_id) references parked_pieces(id) on delete cascade
 );

harmony/harmonydb/sql/20240228-piece-park.sql

@@ -1,31 +1,48 @@
+-- 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,
-    piece_cid TEXT NOT NULL,
-    piece_size BIGINT NOT NULL,
+
     offline BOOLEAN NOT NULL,
     verified BOOLEAN NOT NULL,
-    sp_id BIGINT NOT NULL,
+
     start_epoch BIGINT NOT NULL,
     end_epoch BIGINT NOT NULL,
+
     client_peer_id TEXT NOT NULL,
+
     chain_deal_id BIGINT DEFAULT NULL,
     publish_cid TEXT DEFAULT NULL,
+
+    piece_cid TEXT NOT NULL,
+    piece_size BIGINT NOT NULL,
     length BIGINT DEFAULT NULL,
+
     fast_retrieval BOOLEAN NOT NULL,
     announce_to_ipni BOOLEAN NOT NULL,
+
     url TEXT DEFAULT NULL,
     url_headers jsonb NOT NULL DEFAULT '{}',
+
     error TEXT DEFAULT NULL,
 
     primary key (uuid, sp_id, piece_cid, signed_proposal_cid),
     unique (uuid),
     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,
@@ -46,6 +63,7 @@ CREATE TABLE market_legacy_deals (
     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,
@@ -57,14 +75,19 @@ CREATE TABLE market_piece_metadata (
         unique (piece_cid)
 );
 
+-- This table binds the piece metadata to specific deals (piece indexing)
 CREATE TABLE market_piece_deal (
-    id TEXT NOT NULL,
+    id TEXT NOT NULL, -- (UUID for new deals, PropCID for old)
     piece_cid TEXT NOT NULL,
+
     boost_deal BOOLEAN NOT NULL,
     legacy_deal BOOLEAN NOT NULL DEFAULT FALSE,
+
     chain_deal_id BIGINT NOT NULL DEFAULT 0,
+
     sp_id BIGINT NOT NULL,
     sector_num BIGINT NOT NULL,
+
     piece_offset BIGINT NOT NULL,
     piece_length BIGINT NOT NULL,
     raw_size BIGINT NOT NULL,
@@ -74,6 +97,7 @@ CREATE TABLE market_piece_deal (
         unique (sp_id, id)
 );
 
+-- This function is used to insert piece metadata and piece deal (piece indexing)
 CREATE OR REPLACE FUNCTION process_piece_deal(
     _id TEXT,
     _piece_cid TEXT,
@@ -108,64 +132,95 @@ VALUES (
 END;
 $$ LANGUAGE plpgsql;
 
+-- Storage Ask for ask protocol
 CREATE TABLE market_mk12_storage_ask (
+    sp_id BIGINT NOT NULL,
+
     price BIGINT NOT NULL,
     verified_price BIGINT NOT NULL,
+
     min_size BIGINT NOT NULL,
     max_size BIGINT NOT NULL,
-    sp_id BIGINT NOT NULL,
+
     created_at BIGINT NOT NULL,
     expiry BIGINT NOT NULL,
+
     sequence BIGINT NOT NULL,
     unique (sp_id)
 );
 
+-- Used for processing Mk12 deals
 CREATE TABLE market_mk12_deal_pipeline (
     uuid TEXT NOT NULL,
     sp_id BIGINT NOT NULL,
+
     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
+
+>>>>>>> 48e953d (poller redesign)
     offline BOOLEAN NOT NULL,
-    downloaded BOOLEAN DEFAULT FALSE,
+
     url TEXT DEFAULT NULL,
     headers jsonb NOT NULL DEFAULT '{}',
-    file_size BIGINT DEFAULT NULL,
+
     commp_task_id BIGINT DEFAULT NULL,
     after_commp BOOLEAN DEFAULT FALSE,
-    find_task_id BIGINT DEFAULT NULL,
-    after_find BOOLEAN DEFAULT FALSE,
+
     psd_task_id BIGINT DEFAULT NULL,
     after_psd BOOLEAN DEFAULT FALSE,
+
     psd_wait_time TIMESTAMPTZ,
+
     find_deal_task_id BIGINT DEFAULT NULL,
     after_find_deal BOOLEAN DEFAULT FALSE,
+
     sector BIGINT,
     sector_offset BIGINT,
+
     sealed BOOLEAN DEFAULT FALSE,
     indexed BOOLEAN DEFAULT FALSE,
 
     constraint market_mk12_deal_pipeline_identity_key unique (uuid)
 );
 
+-- 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,
+
     url TEXT NOT NULL,
     headers jsonb NOT NULL DEFAULT '{}',
+
     raw_size BIGINT NOT NULL,
+
     unique (piece_cid)
 );
 
+-- indexing tracker is separate from
 CREATE TABLE market_indexing_tasks (
-    id TEXT NOT NULL,
+    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
@@ -200,6 +255,35 @@ CREATE TABLE direct_deals (
     keep_unsealed_copy BOOLEAN
 );
 
+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)
+);
+
+/*
+ * 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
+);
+
 
 
 

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

@@ -4,6 +4,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"os"
 
 	"golang.org/x/xerrors"
 )
@@ -62,8 +63,33 @@ func (u *UrlPieceReader) Read(p []byte) (n int, err error) {
 			return 0, xerrors.Errorf("a non 200 response code: %s", resp.Status)
 		}
 
-		// Set 'active' to the response body
-		u.active = resp.Body
+		if goUrl.Scheme == "file" {
+			fileUrl := goUrl.Path
+			file, err := os.Open(fileUrl)
+			if err != nil {
+				return 0, xerrors.Errorf("error opening file: %w", err)
+			}
+			u.active = file
+		} else {
+			req, err := http.NewRequest(http.MethodGet, u.Url, nil)
+			if err != nil {
+				return 0, xerrors.Errorf("error creating request: %w", err)
+			}
+			// Add custom headers for security and authentication
+			req.Header = u.Headers
+			// Create a client and make the request
+			client := &http.Client{}
+			resp, err := client.Do(req)
+			if err != nil {
+				return 0, xerrors.Errorf("error making GET request: %w", err)
+			}
+			if resp.StatusCode != 200 {
+				return 0, xerrors.Errorf("a non 200 response code: %s", resp.Status)
+			}
+
+			// Set 'active' to the response body
+			u.active = resp.Body
+		}
 	}
 
 	// Calculate the maximum number of bytes we can read without exceeding RawSize

lib/dealdata/urlpiecereader.go

@@ -1,9 +1,7 @@
 CREATE TABLE IF NOT EXISTS PayloadToPiece (
-    PieceCid BLOB,
+    PieceCid BLOB, -- 20 bytes trimmed
     PayloadMultihash BLOB,
     BlockOffset BIGINT,
     BlockSize BIGINT,
     PRIMARY KEY (PayloadMultihash, PieceCid)
-);
-
-CREATE INDEX piece_to_payload ON PayloadToPiece(PieceCid, PayloadMultihash);
+);