sector import: Remote Commit1

api/api_storage.go

@@ -572,6 +572,18 @@ type RemoteSectorMeta struct {
 	////////
 	// SEALING SERVICE HOOKS
 
-	// todo Commit1Provider
+	// URL
+	// todo better doc
+	RemoteCommit1Endpoint string
+
 	// todo OnDone / OnStateChange
 }
+
+type RemoteCommit1Params struct {
+	Ticket, Seed []byte
+
+	Unsealed cid.Cid
+	Sealed   cid.Cid
+
+	ProofType abi.RegisteredSealProof
+}

cmd/lotus-miner/info.go

@@ -545,6 +545,7 @@ var stateList = []stateMeta{
 	{col: color.FgRed, state: sealing.SealPreCommit2Failed},
 	{col: color.FgRed, state: sealing.PreCommitFailed},
 	{col: color.FgRed, state: sealing.ComputeProofFailed},
+	{col: color.FgRed, state: sealing.RemoteCommit1Failed},
 	{col: color.FgRed, state: sealing.CommitFailed},
 	{col: color.FgRed, state: sealing.CommitFinalizeFailed},
 	{col: color.FgRed, state: sealing.PackingFailed},

documentation/en/api-v0-methods-miner.md

@@ -3298,7 +3298,8 @@ Inputs:
           "Value": "string value"
         }
       ]
-    }
+    },
+    "RemoteCommit1Endpoint": "string value"
   }
 ]
 ```

itests/sector_import_test.go

@@ -3,6 +3,7 @@ package itests
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
@@ -118,6 +119,7 @@ func TestSectorImportAfterPC2(t *testing.T) {
 
 	m := mux.NewRouter()
 	m.HandleFunc("/sectors/{type}/{id}", remoteGetSector(sectorDir)).Methods("GET")
+	m.HandleFunc("/sectors/{id}/commit1", remoteCommit1(sealer)).Methods("POST")
 	srv := httptest.NewServer(m)
 
 	////////
@@ -155,6 +157,8 @@ func TestSectorImportAfterPC2(t *testing.T) {
 			Local: false,
 			URL:   fmt.Sprintf("%s/sectors/cache/s-t0%d-%d", srv.URL, mid, snum),
 		},
+
+		RemoteCommit1Endpoint: fmt.Sprintf("%s/sectors/s-t0%d-%d/commit1", srv.URL, mid, snum),
 	})
 	require.NoError(t, err)
 
@@ -167,6 +171,54 @@ func TestSectorImportAfterPC2(t *testing.T) {
 	miner.WaitSectorsProving(ctx, map[abi.SectorNumber]struct{}{snum: {}})
 }
 
+func remoteCommit1(s *ffiwrapper.Sealer) func(w http.ResponseWriter, r *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		vars := mux.Vars(r)
+
+		// validate sector id
+		id, err := storiface.ParseSectorID(vars["id"])
+		if err != nil {
+			w.WriteHeader(500)
+			return
+		}
+
+		var params api.RemoteCommit1Params
+		if err := json.NewDecoder(r.Body).Decode(&params); err != nil {
+			w.WriteHeader(500)
+			return
+		}
+
+		sref := storiface.SectorRef{
+			ID:        id,
+			ProofType: params.ProofType,
+		}
+
+		ssize, err := params.ProofType.SectorSize()
+		if err != nil {
+			w.WriteHeader(500)
+			return
+		}
+
+		p, err := s.SealCommit1(r.Context(), sref, params.Ticket, params.Seed, []abi.PieceInfo{
+			{
+				Size:     abi.PaddedPieceSize(ssize),
+				PieceCID: params.Unsealed,
+			},
+		}, storiface.SectorCids{
+			Unsealed: params.Unsealed,
+			Sealed:   params.Sealed,
+		})
+		if err != nil {
+			w.WriteHeader(500)
+			return
+		}
+
+		if _, err := w.Write(p); err != nil {
+			fmt.Println("c1 write error")
+		}
+	}
+}
+
 func remoteGetSector(sectorRoot string) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 

storage/pipeline/cbor_gen.go

@@ -31,7 +31,7 @@ func (t *SectorInfo) MarshalCBOR(w io.Writer) error {
 
 	cw := cbg.NewCborWriter(w)
 
-	if _, err := cw.Write([]byte{184, 36}); err != nil {
+	if _, err := cw.Write([]byte{184, 37}); err != nil {
 		return err
 	}
 
@@ -703,6 +703,29 @@ func (t *SectorInfo) MarshalCBOR(w io.Writer) error {
 		return err
 	}
 
+	// t.RemoteCommit1Endpoint (string) (string)
+	if len("RemoteCommit1Endpoint") > cbg.MaxLength {
+		return xerrors.Errorf("Value in field \"RemoteCommit1Endpoint\" was too long")
+	}
+
+	if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("RemoteCommit1Endpoint"))); err != nil {
+		return err
+	}
+	if _, err := io.WriteString(w, string("RemoteCommit1Endpoint")); err != nil {
+		return err
+	}
+
+	if len(t.RemoteCommit1Endpoint) > cbg.MaxLength {
+		return xerrors.Errorf("Value in field t.RemoteCommit1Endpoint was too long")
+	}
+
+	if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.RemoteCommit1Endpoint))); err != nil {
+		return err
+	}
+	if _, err := io.WriteString(w, string(t.RemoteCommit1Endpoint)); err != nil {
+		return err
+	}
+
 	// t.RemoteDataFinalized (bool) (bool)
 	if len("RemoteDataFinalized") > cbg.MaxLength {
 		return xerrors.Errorf("Value in field \"RemoteDataFinalized\" was too long")
@@ -1479,6 +1502,17 @@ func (t *SectorInfo) UnmarshalCBOR(r io.Reader) (err error) {
 				}
 
 			}
+			// t.RemoteCommit1Endpoint (string) (string)
+		case "RemoteCommit1Endpoint":
+
+			{
+				sval, err := cbg.ReadString(cr)
+				if err != nil {
+					return err
+				}
+
+				t.RemoteCommit1Endpoint = string(sval)
+			}
 			// t.RemoteDataFinalized (bool) (bool)
 		case "RemoteDataFinalized":
 

storage/pipeline/fsm.go

@@ -225,6 +225,9 @@ var fsmPlanners = map[SectorState]func(events []statemachine.Event, state *Secto
 		on(SectorRetryComputeProof{}, Committing),
 		on(SectorSealPreCommit1Failed{}, SealPreCommit1Failed),
 	),
+	RemoteCommit1Failed: planOne(
+		on(SectorRetryComputeProof{}, Committing),
+	),
 	CommitFinalizeFailed: planOne(
 		on(SectorRetryFinalize{}, CommitFinalize),
 	),
@@ -539,6 +542,8 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
 		return m.handlePreCommitFailed, processed, nil
 	case ComputeProofFailed:
 		return m.handleComputeProofFailed, processed, nil
+	case RemoteCommit1Failed:
+		return m.handleRemoteCommit1Failed, processed, nil
 	case CommitFailed:
 		return m.handleCommitFailed, processed, nil
 	case CommitFinalizeFailed:
@@ -666,6 +671,8 @@ func planCommitting(events []statemachine.Event, state *SectorInfo) (uint64, err
 			return uint64(i + 1), nil
 		case SectorComputeProofFailed:
 			state.State = ComputeProofFailed
+		case SectorRemoteCommit1Failed:
+			state.State = RemoteCommit1Failed
 		case SectorSealPreCommit1Failed:
 			state.State = SealPreCommit1Failed
 		case SectorCommitFailed:

storage/pipeline/fsm_events.go

@@ -218,6 +218,11 @@ func (evt SectorSeedReady) apply(state *SectorInfo) {
 	state.SeedValue = evt.SeedValue
 }
 
+type SectorRemoteCommit1Failed struct{ error }
+
+func (evt SectorRemoteCommit1Failed) FormatError(xerrors.Printer) (next error) { return evt.error }
+func (evt SectorRemoteCommit1Failed) apply(*SectorInfo)                        {}
+
 type SectorComputeProofFailed struct{ error }
 
 func (evt SectorComputeProofFailed) FormatError(xerrors.Printer) (next error) { return evt.error }

storage/pipeline/receive.go

@@ -114,8 +114,9 @@ func (m *Sealing) checkSectorMeta(ctx context.Context, meta api.RemoteSectorMeta
 		if meta.DataCache == nil {
 			return SectorInfo{}, xerrors.Errorf("expected DataCache to be set")
 		}
-		info.RemoteDataSealed = meta.DataSealed
+		info.RemoteDataSealed = meta.DataSealed // todo make head requests to check?
 		info.RemoteDataCache = meta.DataCache
+		info.RemoteCommit1Endpoint = meta.RemoteCommit1Endpoint
 
 		// If we get a sector after PC2, assume that we're getting finalized sector data
 		// todo: maybe only set if C1 provider is set?

storage/pipeline/sector_state.go

@@ -31,6 +31,7 @@ var ExistSectorStateList = map[SectorState]struct{}{
 	SealPreCommit2Failed:        {},
 	PreCommitFailed:             {},
 	ComputeProofFailed:          {},
+	RemoteCommit1Failed:         {},
 	CommitFailed:                {},
 	PackingFailed:               {},
 	FinalizeFailed:              {},
@@ -124,6 +125,7 @@ const (
 	SealPreCommit2Failed SectorState = "SealPreCommit2Failed"
 	PreCommitFailed      SectorState = "PreCommitFailed"
 	ComputeProofFailed   SectorState = "ComputeProofFailed"
+	RemoteCommit1Failed  SectorState = "RemoteCommit1Failed"
 	CommitFailed         SectorState = "CommitFailed"
 	PackingFailed        SectorState = "PackingFailed" // TODO: deprecated, remove
 	FinalizeFailed       SectorState = "FinalizeFailed"

storage/pipeline/states_failed.go

@@ -184,6 +184,18 @@ func (m *Sealing) handleComputeProofFailed(ctx statemachine.Context, sector Sect
 	return ctx.Send(SectorRetryComputeProof{})
 }
 
+func (m *Sealing) handleRemoteCommit1Failed(ctx statemachine.Context, sector SectorInfo) error {
+	if err := failedCooldown(ctx, sector); err != nil {
+		return err
+	}
+
+	if sector.InvalidProofs > 1 {
+		log.Errorw("consecutive remote commit1 fails", "sector", sector.SectorNumber, "c1url", sector.RemoteCommit1Endpoint)
+	}
+
+	return ctx.Send(SectorRetryComputeProof{})
+}
+
 func (m *Sealing) handleSubmitReplicaUpdateFailed(ctx statemachine.Context, sector SectorInfo) error {
 	if err := failedCooldown(ctx, sector); err != nil {
 		return err

storage/pipeline/states_sealing.go

@@ -3,6 +3,9 @@ package sealing
 import (
 	"bytes"
 	"context"
+	"encoding/json"
+	"io"
+	"net/http"
 
 	"github.com/ipfs/go-cid"
 	"golang.org/x/xerrors"
@@ -569,13 +572,50 @@ func (m *Sealing) handleCommitting(ctx statemachine.Context, sector SectorInfo)
 		return ctx.Send(SectorCommitFailed{xerrors.Errorf("sector had nil commR or commD")})
 	}
 
-	cids := storiface.SectorCids{
+	var c2in storiface.Commit1Out
-		Unsealed: *sector.CommD,
+	if sector.RemoteCommit1Endpoint == "" {
-		Sealed:   *sector.CommR,
+		cids := storiface.SectorCids{
-	}
+			Unsealed: *sector.CommD,
-	c2in, err := m.sealer.SealCommit1(sector.sealingCtx(ctx.Context()), m.minerSector(sector.SectorType, sector.SectorNumber), sector.TicketValue, sector.SeedValue, sector.pieceInfos(), cids)
+			Sealed:   *sector.CommR,
-	if err != nil {
+		}
-		return ctx.Send(SectorComputeProofFailed{xerrors.Errorf("computing seal proof failed(1): %w", err)})
+		c2in, err = m.sealer.SealCommit1(sector.sealingCtx(ctx.Context()), m.minerSector(sector.SectorType, sector.SectorNumber), sector.TicketValue, sector.SeedValue, sector.pieceInfos(), cids)
+		if err != nil {
+			return ctx.Send(SectorComputeProofFailed{xerrors.Errorf("computing seal proof failed(1): %w", err)})
+		}
+	} else {
+		reqData := api.RemoteCommit1Params{
+			Ticket:    sector.TicketValue,
+			Seed:      sector.SeedValue,
+			Unsealed:  *sector.CommD,
+			Sealed:    *sector.CommR,
+			ProofType: sector.SectorType,
+		}
+		reqBody, err := json.Marshal(&reqData)
+		if err != nil {
+			return xerrors.Errorf("marshaling remote commit1 request: %w", err)
+		}
+
+		req, err := http.NewRequest("POST", sector.RemoteCommit1Endpoint, bytes.NewReader(reqBody))
+		if err != nil {
+			return ctx.Send(SectorRemoteCommit1Failed{xerrors.Errorf("creating new remote commit1 request: %w", err)})
+		}
+		req.Header.Set("Content-Type", "application/json")
+		req = req.WithContext(ctx.Context())
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return ctx.Send(SectorRemoteCommit1Failed{xerrors.Errorf("requesting remote commit1: %w", err)})
+		}
+
+		defer resp.Body.Close()
+
+		if resp.StatusCode != http.StatusOK {
+			return ctx.Send(SectorRemoteCommit1Failed{xerrors.Errorf("remote commit1 received non-200 http response %s", resp.Status)})
+		}
+
+		c2in, err = io.ReadAll(resp.Body) // todo some len constraint
+		if err != nil {
+			return ctx.Send(SectorRemoteCommit1Failed{xerrors.Errorf("reading commit1 response: %w", err)})
+		}
 	}
 
 	proof, err := m.sealer.SealCommit2(sector.sealingCtx(ctx.Context()), m.minerSector(sector.SectorType, sector.SectorNumber), c2in)

storage/pipeline/types.go

@@ -94,10 +94,11 @@ type SectorInfo struct {
 	TerminatedAt     abi.ChainEpoch
 
 	// Remote import
-	RemoteDataUnsealed  *storiface.SectorData
+	RemoteDataUnsealed    *storiface.SectorData
-	RemoteDataSealed    *storiface.SectorData
+	RemoteDataSealed      *storiface.SectorData
-	RemoteDataCache     *storiface.SectorData
+	RemoteDataCache       *storiface.SectorData
-	RemoteDataFinalized bool
+	RemoteCommit1Endpoint string
+	RemoteDataFinalized   bool
 
 	// Debug
 	LastErr string