finish libp2p init, fix tests

Author: LexLuthr

Makefile

@@ -118,7 +118,7 @@ debug: build
 all: build 
 .PHONY: all
 
-build: curio
+build: curio sptool
 	@[[ $$(type -P "curio") ]] && echo "Caution: you have \
 an existing curio binary in your PATH. This may cause problems if you don't run 'sudo make install'" || true
 

cmd/curio/rpc/rpc.go

@@ -250,7 +250,6 @@ func (p *CurioAPI) LogSetLevel(ctx context.Context, subsystem, level string) err
 }
 
 func ListenAndServe(ctx context.Context, dependencies *deps.Deps, shutdownChan chan struct{}) error {
-	log.Errorf("ENTERED RPC SERVER")
 	fh := &paths.FetchHandler{Local: dependencies.LocalStore, PfHandler: &paths.DefaultPartialFileHandler{}}
 	remoteHandler := func(w http.ResponseWriter, r *http.Request) {
 		if !auth.HasPerm(r.Context(), nil, lapi.PermAdmin) {
@@ -262,8 +261,6 @@ func ListenAndServe(ctx context.Context, dependencies *deps.Deps, shutdownChan c
 		fh.ServeHTTP(w, r)
 	}
 
-	log.Errorf("CREATED STORAGE HANDLER")
-
 	var authVerify func(context.Context, string) ([]auth.Permission, error)
 	{
 		privateKey, err := base64.StdEncoding.DecodeString(dependencies.Cfg.Apis.StorageRPCSecret)

cmd/curio/run.go

@@ -130,7 +130,6 @@ var runCmd = &cli.Command{
 		}
 		defer taskEngine.GracefullyTerminate()
 
-		log.Infof("WILL START RPC SERVER NOW")
 		err = rpc.ListenAndServe(ctx, dependencies, shutdownChan) // Monitor for shutdown.
 		if err != nil {
 			return err

cmd/curio/tasks/tasks.go

@@ -27,6 +27,7 @@ import (
 	"github.com/filecoin-project/curio/lib/curiochain"
 	"github.com/filecoin-project/curio/lib/fastparamfetch"
 	"github.com/filecoin-project/curio/lib/ffi"
+	"github.com/filecoin-project/curio/lib/libp2p"
 	"github.com/filecoin-project/curio/lib/multictladdr"
 	"github.com/filecoin-project/curio/lib/paths"
 	"github.com/filecoin-project/curio/lib/slotmgr"
@@ -221,15 +222,11 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps) (*harmonytask.Task
 		dealFindTask := storage_market.NewFindDealTask(dm, db, full, &cfg.Market.StorageMarketConfig.MK12)
 		activeTasks = append(activeTasks, psdTask, dealFindTask)
 
-		//p2pMap, err := libp2p.NewLibp2pHost(ctx, db, cfg)
-		//if err != nil {
-		//	return nil, err
-		//}
-		//
-		//for _, h := range p2pMap {
-		//	dp := libp2pimpl.NewDealProvider(h, db, dm.MK12Handler, full)
-		//	go dp.Start(ctx)
-		//}
+		// Start libp2p hosts and handle streams
+		err = libp2p.NewDealProvider(ctx, db, cfg, dm.MK12Handler, full, machine)
+		if err != nil {
+			return nil, err
+		}
 
 		indexingTask := indexing.NewIndexingTask(db, sc, iStore, pp, cfg)
 		activeTasks = append(activeTasks, indexingTask)

deps/config/doc_gen.go

@@ -74,7 +74,7 @@ func DefaultCurioConfig() *CurioConfig {
 					InsertBatchSize:   15000,
 				},
 				MK12: MK12Config{
-					Miners:                    []string{},
+					Libp2p:                    []Libp2pConfig{},
 					PublishMsgPeriod:          Duration(30 * time.Minute),
 					MaxDealsPerPublishMsg:     8,
 					MaxPublishDealsFee:        types.MustParseFIL("5 FIL"),
@@ -556,8 +556,9 @@ type StorageMarketConfig struct {
 }
 
 type MK12Config struct {
-	// Miners is a list of miner addresses to enable MK12 deals(Boost) for
-	Miners []string
+	// Libp2p is a list of libp2p config for each miner ID. These values must be set explicitly
+	// for each miner ID.
+	Libp2p []Libp2pConfig
 
 	// When a deal is ready to publish, the amount of time to wait for more
 	// deals to be ready to publish before publishing them all as a batch
@@ -598,3 +599,18 @@ type IndexingConfig struct {
 	// Number of concurrent inserts to split AddIndex calls to
 	InsertConcurrency int
 }
+
+type Libp2pConfig struct {
+	// Miners ID for which MK12 deals (boosts) should be enabled and associated with this libp2p configuration.
+	Miner string
+	// Binding address for the libp2p host - 0 means random port.
+	// Format: multiaddress; see https://multiformats.io/multiaddr/
+	ListenAddresses []string
+	// Addresses to explicitally announce to other peers. If not specified,
+	// all interface addresses are announced
+	// Format: multiaddress
+	AnnounceAddresses []string
+	// Addresses to not announce
+	// Format: multiaddress
+	NoAnnounceAddresses []string
+}

deps/config/types.go

@@ -412,10 +412,11 @@ description: The default curio configuration
       #InsertConcurrency = 8
 
     [Market.StorageMarketConfig.MK12]
-      # Miners is a list of miner addresses to enable MK12 deals(Boost) for
+      # Libp2p is a list of libp2p config for each miner ID. These values must be set explicitly
+      # for each miner ID.
       #
-      # type: []string
-      #Miners = []
+      # type: []Libp2pConfig
+      #Libp2p = []
 
       # When a deal is ready to publish, the amount of time to wait for more
       # deals to be ready to publish before publishing them all as a batch

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

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

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

@@ -240,14 +240,75 @@ CREATE TABLE market_offline_urls (
     unique (piece_cid)
 );
 
-CREATE TABLE libp2p_keys (
+-- This table is used for coordinating libp2p nodes
+CREATE TABLE libp2p (
     sp_id BIGINT NOT NULL,
     priv_key BYTEA NOT NULL,
-    listen_address TEXT NOT NULL,
-    announce_address TEXT NOT NULL,
-    no_announce_address TEXT NOT NULL
+    listen_address TEXT DEFAULT NULL,
+    announce_address TEXT DEFAULT NULL,
+    no_announce_address TEXT DEFAULT NULL,
+    running_on TEXT DEFAULT NULL,
+    updated_at TIMESTAMPTZ DEFAULT NULL,
+
+    -- sp_id, priv_key must be unique for miner<>peerID combo uniqueness
+    -- announce address should be unique to avoid having 2 peerID reachable on same address
+    constraint market_libp2p_identity_key unique (sp_id, priv_key, announce_address)
 );
 
+-- -- Function used to update the libp2p table
+CREATE OR REPLACE FUNCTION insert_or_update_libp2p(
+    _sp_id BIGINT,
+    _listen_address TEXT,
+    _announce_address TEXT,
+    _no_announce_address TEXT,
+    _running_on TEXT
+)
+RETURNS BYTEA AS $$
+DECLARE
+_priv_key BYTEA;
+    _current_running_on TEXT;
+    _current_updated_at TIMESTAMPTZ;
+BEGIN
+    -- Check if the sp_id exists and retrieve the current values
+    SELECT priv_key, running_on, updated_at INTO _priv_key, _current_running_on, _current_updated_at
+    FROM libp2p
+    WHERE sp_id = _sp_id;
+
+    -- Raise an exception if no row was found
+    IF NOT FOUND THEN
+            RAISE EXCEPTION 'libp2p key for sp_id "%" does not exist', _sp_id;
+    END IF;
+
+    -- If the sp_id exists and running_on is NULL or matches _running_on
+    IF _current_running_on IS NULL OR _current_running_on = _running_on THEN
+        -- Update the record with the provided values and set updated_at to NOW
+        UPDATE libp2p
+        SET
+            listen_address = _listen_address,
+            announce_address = _announce_address,
+            no_announce_address = _no_announce_address,
+            running_on = _running_on,
+            updated_at = NOW() AT TIME ZONE 'UTC'
+        WHERE sp_id = _sp_id;
+    ELSIF _current_updated_at > NOW() - INTERVAL '10 seconds' THEN
+            -- Raise an exception if running_on is different and updated_at is recent
+            RAISE EXCEPTION 'Libp2p node already running on "%"', _current_running_on;
+    ELSE
+            -- Update running_on and other columns if updated_at is older than 10 seconds
+        UPDATE libp2p
+        SET
+            listen_address = _listen_address,
+            announce_address = _announce_address,
+            no_announce_address = _no_announce_address,
+            running_on = _running_on,
+            updated_at = NOW() AT TIME ZONE 'UTC'
+        WHERE sp_id = _sp_id;
+    END IF;
+
+RETURN _priv_key;
+END;
+$$ LANGUAGE plpgsql;
+
 -- Add host column to allow local file based
 -- piece park
 ALTER TABLE parked_piece_refs
@@ -275,9 +336,6 @@ CREATE TABLE market_legacy_deals (
     publish_cid TEXT  NOT NULL,
     chain_deal_id BIGINT  NOT NULL,
 
-    piece_cid TEXT NOT NULL,
-    piece_size BIGINT NOT NULL,
-
     fast_retrieval BOOLEAN  NOT NULL,
 
     created_at TIMESTAMPTZ  NOT NULL,

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

@@ -27,7 +27,8 @@ func TestAlertNow(t *testing.T) {
 		Port:     "5433",
 	}
 	// Create dependencies
-	db, err := harmonydb.NewFromConfigWithITestID(t, dbConfig, "alertnow")
+	sharedITestID := harmonydb.ITestNewID()
+	db, err := harmonydb.NewFromConfigWithITestID(t, dbConfig, sharedITestID)
 	require.NoError(t, err)
 
 	an := alertmanager.NewAlertNow(db, "alertNowMachine")
@@ -40,7 +41,7 @@ func TestAlertNow(t *testing.T) {
 	done, err := at.Do(123, func() bool { return true })
 	require.NoError(t, err)
 	require.True(t, done)
-	require.Equal(t, "alertNowMachine: testMessage", tp.output)
+	require.Equal(t, "Machine alertNowMachine: testMessage", tp.output)
 }
 
 // testPlugin is a test plugin
@@ -49,6 +50,6 @@ type testPlugin struct {
 }
 
 func (tp *testPlugin) SendAlert(data *plugin.AlertPayload) error {
-	tp.output = data.Summary
+	tp.output = data.Details["NowCheck"].(string)
 	return nil
 }