feat: Remote Seal

.github/workflows/ci.yml

@@ -181,6 +181,8 @@ jobs:
             target: "./itests/pdp_prove_test.go"
           - name: test-itest-pieces-unseal
             target: "./itests/pieces_unseal_test.go"
+          - name: test-itest-remoteseal
+            target: "./itests/remoteseal_test.go"
           - name: test-itest-retrievals-test
             target: "./itests/retrievals_test.go"
           - name: test-itest-sql-idempotent-test
@@ -210,6 +212,7 @@ jobs:
 
       - name: Run tests with coverage
         env:
+          CURIO_OPTIMAL_LIBFILCRYPTO: 0
           FFI_USE_OPENCL: 1
         run: |
           mkdir -p coverage

cmd/curio/tasks/tasks.go

@@ -39,6 +39,7 @@ import (
 	"github.com/filecoin-project/curio/lib/slotmgr"
 	"github.com/filecoin-project/curio/lib/storiface"
 	"github.com/filecoin-project/curio/market/libp2p"
+	"github.com/filecoin-project/curio/market/sealmarket"
 	"github.com/filecoin-project/curio/tasks/balancemgr"
 	"github.com/filecoin-project/curio/tasks/expmgr"
 	"github.com/filecoin-project/curio/tasks/f3"
@@ -51,6 +52,7 @@ import (
 	"github.com/filecoin-project/curio/tasks/pdpv0"
 	piece2 "github.com/filecoin-project/curio/tasks/piece"
 	"github.com/filecoin-project/curio/tasks/proofshare"
+	"github.com/filecoin-project/curio/tasks/remoteseal"
 	"github.com/filecoin-project/curio/tasks/scrub"
 	"github.com/filecoin-project/curio/tasks/seal"
 	"github.com/filecoin-project/curio/tasks/sealsupra"
@@ -234,7 +236,9 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 		cfg.Subsystems.EnableUpdateSubmit ||
 		cfg.Subsystems.EnableCommP ||
 		cfg.Subsystems.EnableProofShare ||
-		cfg.Subsystems.EnableRemoteProofs
+		cfg.Subsystems.EnableRemoteProofs ||
+		cfg.Subsystems.EnableRemoteSealProvider ||
+		cfg.Subsystems.EnableRemoteSealClient
 
 	var p2Active sealsupra.P2Active
 	if hasAnySealingTask {
@@ -375,11 +379,15 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 		pdpv0IpniTask := indexing.NewPDPV0IPNITask(db, cfg, idxMax, iStore)
 		activeTasks = append(activeTasks, ipniTask, indexingTask, pdpv1IdxTask, pdpv1IPNITask, fixRawSizeTask, pdpv0IndexingTask, pdpv0IpniTask)
 
+		// Create SealMarket for remote seal HTTP API
+		if cfg.Subsystems.EnableRemoteSealProvider || cfg.Subsystems.EnableRemoteSealClient {
+			sdeps.SealMarket = sealmarket.NewSealMarket(db, stor, full)
+		}
+
 		if cfg.HTTP.Enable {
-			// TODO: Put this back once PDPv1 is also being used
-			//if !cfg.Subsystems.EnableDealMarket {
-			//	return nil, xerrors.New("deal market must be enabled on HTTP server")
-			//}
+			if !cfg.Subsystems.EnableDealMarket && !cfg.Subsystems.EnableRemoteSealProvider && !cfg.Subsystems.EnableRemoteSealClient {
+				return nil, xerrors.New("deal market or remote seal must be enabled on HTTP server")
+			}
 			err = cuhttp.StartHTTPServer(ctx, dependencies, &sdeps)
 			if err != nil {
 				return nil, xerrors.Errorf("failed to start the HTTP server: %w", err)
@@ -450,6 +458,11 @@ func addSealingTasks(
 	var slotMgr *slotmgr.SlotMgr
 	var addFinalize bool
 
+	// Create the provider poller unconditionally so that sealing nodes (SDR/Batch)
+	// can register the RSealProviderFinalize task even without EnableRemoteSealProvider.
+	// RunPoller is started by the addFinalize block or the EnableRemoteSealProvider block.
+	provPoller := remoteseal.NewProviderPoller(db)
+
 	// NOTE: Tasks with the LEAST priority are at the top
 	if cfg.Subsystems.EnableCommP {
 		scrubUnsealedTask := scrub.NewCommDCheckTask(db, slr)
@@ -477,21 +490,48 @@ func addSealingTasks(
 	if cfg.Subsystems.EnableSealSDR {
 		sdrMax := taskhelp.Max(cfg.Subsystems.SealSDRMaxTasks)
 
-		sdrTask := seal.NewSDRTask(full, db, sp, slr, sdrMax, cfg.Subsystems.SealSDRMinTasks)
+		// provPoller is passed so SDR registers its AddTaskFunc with both SealPoller
+		// and RSealProviderPoller. When EnableRemoteSealProvider is off, only the
+		// local SealPoller is used for scheduling SDR tasks.
+		var sdrProvPoller seal.ProviderPollerSDR
+		if cfg.Subsystems.EnableRemoteSealProvider {
+			sdrProvPoller = provPoller
+		}
+		sdrTask := seal.NewSDRTask(full, db, sp, slr, sdrMax, cfg.Subsystems.SealSDRMinTasks, sdrProvPoller)
 		keyTask := unseal.NewTaskUnsealSDR(slr, db, sdrMax, full)
 
 		activeTasks = append(activeTasks, sdrTask, keyTask)
 	}
 	if cfg.Subsystems.EnableSealSDRTrees {
-		treeDTask := seal.NewTreeDTask(sp, db, slr, cfg.Subsystems.SealSDRTreesMaxTasks, cfg.Subsystems.BindSDRTreeToNode)
-		treeRCTask := seal.NewTreeRCTask(sp, db, slr, cfg.Subsystems.SealSDRTreesMaxTasks)
+		var treeDProvPoller seal.ProviderPollerTreeD
+		var treeRCProvPoller seal.ProviderPollerTreeRC
+		if cfg.Subsystems.EnableRemoteSealProvider {
+			treeDProvPoller = provPoller
+			treeRCProvPoller = provPoller
+		}
+		treeDTask := seal.NewTreeDTask(sp, db, slr, cfg.Subsystems.SealSDRTreesMaxTasks, cfg.Subsystems.BindSDRTreeToNode, treeDProvPoller)
+		treeRCTask := seal.NewTreeRCTask(sp, db, slr, cfg.Subsystems.SealSDRTreesMaxTasks, treeRCProvPoller)
 		synthTask := seal.NewSyntheticProofTask(sp, db, slr, cfg.Subsystems.SyntheticPoRepMaxTasks)
 		activeTasks = append(activeTasks, treeDTask, synthTask, treeRCTask)
 		addFinalize = true
 	}
+	// Remote seal client needs the Finalize task to run after PoRep.
+	// The client skips SDR/Trees (done by provider) but still runs the
+	// standard pipeline from precommit onward, which requires Finalize.
+	if cfg.Subsystems.EnableRemoteSealClient {
+		addFinalize = true
+	}
 	if addFinalize {
 		finalizeTask := seal.NewFinalizeTask(cfg.Subsystems.FinalizeMaxTasks, sp, slr, db, slotMgr)
-		activeTasks = append(activeTasks, finalizeTask)
+		// RSealProviderFinalize must run on sealing nodes (SDR/Batch) that hold
+		// the sector data and batch slots, not only on EnableRemoteSealProvider nodes.
+		provFinalizeTask := remoteseal.NewProviderFinalizeTask(db, provPoller, slr, slotMgr, cfg.Subsystems.FinalizeMaxTasks)
+		activeTasks = append(activeTasks, finalizeTask, provFinalizeTask)
+
+		// Start the provider poller so it can schedule RSealProviderFinalize tasks
+		// on sealing nodes. The poller safely skips task types whose AddTaskFunc
+		// hasn't been registered (e.g. Notify, Cleanup on non-provider nodes).
+		go provPoller.RunPoller(ctx)
 	}
 
 	if cfg.Subsystems.EnableSendPrecommitMsg {
@@ -564,6 +604,48 @@ func addSealingTasks(
 		activeTasks = append(activeTasks, remoteUploadTask, remotePollTask, remoteSendTask)
 	}
 
+	// Remote seal provider tasks (Notify, Cleanup; Finalize is registered above with addFinalize)
+	if cfg.Subsystems.EnableRemoteSealProvider {
+		if !addFinalize {
+			// If no sealing tasks started the poller, start it now for the provider.
+			go provPoller.RunPoller(ctx)
+		}
+
+		provMaxTasks := cfg.Subsystems.RemoteSealProviderMaxTasks
+		cleanupTimeout := cfg.Subsystems.RemoteSealCleanupTimeout
+
+		notifyTask := remoteseal.NewProviderNotifyTask(db, provPoller, provMaxTasks, cleanupTimeout)
+		provCleanupTask := remoteseal.NewProviderCleanupTask(db, provPoller, stor, slotMgr, cfg.Subsystems.FinalizeMaxTasks)
+
+		activeTasks = append(activeTasks, notifyTask, provCleanupTask)
+
+		if !addFinalize {
+			// If the provider finalize task wasn't already registered via addFinalize
+			// (e.g. provider-only node without local sealing), register it here.
+			provFinalizeTask := remoteseal.NewProviderFinalizeTask(db, provPoller, slr, slotMgr, cfg.Subsystems.FinalizeMaxTasks)
+			activeTasks = append(activeTasks, provFinalizeTask)
+		}
+
+		// Provider-side SDR/Tree tasks are handled by the existing SDR/TreeD/TreeRC tasks
+		// via UNION ALL queries - they just need to be enabled (EnableSealSDR/EnableSealSDRTrees).
+		// The SDR/TreeD/TreeRC tasks register their AddTaskFunc with provPoller in their Adder() methods.
+	}
+
+	// Remote seal client tasks
+	if cfg.Subsystems.EnableRemoteSealClient {
+		clientPoller := remoteseal.NewRSealClientPoller(db)
+		go clientPoller.RunPoller(ctx)
+
+		rsealClient := remoteseal.NewRSealClient()
+
+		delegateTask := remoteseal.NewRSealDelegate(db, full, rsealClient)
+		pollTask := remoteseal.NewRSealClientPoll(db, rsealClient, clientPoller)
+		fetchTask := remoteseal.NewRSealClientFetch(db, rsealClient, slr, clientPoller, cfg.Subsystems.RSealClientFetchMaxTasks)
+		cleanupTask := remoteseal.NewRSealClientCleanup(db, rsealClient, clientPoller)
+
+		activeTasks = append(activeTasks, delegateTask, pollTask, fetchTask, cleanupTask)
+	}
+
 	// harmony treats the first task as highest priority, so reverse the order
 	// (we could have just appended to this list in the reverse order, but defining
 	//  tasks in pipeline order is more intuitive)

cmd/curio/test-supra.go

@@ -12,9 +12,12 @@ import (
 	"github.com/urfave/cli/v2"
 	"golang.org/x/xerrors"
 
+	ffi "github.com/filecoin-project/filecoin-ffi"
 	"github.com/filecoin-project/go-state-types/abi"
 
+	"github.com/filecoin-project/curio/build"
 	"github.com/filecoin-project/curio/cmd/curio/internal/translations"
+	"github.com/filecoin-project/curio/harmony/resources"
 	"github.com/filecoin-project/curio/lib/ffi/cunative"
 	"github.com/filecoin-project/curio/lib/supraffi"
 )
@@ -64,6 +67,26 @@ var testSupraSystemInfoCmd = &cli.Command{
 
 		fmt.Println("CUDA:")
 		fmt.Printf("  Usable CUDA GPU detected:    %s\n", yesNo(supraffi.HasUsableCUDAGPU()))
+		fmt.Println()
+
+		fmt.Println("GPU Devices (ffi):")
+		gpuMode := "OpenCL"
+		if build.IsOpencl != "1" {
+			gpuMode = "CUDA"
+		}
+		fmt.Printf("  Mode:                        %s\n", gpuMode)
+		fmt.Printf("  Overprovision factor:        %d\n", resources.GpuOverprovisionFactor)
+		gpus, err := ffi.GetGPUDevices()
+		if err != nil {
+			fmt.Printf("  Error listing GPUs:          %s\n", err)
+		} else if len(gpus) == 0 {
+			fmt.Println("  No GPU devices found")
+		} else {
+			fmt.Printf("  Devices (%d):\n", len(gpus))
+			for i, name := range gpus {
+				fmt.Printf("    [%d] %s\n", i, name)
+			}
+		}
 
 		return nil
 	},

cuhttp/server.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net"
 	"net/http"
 	"os"
 	"strings"
@@ -29,6 +30,7 @@ import (
 	ipni_provider "github.com/filecoin-project/curio/market/ipni/ipni-provider"
 	"github.com/filecoin-project/curio/market/libp2p"
 	"github.com/filecoin-project/curio/market/retrieval"
+	"github.com/filecoin-project/curio/market/sealmarket"
 	"github.com/filecoin-project/curio/pdp"
 	"github.com/filecoin-project/curio/tasks/message"
 	storage_market "github.com/filecoin-project/curio/tasks/storage-market"
@@ -143,6 +145,7 @@ func isWebSocketUpgrade(r *http.Request) bool {
 type ServiceDeps struct {
 	EthSender  *message.SenderETH
 	DealMarket *storage_market.CurioStorageDealMarket
+	SealMarket *sealmarket.SealMarket
 	AlertTask  *alertmanager.AlertTask
 }
 
@@ -206,7 +209,6 @@ func StartHTTPServer(ctx context.Context, d *deps.Deps, sd *ServiceDeps) error {
 
 	// Set up the HTTP server with proper timeouts
 	server := &http.Server{
-		Addr:              cfg.ListenAddress,
 		Handler:           libp2pConnMiddleware(loggingMiddleware(compressionMw(chiRouter))), // Attach middlewares
 		ReadTimeout:       cfg.ReadTimeout,
 		WriteTimeout:      time.Hour * 2,
@@ -225,19 +227,27 @@ func StartHTTPServer(ctx context.Context, d *deps.Deps, sd *ServiceDeps) error {
 		server.TLSConfig = certManager.TLSConfig()
 	}
 
-	// We don't need to run an HTTP server. Any HTTP request should simply be handled as HTTPS.
+	// Bind the listener before starting the goroutine so the caller knows
+	// the actual address (important when ListenAddress uses port 0).
+	ln, err := net.Listen("tcp", cfg.ListenAddress)
+	if err != nil {
+		return xerrors.Errorf("binding HTTP listener on %s: %w", cfg.ListenAddress, err)
+	}
+
+	d.HTTPListenAddr = ln.Addr().String()
+	log.Infof("HTTP server listening on %s (requested %s)", d.HTTPListenAddr, cfg.ListenAddress)
 
-	// Start the server with TLS
+	// Start the server
 	go func() {
-		log.Infof("Starting HTTPS server for https://%s on %s", cfg.DomainName, cfg.ListenAddress)
+		log.Infof("Starting HTTP server for %s on %s", cfg.DomainName, d.HTTPListenAddr)
 		var serr error
 		if !cfg.DelegateTLS {
-			serr = server.ListenAndServeTLS("", "")
+			serr = server.ServeTLS(ln, "", "")
 		} else {
-			serr = server.ListenAndServe()
+			serr = server.Serve(ln)
 		}
 		if serr != nil && !errors.Is(serr, http.ErrServerClosed) {
-			log.Errorf("Failed to start HTTPS server: %s", serr)
+			log.Errorf("Failed to start HTTP server: %s", serr)
 			panic(serr)
 		}
 	}()
@@ -294,37 +304,40 @@ func (c cache) Delete(ctx context.Context, key string) error {
 var _ autocert.Cache = cache{}
 
 func attachRouters(ctx context.Context, r *chi.Mux, d *deps.Deps, sd *ServiceDeps) (*chi.Mux, error) {
-	// Create denylist filter for retrieval endpoints
-	df := denylist.NewFilter(ctx, d.Cfg.HTTP.DenylistServers)
-
-	// Attach retrievals with denylist filtering at both URL and blockstore level
-	rp := retrieval.NewRetrievalProvider(ctx, d.DB, d.IndexStore, d.CachedPieceReader, df)
-	retrieval.Router(r, rp, df)
+	// Deal market routers (retrieval, IPNI, libp2p, market handler) are only
+	// attached when the deal market subsystem is enabled. Other HTTP features
+	// (e.g. remote seal) can run without these dependencies.
+	if d.Cfg.Subsystems.EnableDealMarket {
+		df := denylist.NewFilter(ctx, d.Cfg.HTTP.DenylistServers)
+		rp := retrieval.NewRetrievalProvider(ctx, d.DB, d.IndexStore, d.CachedPieceReader, df)
+		retrieval.Router(r, rp, df)
+
+		ipp, err := ipni_provider.NewProvider(d)
+		if err != nil {
+			return nil, xerrors.Errorf("failed to create new ipni provider: %w", err)
+		}
+		ipni_provider.Routes(r, ipp)
 
-	// Attach IPNI
-	ipp, err := ipni_provider.NewProvider(d)
-	if err != nil {
-		return nil, xerrors.Errorf("failed to create new ipni provider: %w", err)
-	}
-	ipni_provider.Routes(r, ipp)
+		go ipp.StartPublishing(ctx)
 
-	go ipp.StartPublishing(ctx)
+		rd := libp2p.NewRedirector(d.DB)
+		libp2p.Router(r, rd)
 
-	// Attach LibP2P redirector
-	rd := libp2p.NewRedirector(d.DB)
-	libp2p.Router(r, rd)
+		dh, err := mhttp.NewMarketHandler(d.DB, d.Cfg, sd.DealMarket, must.One(d.EthClient.Get()), d.Chain, sd.EthSender, d.LocalStore)
+		if err != nil {
+			return nil, xerrors.Errorf("failed to create new market handler: %w", err)
+		}
+		mhttp.Router(r, dh)
+	}
 
 	if sd.EthSender != nil {
 		pdsvc := pdp.NewPDPService(ctx, d.DB, d.LocalStore, must.One(d.EthClient.Get()), d.Chain, sd.EthSender, sd.AlertTask)
 		pdp.Routes(r, pdsvc)
 	}
 
-	// Attach the market handler
-	dh, err := mhttp.NewMarketHandler(d.DB, d.Cfg, sd.DealMarket, must.One(d.EthClient.Get()), d.Chain, sd.EthSender, d.LocalStore)
-	if err != nil {
-		return nil, xerrors.Errorf("failed to create new market handler: %w", err)
+	if sd.SealMarket != nil {
+		sealmarket.Routes(r, sd.SealMarket)
 	}
-	mhttp.Router(r, dh)
 
 	return r, nil
 }