Merge commit '316b6fcf82f624742949258556100a509f5e046b' into v1.28.0-rc5

* fix get logs slowness and handling of null blocks

CHANGELOG.md

@@ -12,12 +12,15 @@
 
 ## Improvements
 
-# v1.28.0-rc2 / 2024-07-04
+# v1.28.0-rc5 / 2024-07-11
 
-This is the second release candidate of the upcoming MANDATORY Lotus v1.28.0 release, which will deliver the Filecoin network version 23, codenamed Waffle 🧇.
+This is the fifth release candidate of the upcoming MANDATORY Lotus v1.28.0 release, which will deliver the Filecoin network version 23, codenamed Waffle 🧇.
 
 **This release candidate sets the calibration network to upgrade at epoch 1779094, corresponding to 2024-07-11T12:00:00Z.** This release does NOT set the mainnet upgrade epoch yet, in which will be updated in the final release.
 
+Compared to `Lotus v1.28.0-rc4`, the `Lotus v1.28.0-rc5` release addresses some performance problems in the `eth_getLogs` API.
+It also addresses a bug in the `eth_getLogs` API around handling null blocks.
+
 ☢️ Upgrade Warnings ☢️
 
 If you are running the `v1.26.0` or an earlier version of Lotus, please go through the `Upgrade Warnings` section for the `v1.27.*` releases, before upgrading to this RC.
@@ -90,7 +93,9 @@ For certain node operators, such as full archival nodes or systems that need to
 - Ignore market balance after nv23 (https://github.com/filecoin-project/lotus/pull/11976)
 - Add finality-related params for `eth_getBlockByNumber` (https://github.com/filecoin-project/lotus/pull/12110)
 - rename `Actor.Address` to `Actor.DelegatedAddress` and only use it for f4 addresses (https://github.com/filecoin-project/lotus/pull/12155)
-
+- feat:ec: integrate F3 dynamic manifest #12185
+- fix: f3: Fix F3 build parameters for testground target (#12189) ([filecoin-project/lotus#12189](https://github.com/filecoin-project/lotus/pull/12189))
+- fix: eth_getLogs: https://github.com/filecoin-project/lotus/pull/12212
 
 # v1.27.1 / 2024-06-24
 

build/openrpc/full.json

@@ -2,7 +2,7 @@
     "openrpc": "1.2.6",
     "info": {
         "title": "Lotus RPC API",
-        "version": "1.28.0-rc2"
+        "version": "1.28.0-rc5"
     },
     "methods": [
         {

build/openrpc/gateway.json

@@ -2,7 +2,7 @@
     "openrpc": "1.2.6",
     "info": {
         "title": "Lotus RPC API",
-        "version": "1.28.0-rc2"
+        "version": "1.28.0-rc5"
     },
     "methods": [
         {

build/openrpc/miner.json

@@ -2,7 +2,7 @@
     "openrpc": "1.2.6",
     "info": {
         "title": "Lotus RPC API",
-        "version": "1.28.0-rc2"
+        "version": "1.28.0-rc5"
     },
     "methods": [
         {

build/openrpc/worker.json

@@ -2,7 +2,7 @@
     "openrpc": "1.2.6",
     "info": {
         "title": "Lotus RPC API",
-        "version": "1.28.0-rc2"
+        "version": "1.28.0-rc5"
     },
     "methods": [
         {

build/params_2k.go

@@ -190,4 +190,5 @@ const Eip155ChainId = 31415926
 var WhitelistedBlock = cid.Undef
 
 const f3Enabled = true
+const ManifestServerID = "12D3KooWHcNBkqXEBrsjoveQvj6zDF3vK5S9tAfqyYaQF1LGSJwG"
 const F3BootstrapEpoch abi.ChainEpoch = 100

build/params_butterfly.go

@@ -107,4 +107,5 @@ const Eip155ChainId = 3141592
 var WhitelistedBlock = cid.Undef
 
 const f3Enabled = true
+const ManifestServerID = "12D3KooWJr9jy4ngtJNR7JC1xgLFra3DjEtyxskRYWvBK9TC3Yn6"
 const F3BootstrapEpoch abi.ChainEpoch = 200

build/params_calibnet.go

@@ -153,4 +153,5 @@ const Eip155ChainId = 314159
 var WhitelistedBlock = cid.Undef
 
 const f3Enabled = true
+const ManifestServerID = "12D3KooWS9vD9uwm8u2uPyJV32QBAhKAmPYwmziAgr3Xzk2FU1Mr"
 const F3BootstrapEpoch abi.ChainEpoch = UpgradeWaffleHeight + 100

build/params_interop.go

@@ -146,4 +146,5 @@ const Eip155ChainId = 3141592
 var WhitelistedBlock = cid.Undef
 
 const f3Enabled = true
+const ManifestServerID = "12D3KooWQJ2rdVnG4okDUB6yHQhAjNutGNemcM7XzqC9Eo4z9Jce"
 const F3BootstrapEpoch abi.ChainEpoch = 1000

build/params_mainnet.go

@@ -170,4 +170,5 @@ const Eip155ChainId = 314
 var WhitelistedBlock = MustParseCid("bafy2bzaceapyg2uyzk7vueh3xccxkuwbz3nxewjyguoxvhx77malc2lzn2ybi")
 
 const f3Enabled = false
+const ManifestServerID = "12D3KooWENMwUF9YxvQxar7uBWJtZkA6amvK4xWmKXfSiHUo2Qq7"
 const F3BootstrapEpoch abi.ChainEpoch = -1

build/params_testground.go

@@ -136,7 +136,8 @@ var (
 	WhitelistedBlock                 = cid.Undef
 	BootstrappersFile                = ""
 	GenesisFile                      = ""
-	F3Enabled                        = false
+	f3Enabled                        = false
+	ManifestServerID                 = ""
 	F3BootstrapEpoch  abi.ChainEpoch = -1
 )
 

build/version.go

@@ -39,7 +39,7 @@ func BuildTypeString() string {
 }
 
 // NodeBuildVersion is the local build version of the Lotus daemon
-const NodeBuildVersion string = "1.28.0-rc2"
+const NodeBuildVersion string = "1.28.0-rc5"
 
 func NodeUserVersion() BuildVersion {
 	if os.Getenv("LOTUS_VERSION_IGNORE_COMMIT") == "1" {
@@ -50,7 +50,7 @@ func NodeUserVersion() BuildVersion {
 }
 
 // MinerBuildVersion is the local build version of the Lotus miner
-const MinerBuildVersion = "1.28.0-rc2"
+const MinerBuildVersion = "1.28.0-rc5"
 
 func MinerUserVersion() BuildVersion {
 	if os.Getenv("LOTUS_VERSION_IGNORE_COMMIT") == "1" {

chain/events/filter/index.go

@@ -698,11 +698,12 @@ func (ei *EventIndex) GetMaxHeightInIndex(ctx context.Context) (uint64, error) {
 	return maxHeight, err
 }
 
-func (ei *EventIndex) IsHeightProcessed(ctx context.Context, height uint64) (bool, error) {
-	row := ei.stmtIsHeightProcessed.QueryRowContext(ctx, height)
-	var exists bool
-	err := row.Scan(&exists)
-	return exists, err
+func (ei *EventIndex) IsHeightPast(ctx context.Context, height uint64) (bool, error) {
+	maxHeight, err := ei.GetMaxHeightInIndex(ctx)
+	if err != nil {
+		return false, err
+	}
+	return height <= maxHeight, nil
 }
 
 func (ei *EventIndex) IsTipsetProcessed(ctx context.Context, tipsetKeyCid []byte) (bool, error) {

chain/events/filter/index_test.go

@@ -94,17 +94,21 @@ func TestEventIndexPrefillFilter(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, uint64(14000), mh)
 
-	b, err := ei.IsHeightProcessed(context.Background(), 14000)
+	b, err := ei.IsHeightPast(context.Background(), 14000)
 	require.NoError(t, err)
 	require.True(t, b)
 
-	b, err = ei.IsHeightProcessed(context.Background(), 14001)
+	b, err = ei.IsHeightPast(context.Background(), 14001)
 	require.NoError(t, err)
 	require.False(t, b)
 
-	b, err = ei.IsHeightProcessed(context.Background(), 13000)
+	b, err = ei.IsHeightPast(context.Background(), 13000)
 	require.NoError(t, err)
-	require.False(t, b)
+	require.True(t, b)
+
+	b, err = ei.IsHeightPast(context.Background(), 13001)
+	require.NoError(t, err)
+	require.True(t, b)
 
 	tsKey := events14000.msgTs.Key()
 	tsKeyCid, err := tsKey.Cid()

chain/lf3/ec.go

@@ -8,7 +8,7 @@ import (
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-address"
-	"github.com/filecoin-project/go-f3"
+	"github.com/filecoin-project/go-f3/ec"
 	"github.com/filecoin-project/go-f3/gpbft"
 	"github.com/filecoin-project/go-state-types/abi"
 
@@ -23,7 +23,6 @@ import (
 type ecWrapper struct {
 	ChainStore   *store.ChainStore
 	StateManager *stmgr.StateManager
-	Manifest     f3.Manifest
 }
 
 type f3TipSet types.TipSet
@@ -57,7 +56,7 @@ func (ts *f3TipSet) Timestamp() time.Time {
 	return time.Unix(int64(ts.cast().Blocks()[0].Timestamp), 0)
 }
 
-func wrapTS(ts *types.TipSet) f3.TipSet {
+func wrapTS(ts *types.TipSet) ec.TipSet {
 	if ts == nil {
 		return nil
 	}
@@ -66,7 +65,7 @@ func wrapTS(ts *types.TipSet) f3.TipSet {
 
 // GetTipsetByEpoch should return a tipset before the one requested if the requested
 // tipset does not exist due to null epochs
-func (ec *ecWrapper) GetTipsetByEpoch(ctx context.Context, epoch int64) (f3.TipSet, error) {
+func (ec *ecWrapper) GetTipsetByEpoch(ctx context.Context, epoch int64) (ec.TipSet, error) {
 	ts, err := ec.ChainStore.GetTipsetByHeight(ctx, abi.ChainEpoch(epoch), nil, true)
 	if err != nil {
 		return nil, xerrors.Errorf("getting tipset by height: %w", err)
@@ -74,7 +73,7 @@ func (ec *ecWrapper) GetTipsetByEpoch(ctx context.Context, epoch int64) (f3.TipS
 	return wrapTS(ts), nil
 }
 
-func (ec *ecWrapper) GetTipset(ctx context.Context, tsk gpbft.TipSetKey) (f3.TipSet, error) {
+func (ec *ecWrapper) GetTipset(ctx context.Context, tsk gpbft.TipSetKey) (ec.TipSet, error) {
 	tskLotus, err := types.TipSetKeyFromBytes(tsk)
 	if err != nil {
 		return nil, xerrors.Errorf("decoding tsk: %w", err)
@@ -88,16 +87,16 @@ func (ec *ecWrapper) GetTipset(ctx context.Context, tsk gpbft.TipSetKey) (f3.Tip
 	return wrapTS(ts), nil
 }
 
-func (ec *ecWrapper) GetHead(_ context.Context) (f3.TipSet, error) {
+func (ec *ecWrapper) GetHead(_ context.Context) (ec.TipSet, error) {
 	return wrapTS(ec.ChainStore.GetHeaviestTipSet()), nil
 }
 
-func (ec *ecWrapper) GetParent(ctx context.Context, tsF3 f3.TipSet) (f3.TipSet, error) {
+func (ec *ecWrapper) GetParent(ctx context.Context, tsF3 ec.TipSet) (ec.TipSet, error) {
 	var ts *types.TipSet
 	if tsW, ok := tsF3.(*f3TipSet); ok {
 		ts = tsW.cast()
 	} else {
-		// There are only two implementations of F3.TipSet: f3TipSet, and one in fake EC
+		// There are only two implementations of ec.TipSet: f3TipSet, and one in fake EC
 		// backend.
 		//
 		// TODO: Revisit the type check here and remove as needed once testing

chain/lf3/f3.go

@@ -3,13 +3,13 @@ package lf3
 import (
 	"context"
 	"errors"
-	"time"
 
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
 	logging "github.com/ipfs/go-log/v2"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/peer"
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
@@ -17,6 +17,7 @@ import (
 	"github.com/filecoin-project/go-f3/blssig"
 	"github.com/filecoin-project/go-f3/certs"
 	"github.com/filecoin-project/go-f3/gpbft"
+	"github.com/filecoin-project/go-f3/manifest"
 
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
@@ -35,39 +36,39 @@ type F3 struct {
 type F3Params struct {
 	fx.In
 
-	NetworkName  dtypes.NetworkName
-	PubSub       *pubsub.PubSub
-	Host         host.Host
-	ChainStore   *store.ChainStore
-	StateManager *stmgr.StateManager
-	Datastore    dtypes.MetadataDS
-	Wallet       api.Wallet
+	NetworkName      dtypes.NetworkName
+	ManifestProvider manifest.ManifestProvider
+	PubSub           *pubsub.PubSub
+	Host             host.Host
+	ChainStore       *store.ChainStore
+	StateManager     *stmgr.StateManager
+	Datastore        dtypes.MetadataDS
+	Wallet           api.Wallet
 }
 
 var log = logging.Logger("f3")
 
 func New(mctx helpers.MetricsCtx, lc fx.Lifecycle, params F3Params) (*F3, error) {
-	manifest := f3.LocalnetManifest()
-	manifest.NetworkName = gpbft.NetworkName(params.NetworkName)
-	manifest.ECDelay = 2 * time.Duration(build.BlockDelaySecs) * time.Second
-	manifest.ECPeriod = manifest.ECDelay
-	manifest.BootstrapEpoch = int64(build.F3BootstrapEpoch)
-	manifest.ECFinality = int64(build.Finality)
 
 	ds := namespace.Wrap(params.Datastore, datastore.NewKey("/f3"))
 	ec := &ecWrapper{
 		ChainStore:   params.ChainStore,
 		StateManager: params.StateManager,
-		Manifest:     manifest,
 	}
 	verif := blssig.VerifierWithKeyOnG1()
 
-	module, err := f3.New(mctx, manifest, ds,
-		params.Host, params.PubSub, verif, ec, log, nil)
+	senderID, err := peer.Decode(build.ManifestServerID)
+	if err != nil {
+		return nil, xerrors.Errorf("decoding F3 manifest server identity: %w", err)
+	}
+
+	module, err := f3.New(mctx, params.ManifestProvider, ds,
+		params.Host, senderID, params.PubSub, verif, ec, log, nil)
 
 	if err != nil {
 		return nil, xerrors.Errorf("creating F3: %w", err)
 	}
+	params.ManifestProvider.SetManifestChangeCallback(f3.ManifestChangeCallback(module))
 
 	fff := &F3{
 		inner:  module,

chain/lf3/manifest.go

@@ -0,0 +1,39 @@
+package lf3
+
+import (
+	"time"
+
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/libp2p/go-libp2p/core/peer"
+
+	"github.com/filecoin-project/go-f3/gpbft"
+	"github.com/filecoin-project/go-f3/manifest"
+
+	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/stmgr"
+	"github.com/filecoin-project/lotus/chain/store"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
+)
+
+func NewManifestProvider(nn dtypes.NetworkName, cs *store.ChainStore, sm *stmgr.StateManager, ps *pubsub.PubSub) manifest.ManifestProvider {
+	m := manifest.LocalDevnetManifest()
+	m.NetworkName = gpbft.NetworkName(nn)
+	m.ECDelay = 2 * time.Duration(build.BlockDelaySecs) * time.Second
+	m.ECPeriod = m.ECDelay
+	m.BootstrapEpoch = int64(build.F3BootstrapEpoch)
+	m.ECFinality = int64(build.Finality)
+	m.CommiteeLookback = 5
+
+	ec := &ecWrapper{
+		ChainStore:   cs,
+		StateManager: sm,
+	}
+
+	switch manifestServerID, err := peer.Decode(build.ManifestServerID); {
+	case err != nil:
+		log.Warnw("Cannot decode F3 manifest sever identity; falling back on static manifest provider", "err", err)
+		return manifest.NewStaticManifestProvider(m)
+	default:
+		return manifest.NewDynamicManifestProvider(m, ps, ec, manifestServerID)
+	}
+}

documentation/en/cli-lotus-miner.md

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

documentation/en/cli-lotus-worker.md

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

documentation/en/cli-lotus.md

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

go.mod

@@ -41,7 +41,7 @@ require (
 	github.com/filecoin-project/go-commp-utils v0.1.3
 	github.com/filecoin-project/go-commp-utils/nonffi v0.0.0-20220905160352-62059082a837
 	github.com/filecoin-project/go-crypto v0.0.1
-	github.com/filecoin-project/go-f3 v0.0.2
+	github.com/filecoin-project/go-f3 v0.0.3-0.20240702063402-d48771055cf4
 	github.com/filecoin-project/go-fil-commcid v0.1.0
 	github.com/filecoin-project/go-hamt-ipld/v3 v3.1.0
 	github.com/filecoin-project/go-jsonrpc v0.3.2

go.sum

@@ -270,8 +270,8 @@ github.com/filecoin-project/go-commp-utils/nonffi v0.0.0-20220905160352-62059082
 github.com/filecoin-project/go-crypto v0.0.0-20191218222705-effae4ea9f03/go.mod h1:+viYnvGtUTgJRdy6oaeF4MTFKAfatX071MPDPBL11EQ=
 github.com/filecoin-project/go-crypto v0.0.1 h1:AcvpSGGCgjaY8y1az6AMfKQWreF/pWO2JJGLl6gCq6o=
 github.com/filecoin-project/go-crypto v0.0.1/go.mod h1:+viYnvGtUTgJRdy6oaeF4MTFKAfatX071MPDPBL11EQ=
-github.com/filecoin-project/go-f3 v0.0.2 h1:bzw/GndxntJnUYA+WCaXwHE2qwGRwrFVo9umz3unTUs=
-github.com/filecoin-project/go-f3 v0.0.2/go.mod h1:Wry0mNa8z767TBHb7N0cVb+9j00KsHbD2pzsC3li4R8=
+github.com/filecoin-project/go-f3 v0.0.3-0.20240702063402-d48771055cf4 h1:eQW2fyKyMuiweuySEb/zMIc3WLSAnIOY8lpqCVQM7pU=
+github.com/filecoin-project/go-f3 v0.0.3-0.20240702063402-d48771055cf4/go.mod h1:Wry0mNa8z767TBHb7N0cVb+9j00KsHbD2pzsC3li4R8=
 github.com/filecoin-project/go-fil-commcid v0.0.0-20201016201715-d41df56b4f6a/go.mod h1:Eaox7Hvus1JgPrL5+M3+h7aSPHc0cVqpSxA+TxIEpZQ=
 github.com/filecoin-project/go-fil-commcid v0.1.0 h1:3R4ds1A9r6cr8mvZBfMYxTS88OqLYEo6roi+GiIeOh8=
 github.com/filecoin-project/go-fil-commcid v0.1.0/go.mod h1:Eaox7Hvus1JgPrL5+M3+h7aSPHc0cVqpSxA+TxIEpZQ=

node/builder_chain.go

@@ -6,6 +6,8 @@ import (
 	"go.uber.org/fx"
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/go-f3/manifest"
+
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain"
@@ -151,7 +153,10 @@ var ChainNode = Options(
 		Override(HandleIncomingBlocksKey, modules.HandleIncomingBlocks),
 	),
 
-	If(build.IsF3Enabled(), Override(new(*lf3.F3), lf3.New)),
+	If(build.IsF3Enabled(),
+		Override(new(manifest.ManifestProvider), lf3.NewManifestProvider),
+		Override(new(*lf3.F3), lf3.New),
+	),
 )
 
 func ConfigFullNode(c interface{}) Option {

node/impl/full/eth.go

@@ -255,7 +255,6 @@ func (a *EthModule) EthGetBlockByNumber(ctx context.Context, blkParam string, fu
 
 func (a *EthModule) EthGetTransactionByHash(ctx context.Context, txHash *ethtypes.EthHash) (*ethtypes.EthTx, error) {
 	return a.EthGetTransactionByHashLimited(ctx, txHash, api.LookbackNoLimit)
-
 }
 
 func (a *EthModule) EthGetTransactionByHashLimited(ctx context.Context, txHash *ethtypes.EthHash, limit abi.ChainEpoch) (*ethtypes.EthTx, error) {
@@ -1276,9 +1275,17 @@ func (e *EthEventHandler) EthGetLogs(ctx context.Context, filterSpec *ethtypes.E
 	if pf.tipsetCid == cid.Undef {
 		maxHeight := pf.maxHeight
 		if maxHeight == -1 {
-			maxHeight = e.Chain.GetHeaviestTipSet().Height()
+			// heaviest tipset doesn't have events because its messages haven't been executed yet
+			maxHeight = e.Chain.GetHeaviestTipSet().Height() - 1
 		}
-		if maxHeight > e.Chain.GetHeaviestTipSet().Height() {
+
+		if maxHeight < 0 {
+			return nil, xerrors.Errorf("maxHeight requested is less than 0")
+		}
+
+		// we can't return events for the heaviest tipset as the transactions in that tipset will be executed
+		// in the next non null tipset (because of Filecoin's "deferred execution" model)
+		if maxHeight > e.Chain.GetHeaviestTipSet().Height()-1 {
 			return nil, xerrors.Errorf("maxHeight requested is greater than the heaviest tipset")
 		}
 
@@ -1286,13 +1293,13 @@ func (e *EthEventHandler) EthGetLogs(ctx context.Context, filterSpec *ethtypes.E
 		if err != nil {
 			return nil, err
 		}
-
-		// should also have the minHeight in the filter indexed
-		if b, err := e.EventFilterManager.EventIndex.IsHeightProcessed(ctx, uint64(pf.minHeight)); err != nil {
-			return nil, xerrors.Errorf("failed to check if event index has events for the minHeight: %w", err)
-		} else if !b {
-			return nil, xerrors.Errorf("event index does not have event for epoch %d", pf.minHeight)
-		}
+		// TODO: Ideally we should also check that events for the epoch at `pf.minheight` have been indexed
+		// However, it is currently tricky to check/guarantee this for two reasons:
+		// a) Event Index is not aware of null-blocks. This means that the Event Index wont be able to say whether the block at
+		//    `pf.minheight` is a null block or whether it has no events
+		// b) There can be holes in the index where events at certain epoch simply haven't been indexed because of edge cases around
+		//    node restarts while indexing. This needs a long term "auto-repair"/"automated-backfilling" implementation in the index
+		// So, for now, the best we can do is ensure that the event index has evenets for events at height >= `pf.maxHeight`
 	} else {
 		ts, err := e.Chain.GetTipSetByCid(ctx, pf.tipsetCid)
 		if err != nil {
@@ -1324,6 +1331,8 @@ func (e *EthEventHandler) EthGetLogs(ctx context.Context, filterSpec *ethtypes.E
 	return ethFilterResultFromEvents(ctx, ces, e.SubManager.StateAPI)
 }
 
+// note that we can have null blocks at the given height and the event Index is not null block aware
+// so, what we do here is wait till we see the event index contain a block at a height greater than the given height
 func (e *EthEventHandler) waitForHeightProcessed(ctx context.Context, height abi.ChainEpoch) error {
 	ei := e.EventFilterManager.EventIndex
 	if height > e.Chain.GetHeaviestTipSet().Height() {
@@ -1334,7 +1343,7 @@ func (e *EthEventHandler) waitForHeightProcessed(ctx context.Context, height abi
 	defer cancel()
 
 	// if the height we're interested in has already been indexed -> there's nothing to do here
-	if b, err := ei.IsHeightProcessed(ctx, uint64(height)); err != nil {
+	if b, err := ei.IsHeightPast(ctx, uint64(height)); err != nil {
 		return xerrors.Errorf("failed to check if event index has events for given height: %w", err)
 	} else if b {
 		return nil
@@ -1345,7 +1354,7 @@ func (e *EthEventHandler) waitForHeightProcessed(ctx context.Context, height abi
 	defer unSubscribeF()
 
 	// it could be that the event index was update while the subscription was being processed -> check if index has what we need now
-	if b, err := ei.IsHeightProcessed(ctx, uint64(height)); err != nil {
+	if b, err := ei.IsHeightPast(ctx, uint64(height)); err != nil {
 		return xerrors.Errorf("failed to check if event index has events for given height: %w", err)
 	} else if b {
 		return nil
@@ -1354,7 +1363,7 @@ func (e *EthEventHandler) waitForHeightProcessed(ctx context.Context, height abi
 	for {
 		select {
 		case <-subCh:
-			if b, err := ei.IsHeightProcessed(ctx, uint64(height)); err != nil {
+			if b, err := ei.IsHeightPast(ctx, uint64(height)); err != nil {
 				return xerrors.Errorf("failed to check if event index has events for given height: %w", err)
 			} else if b {
 				return nil

node/modules/lp2p/pubsub.go

@@ -3,6 +3,7 @@ package lp2p
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net"
 	"time"
 
@@ -17,6 +18,7 @@ import (
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-f3/gpbft"
+	"github.com/filecoin-project/go-f3/manifest"
 
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/metrics"
@@ -46,6 +48,12 @@ const (
 	GraylistScoreThreshold           = -2500
 	AcceptPXScoreThreshold           = 1000
 	OpportunisticGraftScoreThreshold = 3.5
+
+	// Determines the max. number of configuration changes
+	// that are allowed for the dynamic manifest.
+	// If the manifest changes more than this number, the F3
+	// message topic will be filtered
+	MaxDynamicManifestChangesAllowed = 1000
 )
 
 func ScoreKeeper() *dtypes.ScoreKeeper {
@@ -382,7 +390,28 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
 	}
 
 	if build.IsF3Enabled() {
-		allowTopics = append(allowTopics, gpbft.NetworkName(in.Nn).PubSubTopic())
+		f3TopicName := manifest.PubSubTopicFromNetworkName(gpbft.NetworkName(in.Nn))
+		allowTopics = append(allowTopics, f3TopicName)
+
+		// allow dynamic manifest topic and the new topic names after a reconfiguration.
+		// Note: This is pretty ugly, but I tried to use a regex subscription filter
+		// as the commented code below, but unfortunately it overwrites previous filters. A simple fix would
+		// be to allow combining several topic filters, but for now this works.
+		//
+		// 	pattern := fmt.Sprintf(`^\/f3\/%s\/0\.0\.1\/?[0-9]*$`, in.Nn)
+		// 	rx, err := regexp.Compile(pattern)
+		// 	if err != nil {
+		// 		return nil, xerrors.Errorf("failed to compile manifest topic regex: %w", err)
+		// 	}
+		// 	options = append(options,
+		// 		pubsub.WithSubscriptionFilter(
+		// 			pubsub.WrapLimitSubscriptionFilter(
+		// 				pubsub.NewRegexpSubscriptionFilter(rx),
+		// 				100)))
+		allowTopics = append(allowTopics, manifest.ManifestPubSubTopicName)
+		for i := 0; i < MaxDynamicManifestChangesAllowed; i++ {
+			allowTopics = append(allowTopics, f3TopicName+"/"+fmt.Sprintf("%d", i))
+		}
 	}
 
 	allowTopics = append(allowTopics, drandTopics...)