add index to event.emitter_addr

* exclude reverted events from results returned by eth_getLogs

* unit test

* update CHANGELOG.md

.circleci/config.yml

@@ -945,6 +945,7 @@ workflows:
             - build
           suite: itest-sector_pledge
           target: "./itests/sector_pledge_test.go"
+          resource_class: 2xlarge
           get-params: true
           
       - test:

.circleci/template.yml

@@ -551,7 +551,7 @@ workflows:
             - build
           suite: itest-[[ $name ]]
           target: "./itests/[[ $file ]]"
-          [[- if or (eq $name "worker") (eq $name "deals_concurrent") (eq $name "wdpost_worker_config")]]
+          [[- if or (eq $name "worker") (eq $name "deals_concurrent") (eq $name "wdpost_worker_config") (eq $name "sector_pledge")]]
           resource_class: 2xlarge
           [[- end]]
           [[- if or (eq $name "wdpost") (eq $name "sector_pledge")]]

CHANGELOG.md

@@ -6,12 +6,12 @@
 
 ## Improvements
 
-# v1.26.0-rc2 / 2024-03-0y
+# v1.26.0-rc3 / 2024-03-13
 
-This is a release candidate of the upcoming MANDATORY Lotus v1.26.0 release, which will deliver the Filecoin network version 22, codenamed Dragon 🐉.
+This is the third release candidate of the upcoming MANDATORY Lotus v1.26.0 release, which will deliver the Filecoin network version 22, codenamed Dragon 🐉. This third release candidate brings in multiple storage provider fixes, and cleanups for issues encountered during testing in the Calibration network.
+
+**This release candidate only has the calibration network to upgrade set, which happened at epoch 1427974, 2024-03-11T14:00:00Z. This release does NOT set the mainnet upgrade epoch, which will be updated in the final release.**
 
-**This release candidate sets the calibration network to upgrade at epoch 1427974, which is 2024-03-11T14:00:00Z**
-This release does NOT set the mainnet upgrade epoch yet, in which will be updated in the final release.
 The Filecoin network version 22 delivers the following FIPs:
 
 - [FIP-0063: Switching to new Drand mainnet network](https://github.com/filecoin-project/FIPs/blob/master/FIPS/fip-0063.md)
@@ -68,6 +68,14 @@ For certain node operators, such as full archival nodes or systems that need to
 - feat: events: Add Lotus APIs to consume smart contract and built-in actor events ([filecoin-project/lotus#11618](https://github.com/filecoin-project/lotus/pull/11618))
 
 ## Improvements
+- fix: commit batch: Always go through commit batcher ([filecoin-project/lotus#11704](https://github.com/filecoin-project/lotus/pull/11704))
+- sealing pipeline: Fix panic on padding pieces in WaitDeals ([filecoin-project/lotus#11708](https://github.com/filecoin-project/lotus/pull/11708))
+- sealing pipeline: Output DDO pieces in SectorStatus ([filecoin-project/lotus#11709](https://github.com/filecoin-project/lotus/pull/11709))
+- sealing pipeline: Fix failing ProveCommit3 aggregate ([filecoin-project/lotus#11710](https://github.com/filecoin-project/lotus/pull/11710))
+- fix: beacon: validate drand change at nv16 correctly ([filecoin-project/lotus#11690](https://github.com/filecoin-project/lotus/pull/11690))
+- fix: events: properly disable actor events API via cfg ([filecoin-project/lotus#11701](https://github.com/filecoin-project/lotus/pull/11701))
+- fix: events: Enable event storage config for Actor events as well ([filecoin-project/lotus#11712](https://github.com/filecoin-project/lotus/pull/11712))
+
 
 ## Tracing API
 
@@ -203,6 +211,9 @@ OPTIONS:
 - fix: add UpgradePhoenixHeight to StateGetNetworkParams (#11648) ([filecoin-project/lotus#11648](https://github.com/filecoin-project/lotus/pull/11648))
 - feat: drand quicknet: allow scheduling drand quicknet upgrade before nv22 on 2k devnet ([filecoin-project/lotus#11667]https://github.com/filecoin-project/lotus/pull/11667)
 - chore: backport #11632 to release/v1.26.0 ([filecoin-project/lotus#11667](https://github.com/filecoin-project/lotus/pull/11667))
+- release: bump to v1.26.0-rc2 ([filecoin-project/lotus#11691](https://github.com/filecoin-project/lotus/pull/11691))
+- Docs: Drand: document the meaning of "IsChained ([filecoin-project/lotus#11692](https://github.com/filecoin-project/lotus/pull/11692))
+- chore: remove old calibnet bootstrappers ([filecoin-project/lotus#11702](https://github.com/filecoin-project/lotus/pull/11702))
 
 # v1.25.2 / 2024-01-11 
 
@@ -396,8 +407,6 @@ This patch release allows for up to 10k messages per block. Additionally, it int
 ## Improvements
 - fix: exchange: allow up to 10k messages per block ([filecoin-project/lotus#11506](https://github.com/filecoin-project/lotus/pull/11506))
 
->>>>>>> releases
-
 # v 1.25.0 / 2023-11-22
 
 This is a highly recommended feature release of Lotus. This optional release supports the Filecoin network version 21 upgrade, codenamed Watermelon 🍉, in addition to the numerous improvements and enhancements for node operators, ETH RPC-providers and storage providers.
@@ -485,6 +494,7 @@ Lotus-workers can now be built to leverage the SupraSeal C2 sealing optimization
 - fix: lotus-provider: lotus-provider msg sending ([filecoin-project/lotus#11480](https://github.com/filecoin-project/lotus/pull/11480))
 - fix: lotus-provider: Fix winning PoSt ([filecoin-project/lotus#11483](https://github.com/filecoin-project/lotus/pull/11483))
 - chore: fix: sql Scan cannot write to an object ([filecoin-project/lotus#11487](https://github.com/filecoin-project/lotus/pull/11487))
+- fix: Exclude reverted events in `eth_getLogs` results [filecoin-project/lotus#11318](https://github.com/filecoin-project/lotus/pull/11318)
 
 ## Dependencies
 - deps: update go-libp2p to v0.28.1 ([filecoin-project/lotus#10998](https://github.com/filecoin-project/lotus/pull/10998))

build/bootstrap/calibnet.pi

@@ -1,7 +1,3 @@
-/dns4/bootstrap-0.calibration.fildev.network/tcp/1347/p2p/12D3KooWCi2w8U4DDB9xqrejb5KYHaQv2iA2AJJ6uzG3iQxNLBMy
-/dns4/bootstrap-1.calibration.fildev.network/tcp/1347/p2p/12D3KooWDTayrBojBn9jWNNUih4nNQQBGJD7Zo3gQCKgBkUsS6dp
-/dns4/bootstrap-2.calibration.fildev.network/tcp/1347/p2p/12D3KooWNRxTHUn8bf7jz1KEUPMc2dMgGfa4f8ZJTsquVSn3vHCG
-/dns4/bootstrap-3.calibration.fildev.network/tcp/1347/p2p/12D3KooWFWUqE9jgXvcKHWieYs9nhyp6NF4ftwLGAHm4sCv73jjK
 /dns4/calibration.node.glif.io/tcp/1237/p2p/12D3KooWQPYouEAsUQKzvFUA9sQ8tz4rfpqtTzh2eL6USd9bwg7x
 /dns4/bootstrap-calibnet-0.chainsafe-fil.io/tcp/34000/p2p/12D3KooWABQ5gTDHPWyvhJM7jPhtNwNJruzTEo32Lo4gcS5ABAMm
 /dns4/bootstrap-calibnet-1.chainsafe-fil.io/tcp/34000/p2p/12D3KooWS3ZRhMYL67b4bD5XQ6fcpTyVQXnDe8H89LvwrDqaSbiT

build/version.go

@@ -37,7 +37,7 @@ func BuildTypeString() string {
 }
 
 // BuildVersion is the local build version
-const BuildVersion = "1.26.0-rc2"
+const BuildVersion = "1.26.0-rc3"
 
 func UserVersion() string {
 	if os.Getenv("LOTUS_VERSION_IGNORE_COMMIT") == "1" {

chain/events/filter/index.go

@@ -46,6 +46,7 @@ var ddls = []string{
 	)`,
 
 	`CREATE INDEX IF NOT EXISTS height_tipset_key ON event (height,tipset_key)`,
+	`CREATE INDEX IF NOT EXISTS event_emitter_addr ON event (emitter_addr)`,
 
 	`CREATE TABLE IF NOT EXISTS event_entry (
 		event_id INTEGER,
@@ -56,6 +57,8 @@ var ddls = []string{
 		value BLOB NOT NULL
 	)`,
 
+	`CREATE INDEX IF NOT EXISTS event_entry_key_index ON event_entry (key)`,
+
 	// metadata containing version of schema
 	`CREATE TABLE IF NOT EXISTS _meta (
     	version UINT64 NOT NULL UNIQUE
@@ -63,6 +66,7 @@ var ddls = []string{
 
 	`INSERT OR IGNORE INTO _meta (version) VALUES (1)`,
 	`INSERT OR IGNORE INTO _meta (version) VALUES (2)`,
+	`INSERT OR IGNORE INTO _meta (version) VALUES (3)`,
 }
 
 var (
@@ -70,7 +74,7 @@ var (
 )
 
 const (
-	schemaVersion = 2
+	schemaVersion = 3
 
 	eventExists          = `SELECT MAX(id) FROM event WHERE height=? AND tipset_key=? AND tipset_key_cid=? AND emitter_addr=? AND event_index=? AND message_cid=? AND message_index=?`
 	insertEvent          = `INSERT OR IGNORE INTO event(height, tipset_key, tipset_key_cid, emitter_addr, event_index, message_cid, message_index, reverted) VALUES(?, ?, ?, ?, ?, ?, ?, ?)`
@@ -321,6 +325,22 @@ func NewEventIndex(ctx context.Context, path string, chainStore *store.ChainStor
 			version = 2
 		}
 
+		if version == 2 {
+			log.Infof("upgrading event index from version 2 to version 3")
+
+			// to upgrade to version 3 we only need to create an index on the event_entry.key column
+			// and on the event.emitter_addr column
+			// which means we can just reapply the schema (it will not have any effect on existing data)
+			for _, ddl := range ddls {
+				if _, err := db.Exec(ddl); err != nil {
+					_ = db.Close()
+					return nil, xerrors.Errorf("could not upgrade index to version 3, exec ddl %q: %w", ddl, err)
+				}
+			}
+
+			version = 3
+		}
+
 		if version != schemaVersion {
 			_ = db.Close()
 			return nil, xerrors.Errorf("invalid database version: got %d, expected %d", version, schemaVersion)

documentation/en/cli-lotus-miner.md

@@ -7,7 +7,7 @@ USAGE:
    lotus-miner [global options] command [command options] [arguments...]
 
 VERSION:
-   1.26.0-rc2
+   1.26.0-rc3
 
 COMMANDS:
    init     Initialize a lotus miner repo

documentation/en/cli-lotus-provider.md

@@ -7,7 +7,7 @@ USAGE:
    lotus-provider [global options] command [command options] [arguments...]
 
 VERSION:
-   1.26.0-rc2
+   1.26.0-rc3
 
 COMMANDS:
    run      Start a lotus provider process

documentation/en/cli-lotus-worker.md

@@ -7,7 +7,7 @@ USAGE:
    lotus-worker [global options] command [command options] [arguments...]
 
 VERSION:
-   1.26.0-rc2
+   1.26.0-rc3
 
 COMMANDS:
    run        Start lotus worker

documentation/en/cli-lotus.md

@@ -7,7 +7,7 @@ USAGE:
    lotus [global options] command [command options] [arguments...]
 
 VERSION:
-   1.26.0-rc2
+   1.26.0-rc3
 
 COMMANDS:
    daemon   Start a lotus daemon process

itests/kit/ensemble_opts.go

@@ -35,7 +35,13 @@ var DefaultEnsembleOpts = ensembleOpts{
 }
 
 // MockProofs activates mock proofs for the entire ensemble.
-func MockProofs() EnsembleOpt {
+func MockProofs(e ...bool) EnsembleOpt {
+	if len(e) > 0 && !e[0] {
+		return func(opts *ensembleOpts) error {
+			return nil
+		}
+	}
+
 	return func(opts *ensembleOpts) error {
 		opts.mockProofs = true
 		// since we're using mock proofs, we don't need to download

itests/sector_pledge_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/big"
 	miner5 "github.com/filecoin-project/specs-actors/v5/actors/builtin/miner"
 
 	"github.com/filecoin-project/lotus/api"
@@ -39,7 +40,7 @@ func TestPledgeSectors(t *testing.T) {
 		defer cancel()
 
 		_, miner, ens := kit.EnsembleMinimal(t, kit.MockProofs())
-		ens.InterconnectAll().BeginMining(blockTime)
+		ens.InterconnectAll().BeginMiningMustPost(blockTime)
 
 		miner.PledgeSectors(ctx, nSectors, 0, nil)
 	}
@@ -65,12 +66,18 @@ func TestPledgeBatching(t *testing.T) {
 	//stm: @SECTOR_PRE_COMMIT_FLUSH_001, @SECTOR_COMMIT_FLUSH_001
 	blockTime := 50 * time.Millisecond
 
-	runTest := func(t *testing.T, nSectors int) {
+	runTest := func(t *testing.T, nSectors int, aggregate bool) {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
-		client, miner, ens := kit.EnsembleMinimal(t, kit.MockProofs())
-		ens.InterconnectAll().BeginMining(blockTime)
+		kit.QuietMiningLogs()
+
+		client, miner, ens := kit.EnsembleMinimal(t, kit.MockProofs(!aggregate), kit.MutateSealingConfig(func(sc *config.SealingConfig) {
+			if aggregate {
+				sc.AggregateAboveBaseFee = types.FIL(big.Zero())
+			}
+		}))
+		ens.InterconnectAll().BeginMiningMustPost(blockTime)
 
 		client.WaitTillChain(ctx, kit.HeightAtLeast(10))
 
@@ -114,7 +121,10 @@ func TestPledgeBatching(t *testing.T) {
 	}
 
 	t.Run("100", func(t *testing.T) {
-		runTest(t, 100)
+		runTest(t, 100, false)
+	})
+	t.Run("10-agg", func(t *testing.T) {
+		runTest(t, 10, true)
 	})
 }
 

node/builder_chain.go

@@ -225,7 +225,7 @@ func ConfigFullNode(c interface{}) Option {
 		// If the Eth JSON-RPC is enabled, enable storing events at the ChainStore.
 		// This is the case even if real-time and historic filtering are disabled,
 		// as it enables us to serve logs in eth_getTransactionReceipt.
-		If(cfg.Fevm.EnableEthRPC, Override(StoreEventsKey, modules.EnableStoringEvents)),
+		If(cfg.Fevm.EnableEthRPC || cfg.Events.EnableActorEventsAPI, Override(StoreEventsKey, modules.EnableStoringEvents)),
 
 		Override(new(dtypes.ClientImportMgr), modules.ClientImportMgr),
 

node/impl/full/eth.go

@@ -42,6 +42,8 @@ import (
 
 var ErrUnsupported = errors.New("unsupported method")
 
+const maxEthFeeHistoryRewardPercentiles = 100
+
 type EthModuleAPI interface {
 	EthBlockNumber(ctx context.Context) (ethtypes.EthUint64, error)
 	EthAccounts(ctx context.Context) ([]ethtypes.EthAddress, error)
@@ -698,6 +700,9 @@ func (a *EthModule) EthFeeHistory(ctx context.Context, p jsonrpc.RawParams) (eth
 	}
 	rewardPercentiles := make([]float64, 0)
 	if params.RewardPercentiles != nil {
+		if len(*params.RewardPercentiles) > maxEthFeeHistoryRewardPercentiles {
+			return ethtypes.EthFeeHistory{}, errors.New("length of the reward percentile array cannot be greater than 100")
+		}
 		rewardPercentiles = append(rewardPercentiles, *params.RewardPercentiles...)
 	}
 	for i, rp := range rewardPercentiles {

node/modules/actorevent.go

@@ -164,9 +164,13 @@ func EventFilterManager(cfg config.EventsConfig) func(helpers.MetricsCtx, repo.L
 
 func ActorEventHandler(cfg config.EventsConfig) func(helpers.MetricsCtx, repo.LockedRepo, fx.Lifecycle, *filter.EventFilterManager, *store.ChainStore, *stmgr.StateManager, EventHelperAPI, *messagepool.MessagePool, full.StateAPI, full.ChainAPI) (*full.ActorEventHandler, error) {
 	return func(mctx helpers.MetricsCtx, r repo.LockedRepo, lc fx.Lifecycle, fm *filter.EventFilterManager, cs *store.ChainStore, sm *stmgr.StateManager, evapi EventHelperAPI, mp *messagepool.MessagePool, stateapi full.StateAPI, chainapi full.ChainAPI) (*full.ActorEventHandler, error) {
-
 		if !cfg.EnableActorEventsAPI || cfg.DisableRealTimeFilterAPI {
-			fm = nil
+			return full.NewActorEventHandler(
+				cs,
+				nil, // no EventFilterManager disables API calls
+				time.Duration(build.BlockDelaySecs)*time.Second,
+				abi.ChainEpoch(cfg.MaxFilterHeightRange),
+			), nil
 		}
 
 		return full.NewActorEventHandler(

storage/pipeline/commit_batch.go

@@ -209,7 +209,7 @@ func (b *CommitBatcher) maybeStartBatch(notif bool) ([]sealiface.CommitBatchRes,
 		return nil, xerrors.Errorf("getting config: %w", err)
 	}
 
-	if notif && total < cfg.MaxCommitBatch {
+	if notif && total < cfg.MaxCommitBatch && cfg.AggregateCommits {
 		return nil, nil
 	}
 
@@ -233,7 +233,7 @@ func (b *CommitBatcher) maybeStartBatch(notif bool) ([]sealiface.CommitBatchRes,
 		return false
 	}
 
-	individual := (total < cfg.MinCommitBatch) || (total < miner.MinAggregatedSectors) || blackedOut()
+	individual := (total < cfg.MinCommitBatch) || (total < miner.MinAggregatedSectors) || blackedOut() || !cfg.AggregateCommits
 
 	if !individual && !cfg.AggregateAboveBaseFee.Equals(big.Zero()) {
 		if ts.MinTicketBlock().ParentBaseFee.LessThan(cfg.AggregateAboveBaseFee) {
@@ -331,6 +331,9 @@ func (b *CommitBatcher) processBatchV2(cfg sealiface.Config, sectors []abi.Secto
 		return nil, err
 	}
 
+	// sort sectors by number
+	sort.Slice(sectors, func(i, j int) bool { return sectors[i] < sectors[j] })
+
 	total := len(sectors)
 
 	res := sealiface.CommitBatchRes{
@@ -371,10 +374,6 @@ func (b *CommitBatcher) processBatchV2(cfg sealiface.Config, sectors []abi.Secto
 		return nil, nil
 	}
 
-	sort.Slice(infos, func(i, j int) bool {
-		return infos[i].Number < infos[j].Number
-	})
-
 	proofs := make([][]byte, 0, total)
 	for _, info := range infos {
 		proofs = append(proofs, b.todo[info.Number].Proof)
@@ -444,13 +443,13 @@ func (b *CommitBatcher) processBatchV2(cfg sealiface.Config, sectors []abi.Secto
 	enc := new(bytes.Buffer)
 	if err := params.MarshalCBOR(enc); err != nil {
 		res.Error = err.Error()
-		return []sealiface.CommitBatchRes{res}, xerrors.Errorf("couldn't serialize ProveCommitSectors2Params: %w", err)
+		return []sealiface.CommitBatchRes{res}, xerrors.Errorf("couldn't serialize ProveCommitSectors3Params: %w", err)
 	}
 
 	_, err = simulateMsgGas(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.ProveCommitSectors3, needFunds, maxFee, enc.Bytes())
 
 	if err != nil && (!api.ErrorIsIn(err, []error{&api.ErrOutOfGas{}}) || len(sectors) < miner.MinAggregatedSectors*2) {
-		log.Errorf("simulating CommitBatch message failed: %s", err)
+		log.Errorf("simulating CommitBatch message failed (%x): %s", enc.Bytes(), err)
 		res.Error = err.Error()
 		return []sealiface.CommitBatchRes{res}, xerrors.Errorf("simulating CommitBatch message failed: %w", err)
 	}
@@ -474,7 +473,7 @@ func (b *CommitBatcher) processBatchV2(cfg sealiface.Config, sectors []abi.Secto
 
 	res.Msg = &mcid
 
-	log.Infow("Sent ProveCommitSectors2 message", "cid", mcid, "from", from, "todo", total, "sectors", len(infos))
+	log.Infow("Sent ProveCommitSectors3 message", "cid", mcid, "from", from, "todo", total, "sectors", len(infos))
 
 	return []sealiface.CommitBatchRes{res}, nil
 }
@@ -591,7 +590,7 @@ func (b *CommitBatcher) processBatchV1(cfg sealiface.Config, sectors []abi.Secto
 	_, err = simulateMsgGas(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.ProveCommitAggregate, needFunds, maxFee, enc.Bytes())
 
 	if err != nil && (!api.ErrorIsIn(err, []error{&api.ErrOutOfGas{}}) || len(sectors) < miner.MinAggregatedSectors*2) {
-		log.Errorf("simulating CommitBatch message failed: %s", err)
+		log.Errorf("simulating CommitBatch message failed (%x): %s", enc.Bytes(), err)
 		res.Error = err.Error()
 		return []sealiface.CommitBatchRes{res}, xerrors.Errorf("simulating CommitBatch message failed: %w", err)
 	}

storage/pipeline/fsm.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"os"
 	"reflect"
+	"runtime"
 	"time"
 
 	"golang.org/x/xerrors"
@@ -39,8 +40,27 @@ func (m *Sealing) Plan(events []statemachine.Event, user interface{}) (interface
 		return nil, processed, nil
 	}
 
-	return func(ctx statemachine.Context, si SectorInfo) error {
-		err := next(ctx, si)
+	return func(ctx statemachine.Context, si SectorInfo) (err error) {
+		// handle panics
+		defer func() {
+			if r := recover(); r != nil {
+				buf := make([]byte, 1<<16)
+				n := runtime.Stack(buf, false)
+				buf = buf[:n]
+
+				l := Log{
+					Timestamp: uint64(time.Now().Unix()),
+					Message:   fmt.Sprintf("panic: %v\n%s", r, buf),
+					Kind:      "panic",
+				}
+				si.logAppend(l)
+
+				err = fmt.Errorf("panic: %v\n%s", r, buf)
+			}
+		}()
+
+		// execute the next state
+		err = next(ctx, si)
 		if err != nil {
 			log.Errorf("unhandled sector error (%d): %+v", si.SectorNumber, err)
 			return nil
@@ -127,8 +147,8 @@ var fsmPlanners = map[SectorState]func(events []statemachine.Event, state *Secto
 	),
 	Committing: planCommitting,
 	CommitFinalize: planOne(
-		on(SectorFinalized{}, SubmitCommit),
-		on(SectorFinalizedAvailable{}, SubmitCommit),
+		on(SectorFinalized{}, SubmitCommitAggregate),
+		on(SectorFinalizedAvailable{}, SubmitCommitAggregate),
 		on(SectorFinalizeFailed{}, CommitFinalizeFailed),
 	),
 	SubmitCommit: planOne(
@@ -674,7 +694,7 @@ func planCommitting(events []statemachine.Event, state *SectorInfo) (uint64, err
 			}
 		case SectorCommitted: // the normal case
 			e.apply(state)
-			state.State = SubmitCommit
+			state.State = SubmitCommitAggregate
 		case SectorProofReady: // early finalize
 			e.apply(state)
 			state.State = CommitFinalize

storage/pipeline/fsm_test.go

@@ -70,10 +70,10 @@ func TestHappyPath(t *testing.T) {
 	require.Equal(m.t, m.state.State, Committing)
 
 	m.planSingle(SectorCommitted{})
-	require.Equal(m.t, m.state.State, SubmitCommit)
+	require.Equal(m.t, m.state.State, SubmitCommitAggregate)
 
-	m.planSingle(SectorCommitSubmitted{})
-	require.Equal(m.t, m.state.State, CommitWait)
+	m.planSingle(SectorCommitAggregateSent{})
+	require.Equal(m.t, m.state.State, CommitAggregateWait)
 
 	m.planSingle(SectorProving{})
 	require.Equal(m.t, m.state.State, FinalizeSector)
@@ -81,7 +81,7 @@ func TestHappyPath(t *testing.T) {
 	m.planSingle(SectorFinalized{})
 	require.Equal(m.t, m.state.State, Proving)
 
-	expected := []SectorState{Packing, GetTicket, PreCommit1, PreCommit2, SubmitPreCommitBatch, PreCommitBatchWait, WaitSeed, Committing, SubmitCommit, CommitWait, FinalizeSector, Proving}
+	expected := []SectorState{Packing, GetTicket, PreCommit1, PreCommit2, SubmitPreCommitBatch, PreCommitBatchWait, WaitSeed, Committing, SubmitCommitAggregate, CommitAggregateWait, FinalizeSector, Proving}
 	for i, n := range notif {
 		if n.before.State != expected[i] {
 			t.Fatalf("expected before state: %s, got: %s", expected[i], n.before.State)
@@ -135,9 +135,6 @@ func TestHappyPathFinalizeEarly(t *testing.T) {
 	require.Equal(m.t, m.state.State, CommitFinalize)
 
 	m.planSingle(SectorFinalized{})
-	require.Equal(m.t, m.state.State, SubmitCommit)
-
-	m.planSingle(SectorSubmitCommitAggregate{})
 	require.Equal(m.t, m.state.State, SubmitCommitAggregate)
 
 	m.planSingle(SectorCommitAggregateSent{})
@@ -149,7 +146,7 @@ func TestHappyPathFinalizeEarly(t *testing.T) {
 	m.planSingle(SectorFinalized{})
 	require.Equal(m.t, m.state.State, Proving)
 
-	expected := []SectorState{Packing, GetTicket, PreCommit1, PreCommit2, SubmitPreCommitBatch, PreCommitBatchWait, WaitSeed, Committing, CommitFinalize, SubmitCommit, SubmitCommitAggregate, CommitAggregateWait, FinalizeSector, Proving}
+	expected := []SectorState{Packing, GetTicket, PreCommit1, PreCommit2, SubmitPreCommitBatch, PreCommitBatchWait, WaitSeed, Committing, CommitFinalize, SubmitCommitAggregate, CommitAggregateWait, FinalizeSector, Proving}
 	for i, n := range notif {
 		if n.before.State != expected[i] {
 			t.Fatalf("expected before state: %s, got: %s", expected[i], n.before.State)
@@ -188,9 +185,9 @@ func TestCommitFinalizeFailed(t *testing.T) {
 	require.Equal(m.t, m.state.State, CommitFinalize)
 
 	m.planSingle(SectorFinalized{})
-	require.Equal(m.t, m.state.State, SubmitCommit)
+	require.Equal(m.t, m.state.State, SubmitCommitAggregate)
 
-	expected := []SectorState{Committing, CommitFinalize, CommitFinalizeFailed, CommitFinalize, SubmitCommit}
+	expected := []SectorState{Committing, CommitFinalize, CommitFinalizeFailed, CommitFinalize, SubmitCommitAggregate}
 	for i, n := range notif {
 		if n.before.State != expected[i] {
 			t.Fatalf("expected before state: %s, got: %s", expected[i], n.before.State)
@@ -242,10 +239,10 @@ func TestSeedRevert(t *testing.T) {
 	// not changing the seed this time
 	_, _, err = m.s.plan([]statemachine.Event{{User: SectorSeedReady{SeedValue: nil, SeedEpoch: 5}}, {User: SectorCommitted{}}}, m.state)
 	require.NoError(t, err)
-	require.Equal(m.t, m.state.State, SubmitCommit)
+	require.Equal(m.t, m.state.State, SubmitCommitAggregate)
 
-	m.planSingle(SectorCommitSubmitted{})
-	require.Equal(m.t, m.state.State, CommitWait)
+	m.planSingle(SectorCommitAggregateSent{})
+	require.Equal(m.t, m.state.State, CommitAggregateWait)
 
 	m.planSingle(SectorProving{})
 	require.Equal(m.t, m.state.State, FinalizeSector)

storage/pipeline/input.go

@@ -34,12 +34,16 @@ func (m *Sealing) handleWaitDeals(ctx statemachine.Context, sector SectorInfo) e
 	for _, piece := range sector.Pieces {
 		used += piece.Piece().Size.Unpadded()
 
+		if !piece.HasDealInfo() {
+			continue
+		}
+
 		endEpoch, err := piece.EndEpoch()
 		if err != nil {
 			return xerrors.Errorf("piece.EndEpoch: %w", err)
 		}
 
-		if piece.HasDealInfo() && endEpoch > lastDealEnd {
+		if endEpoch > lastDealEnd {
 			lastDealEnd = endEpoch
 		}
 	}
@@ -953,20 +957,30 @@ func (m *Sealing) SectorsStatus(ctx context.Context, sid abi.SectorNumber, showO
 		return api.SectorInfo{}, err
 	}
 
+	nv, err := m.Api.StateNetworkVersion(ctx, types.EmptyTSK)
+	if err != nil {
+		return api.SectorInfo{}, xerrors.Errorf("getting network version: %w", err)
+	}
+
 	deals := make([]abi.DealID, len(info.Pieces))
 	pieces := make([]api.SectorPiece, len(info.Pieces))
 	for i, piece := range info.Pieces {
-		// todo make this work with DDO deals in some reasonable way
-
 		pieces[i].Piece = piece.Piece()
-		if !piece.HasDealInfo() || piece.Impl().PublishCid == nil {
+
+		if !piece.HasDealInfo() {
+			continue
+		}
+
+		pdi := piece.Impl()
+		if pdi.Valid(nv) != nil {
 			continue
 		}
 
-		pdi := piece.DealInfo().Impl() // copy
 		pieces[i].DealInfo = &pdi
 
-		deals[i] = piece.DealInfo().Impl().DealID
+		if pdi.PublishCid != nil {
+			deals[i] = pdi.DealID
+		}
 	}
 
 	log := make([]api.SectorLog, len(info.Log))

storage/pipeline/sector_state.go

@@ -94,7 +94,7 @@ const (
 	CommitFinalizeFailed SectorState = "CommitFinalizeFailed"
 
 	// single commit
-	SubmitCommit SectorState = "SubmitCommit" // send commit message to the chain
+	SubmitCommit SectorState = "SubmitCommit" // send commit message to the chain (deprecated)
 	CommitWait   SectorState = "CommitWait"   // wait for the commit message to land on chain
 
 	SubmitCommitAggregate SectorState = "SubmitCommitAggregate"

storage/pipeline/states_sealing.go

@@ -18,7 +18,6 @@ import (
 	"github.com/filecoin-project/go-state-types/abi"
 	actorstypes "github.com/filecoin-project/go-state-types/actors"
 	"github.com/filecoin-project/go-state-types/big"
-	"github.com/filecoin-project/go-state-types/builtin"
 	miner2 "github.com/filecoin-project/go-state-types/builtin/v13/miner"
 	verifreg13 "github.com/filecoin-project/go-state-types/builtin/v13/verifreg"
 	"github.com/filecoin-project/go-state-types/builtin/v9/verifreg"
@@ -740,90 +739,11 @@ func (m *Sealing) handleCommitting(ctx statemachine.Context, sector SectorInfo)
 }
 
 func (m *Sealing) handleSubmitCommit(ctx statemachine.Context, sector SectorInfo) error {
-	// TODO: Deprecate this path, always go through batcher, just respect the AggregateCommits config in there
-
-	cfg, err := m.getConfig()
-	if err != nil {
-		return xerrors.Errorf("getting config: %w", err)
-	}
-
-	if cfg.AggregateCommits {
-		nv, err := m.Api.StateNetworkVersion(ctx.Context(), types.EmptyTSK)
-		if err != nil {
-			return xerrors.Errorf("getting network version: %w", err)
-		}
-
-		if nv >= network.Version13 {
+	// like precommit this is a deprecated state, but we keep it around for
+	// existing state machines
+	// todo: drop after nv21
 	return ctx.Send(SectorSubmitCommitAggregate{})
 }
-	}
-
-	ts, err := m.Api.ChainHead(ctx.Context())
-	if err != nil {
-		log.Errorf("handleSubmitCommit: api error, not proceeding: %+v", err)
-		return nil
-	}
-
-	if err := m.checkCommit(ctx.Context(), sector, sector.Proof, ts.Key()); err != nil {
-		return ctx.Send(SectorCommitFailed{xerrors.Errorf("commit check error: %w", err)})
-	}
-
-	enc := new(bytes.Buffer)
-	params := &miner.ProveCommitSectorParams{
-		SectorNumber: sector.SectorNumber,
-		Proof:        sector.Proof,
-	}
-
-	if err := params.MarshalCBOR(enc); err != nil {
-		return ctx.Send(SectorCommitFailed{xerrors.Errorf("could not serialize commit sector parameters: %w", err)})
-	}
-
-	mi, err := m.Api.StateMinerInfo(ctx.Context(), m.maddr, ts.Key())
-	if err != nil {
-		log.Errorf("handleCommitting: api error, not proceeding: %+v", err)
-		return nil
-	}
-
-	pci, err := m.Api.StateSectorPreCommitInfo(ctx.Context(), m.maddr, sector.SectorNumber, ts.Key())
-	if err != nil {
-		return xerrors.Errorf("getting precommit info: %w", err)
-	}
-	if pci == nil {
-		return ctx.Send(SectorCommitFailed{error: xerrors.Errorf("precommit info not found on chain")})
-	}
-
-	collateral, err := m.Api.StateMinerInitialPledgeCollateral(ctx.Context(), m.maddr, pci.Info, ts.Key())
-	if err != nil {
-		return xerrors.Errorf("getting initial pledge collateral: %w", err)
-	}
-
-	collateral = big.Sub(collateral, pci.PreCommitDeposit)
-	if collateral.LessThan(big.Zero()) {
-		collateral = big.Zero()
-	}
-
-	collateral, err = collateralSendAmount(ctx.Context(), m.Api, m.maddr, cfg, collateral)
-	if err != nil {
-		return err
-	}
-
-	goodFunds := big.Add(collateral, big.Int(m.feeCfg.MaxCommitGasFee))
-
-	from, _, err := m.addrSel.AddressFor(ctx.Context(), m.Api, mi, api.CommitAddr, goodFunds, collateral)
-	if err != nil {
-		return ctx.Send(SectorCommitFailed{xerrors.Errorf("no good address to send commit message from: %w", err)})
-	}
-
-	// TODO: check seed / ticket / deals are up to date
-	mcid, err := sendMsg(ctx.Context(), m.Api, from, m.maddr, builtin.MethodsMiner.ProveCommitSector, collateral, big.Int(m.feeCfg.MaxCommitGasFee), enc.Bytes())
-	if err != nil {
-		return ctx.Send(SectorCommitFailed{xerrors.Errorf("pushing message to mpool: %w", err)})
-	}
-
-	return ctx.Send(SectorCommitSubmitted{
-		Message: mcid,
-	})
-}
 
 // processPieces returns either:
 // - a list of piece activation manifests

storage/pipeline/types.go

@@ -289,10 +289,18 @@ func (sp *SafeSectorPiece) handleDealInfo(params handleDealInfoParams) error {
 // SectorPiece Proxy
 
 func (sp *SafeSectorPiece) Impl() piece.PieceDealInfo {
+	if !sp.HasDealInfo() {
+		return piece.PieceDealInfo{}
+	}
+
 	return sp.real.DealInfo.Impl()
 }
 
 func (sp *SafeSectorPiece) String() string {
+	if !sp.HasDealInfo() {
+		return "<no deal info>"
+	}
+
 	return sp.real.DealInfo.String()
 }
 
@@ -305,21 +313,41 @@ func (sp *SafeSectorPiece) Valid(nv network.Version) error {
 }
 
 func (sp *SafeSectorPiece) StartEpoch() (abi.ChainEpoch, error) {
+	if !sp.HasDealInfo() {
+		return 0, xerrors.Errorf("no deal info")
+	}
+
 	return sp.real.DealInfo.StartEpoch()
 }
 
 func (sp *SafeSectorPiece) EndEpoch() (abi.ChainEpoch, error) {
+	if !sp.HasDealInfo() {
+		return 0, xerrors.Errorf("no deal info")
+	}
+
 	return sp.real.DealInfo.EndEpoch()
 }
 
 func (sp *SafeSectorPiece) PieceCID() cid.Cid {
+	if !sp.HasDealInfo() {
+		return sp.real.Piece.PieceCID
+	}
+
 	return sp.real.DealInfo.PieceCID()
 }
 
 func (sp *SafeSectorPiece) KeepUnsealedRequested() bool {
+	if !sp.HasDealInfo() {
+		return false
+	}
+
 	return sp.real.DealInfo.KeepUnsealedRequested()
 }
 
 func (sp *SafeSectorPiece) GetAllocation(ctx context.Context, aapi piece.AllocationAPI, tsk types.TipSetKey) (*verifreg.Allocation, error) {
+	if !sp.HasDealInfo() {
+		return nil, xerrors.Errorf("no deal info")
+	}
+
 	return sp.real.DealInfo.GetAllocation(ctx, aapi, tsk)
 }